---

name: langchain description: LangChain Python framework for building LLM applications. Use for chains, agents, memory, document loaders, text splitters, vector stores, retrievers, and LLM integrations like OpenAI, Anthropic, and Google.

Langchain Skill

Langchain python framework for building llm applications. use for chains, agents, memory, document loaders, text splitters, vector stores, retrievers, and llm integrations like openai, anthropic, and google., generated from official documentation.

When to Use This Skill

This skill should be triggered when:

• Working with langchain
• Asking about langchain features or APIs
• Implementing langchain solutions
• Debugging langchain code
• Learning langchain best practices

Quick Reference

Common Patterns

Pattern 1: Docs by LangChain home pageLangChain + LangGraphSearch...⌘KSupportGitHubTry LangSmithTry LangSmithSearch...NavigationCore componentsModelsLangChainLangGraphDeep AgentsIntegrationsLearnReferenceContributePythonOverviewGet startedInstallQuickstartChangelogPhilosophyCore componentsAgentsModelsMessagesToolsShort-term memoryStreamingStructured outputMiddlewareOverviewBuilt-in middlewareCustom middlewareAdvanced usageGuardrailsRuntimeContext engineeringModel Context Protocol (MCP)Human-in-the-loopMulti-agentRetrievalLong-term memoryAgent developmentLangSmith StudioTestAgent Chat UIDeploy with LangSmithDeploymentObservabilityOn this pageBasic usageInitialize a modelSupported modelsKey methodsParametersInvocationInvokeStreamBatchTool callingStructured outputAdvanced topicsModel profilesMultimodalReasoningLocal modelsPrompt cachingServer-side tool useRate limitingBase URL or proxyLog probabilitiesToken usageInvocation configConfigurable modelsCore componentsModelsCopy pageCopy pageLLMs are powerful AI tools that can interpret and generate text like humans. They’re versatile enough to write content, translate languages, summarize, and answer questions without needing specialized training for each task. In addition to text generation, many models support: Tool calling - calling external tools (like databases queries or API calls) and use results in their responses. Structured output - where the model’s response is constrained to follow a defined format. Multimodality - process and return data other than text, such as images, audio, and video. Reasoning - models perform multi-step reasoning to arrive at a conclusion. Models are the reasoning engine of agents. They drive the agent’s decision-making process, determining which tools to call, how to interpret results, and when to provide a final answer. The quality and capabilities of the model you choose directly impact your agent’s baseline reliability and performance. Different models excel at different tasks - some are better at following complex instructions, others at structured reasoning, and some support larger context windows for handling more information. LangChain’s standard model interfaces give you access to many different provider integrations, which makes it easy to experiment with and switch between models to find the best fit for your use case. For provider-specific integration information and capabilities, see the provider’s chat model page. ​Basic usage Models can be utilized in two ways: With agents - Models can be dynamically specified when creating an agent. Standalone - Models can be called directly (outside of the agent loop) for tasks like text generation, classification, or extraction without the need for an agent framework. The same model interface works in both contexts, which gives you the flexibility to start simple and scale up to more complex agent-based workflows as needed. ​Initialize a model The easiest way to get started with a standalone model in LangChain is to use init_chat_model to initialize one from a chat model provider of your choice (examples below): OpenAI Anthropic Azure Google Gemini AWS Bedrock HuggingFace👉 Read the OpenAI chat model integration docsCopypip install -U "langchain[openai]" init_chat_modelModel ClassCopyimport os from langchain.chat_models import init_chat_model os.environ["OPENAI_API_KEY"] = "sk-..." model = init_chat_model("gpt-4.1") 👉 Read the Anthropic chat model integration docsCopypip install -U "langchain[anthropic]" init_chat_modelModel ClassCopyimport os from langchain.chat_models import init_chat_model os.environ["ANTHROPIC_API_KEY"] = "sk-..." model = init_chat_model("claude-sonnet-4-5-20250929") 👉 Read the Azure chat model integration docsCopypip install -U "langchain[openai]" init_chat_modelModel ClassCopyimport os from langchain.chat_models import init_chat_model os.environ["AZURE_OPENAI_API_KEY"] = "..." os.environ["AZURE_OPENAI_ENDPOINT"] = "..." os.environ["OPENAI_API_VERSION"] = "2025-03-01-preview" model = init_chat_model( "azure_openai:gpt-4.1", azure_deployment=os.environ["AZURE_OPENAI_DEPLOYMENT_NAME"], ) 👉 Read the Google GenAI chat model integration docsCopypip install -U "langchain[google-genai]" init_chat_modelModel ClassCopyimport os from langchain.chat_models import init_chat_model os.environ["GOOGLE_API_KEY"] = "..." model = init_chat_model("google_genai:gemini-2.5-flash-lite") 👉 Read the AWS Bedrock chat model integration docsCopypip install -U "langchain[aws]" init_chat_modelModel ClassCopyfrom langchain.chat_models import init_chat_model # Follow the steps here to configure your credentials: # https://docs.aws.amazon.com/bedrock/latest/userguide/getting-started.html model = init_chat_model( "anthropic.claude-3-5-sonnet-20240620-v1:0", model_provider="bedrock_converse", ) 👉 Read the HuggingFace chat model integration docsCopypip install -U "langchain[huggingface]" init_chat_modelModel ClassCopyimport os from langchain.chat_models import init_chat_model os.environ["HUGGINGFACEHUB_API_TOKEN"] = "hf_..." model = init_chat_model( "microsoft/Phi-3-mini-4k-instruct", model_provider="huggingface", temperature=0.7, max_tokens=1024, ) Copyresponse = model.invoke("Why do parrots talk?") See init_chat_model for more detail, including information on how to pass model parameters. ​Supported models LangChain supports all major model providers, including OpenAI, Anthropic, Google, Azure, AWS Bedrock, and more. Each provider offers a variety of models with different capabilities. For a full list of supported models in LangChain, see the integrations page. ​Key methods InvokeThe model takes messages as input and outputs messages after generating a complete response. StreamInvoke the model, but stream the output as it is generated in real-time. BatchSend multiple requests to a model in a batch for more efficient processing. In addition to chat models, LangChain provides support for other adjacent technologies, such as embedding models and vector stores. See the integrations page for details. ​Parameters A chat model takes parameters that can be used to configure its behavior. The full set of supported parameters varies by model and provider, but standard ones include: ​modelstringrequiredThe name or identifier of the specific model you want to use with a provider. You can also specify both the model and its provider in a single argument using the ’:’ format, for example, ‘openai:o1’. ​api_keystringThe key required for authenticating with the model’s provider. This is usually issued when you sign up for access to the model. Often accessed by setting an environment variable. ​temperaturenumberControls the randomness of the model’s output. A higher number makes responses more creative; lower ones make them more deterministic. ​max_tokensnumberLimits the total number of tokens in the response, effectively controlling how long the output can be. ​timeoutnumberThe maximum time (in seconds) to wait for a response from the model before canceling the request. ​max_retriesnumberThe maximum number of attempts the system will make to resend a request if it fails due to issues like network timeouts or rate limits. Using init_chat_model, pass these parameters as inline **kwargs: Initialize using model parametersCopymodel = init_chat_model( "claude-sonnet-4-5-20250929", # Kwargs passed to the model: temperature=0.7, timeout=30, max_tokens=1000, ) Each chat model integration may have additional params used to control provider-specific functionality.For example, ChatOpenAI has use_responses_api to dictate whether to use the OpenAI Responses or Completions API.To find all the parameters supported by a given chat model, head to the chat model integrations page. ​Invocation A chat model must be invoked to generate an output. There are three primary invocation methods, each suited to different use cases. ​Invoke The most straightforward way to call a model is to use invoke() with a single message or a list of messages. Single messageCopyresponse = model.invoke("Why do parrots have colorful feathers?") print(response) A list of messages can be provided to a chat model to represent conversation history. Each message has a role that models use to indicate who sent the message in the conversation. See the messages guide for more detail on roles, types, and content. Dictionary formatCopyconversation = [ {"role": "system", "content": "You are a helpful assistant that translates English to French."}, {"role": "user", "content": "Translate: I love programming."}, {"role": "assistant", "content": "J'adore la programmation."}, {"role": "user", "content": "Translate: I love building applications."} ] response = model.invoke(conversation) print(response) # AIMessage("J'adore créer des applications.") Message objectsCopyfrom langchain.messages import HumanMessage, AIMessage, SystemMessage conversation = [ SystemMessage("You are a helpful assistant that translates English to French."), HumanMessage("Translate: I love programming."), AIMessage("J'adore la programmation."), HumanMessage("Translate: I love building applications.") ] response = model.invoke(conversation) print(response) # AIMessage("J'adore créer des applications.") If the return type of your invocation is a string, ensure that you are using a chat model as opposed to a LLM. Legacy, text-completion LLMs return strings directly. LangChain chat models are prefixed with “Chat”, e.g., ChatOpenAI(/oss/integrations/chat/openai). ​Stream Most models can stream their output content while it is being generated. By displaying output progressively, streaming significantly improves user experience, particularly for longer responses. Calling stream() returns an iterator that yields output chunks as they are produced. You can use a loop to process each chunk in real-time: Basic text streamingStream tool calls, reasoning, and other contentCopyfor chunk in model.stream("Why do parrots have colorful feathers?"): print(chunk.text, end="|", flush=True) As opposed to invoke(), which returns a single AIMessage after the model has finished generating its full response, stream() returns multiple AIMessageChunk objects, each containing a portion of the output text. Importantly, each chunk in a stream is designed to be gathered into a full message via summation: Construct an AIMessageCopyfull = None # None | AIMessageChunk for chunk in model.stream("What color is the sky?"): full = chunk if full is None else full + chunk print(full.text) # The # The sky # The sky is # The sky is typically # The sky is typically blue # ... print(full.content_blocks) # [{"type": "text", "text": "The sky is typically blue..."}] The resulting message can be treated the same as a message that was generated with invoke() – for example, it can be aggregated into a message history and passed back to the model as conversational context. Streaming only works if all steps in the program know how to process a stream of chunks. For instance, an application that isn’t streaming-capable would be one that needs to store the entire output in memory before it can be processed. Advanced streaming topicsStreaming eventsLangChain chat models can also stream semantic events using astream_events().This simplifies filtering based on event types and other metadata, and will aggregate the full message in the background. See below for an example.Copyasync for event in model.astream_events("Hello"): if event["event"] == "on_chat_model_start": print(f"Input: {event['data']['input']}") elif event["event"] == "on_chat_model_stream": print(f"Token: {event['data']['chunk'].text}") elif event["event"] == "on_chat_model_end": print(f"Full message: {event['data']['output'].text}") else: pass CopyInput: Hello Token: Hi Token: there Token: ! Token: How Token: can Token: I ... Full message: Hi there! How can I help today? See the astream_events() reference for event types and other details."Auto-streaming" chat modelsLangChain simplifies streaming from chat models by automatically enabling streaming mode in certain cases, even when you’re not explicitly calling the streaming methods. This is particularly useful when you use the non-streaming invoke method but still want to stream the entire application, including intermediate results from the chat model.In LangGraph agents, for example, you can call model.invoke() within nodes, but LangChain will automatically delegate to streaming if running in a streaming mode.​How it worksWhen you invoke() a chat model, LangChain will automatically switch to an internal streaming mode if it detects that you are trying to stream the overall application. The result of the invocation will be the same as far as the code that was using invoke is concerned; however, while the chat model is being streamed, LangChain will take care of invoking on_llm_new_token events in LangChain’s callback system.Callback events allow LangGraph stream() and astream_events() to surface the chat model’s output in real-time. ​Batch Batching a collection of independent requests to a model can significantly improve performance and reduce costs, as the processing can be done in parallel: BatchCopyresponses = model.batch([ "Why do parrots have colorful feathers?", "How do airplanes fly?", "What is quantum computing?" ]) for response in responses: print(response) This section describes a chat model method batch(), which parallelizes model calls client-side.It is distinct from batch APIs supported by inference providers, such as OpenAI or Anthropic. By default, batch() will only return the final output for the entire batch. If you want to receive the output for each individual input as it finishes generating, you can stream results with batch_as_completed(): Yield batch responses upon completionCopyfor response in model.batch_as_completed([ "Why do parrots have colorful feathers?", "How do airplanes fly?", "What is quantum computing?" ]): print(response) When using batch_as_completed(), results may arrive out of order. Each includes the input index for matching to reconstruct the original order as needed. When processing a large number of inputs using batch() or batch_as_completed(), you may want to control the maximum number of parallel calls. This can be done by setting the max_concurrency attribute in the RunnableConfig dictionary.Batch with max concurrencyCopymodel.batch( list_of_inputs, config={ 'max_concurrency': 5, # Limit to 5 parallel calls } ) See the RunnableConfig reference for a full list of supported attributes. For more details on batching, see the reference. ​Tool calling Models can request to call tools that perform tasks such as fetching data from a database, searching the web, or running code. Tools are pairings of: A schema, including the name of the tool, a description, and/or argument definitions (often a JSON schema) A function or coroutine to execute. You may hear the term “function calling”. We use this interchangeably with “tool calling”. Here’s the basic tool calling flow between a user and a model: To make tools that you have defined available for use by a model, you must bind them using bind_tools. In subsequent invocations, the model can choose to call any of the bound tools as needed. Some model providers offer built-in tools that can be enabled via model or invocation parameters (e.g. ChatOpenAI, ChatAnthropic). Check the respective provider reference for details. See the tools guide for details and other options for creating tools. Binding user toolsCopyfrom langchain.tools import tool @tool def get_weather(location: str) -> str: """Get the weather at a location.""" return f"It's sunny in {location}." model_with_tools = model.bind_tools([get_weather]) response = model_with_tools.invoke("What's the weather like in Boston?") for tool_call in response.tool_calls: # View tool calls made by the model print(f"Tool: {tool_call['name']}") print(f"Args: {tool_call['args']}") When binding user-defined tools, the model’s response includes a request to execute a tool. When using a model separately from an agent, it is up to you to execute the requested tool and return the result back to the model for use in subsequent reasoning. When using an agent, the agent loop will handle the tool execution loop for you. Below, we show some common ways you can use tool calling. Tool execution loopWhen a model returns tool calls, you need to execute the tools and pass the results back to the model. This creates a conversation loop where the model can use tool results to generate its final response. LangChain includes agent abstractions that handle this orchestration for you.Here’s a simple example of how to do this:Tool execution loopCopy# Bind (potentially multiple) tools to the model model_with_tools = model.bind_tools([get_weather]) # Step 1: Model generates tool calls messages = [{"role": "user", "content": "What's the weather in Boston?"}] ai_msg = model_with_tools.invoke(messages) messages.append(ai_msg) # Step 2: Execute tools and collect results for tool_call in ai_msg.tool_calls: # Execute the tool with the generated arguments tool_result = get_weather.invoke(tool_call) messages.append(tool_result) # Step 3: Pass results back to model for final response final_response = model_with_tools.invoke(messages) print(final_response.text) # "The current weather in Boston is 72°F and sunny." Each ToolMessage returned by the tool includes a tool_call_id that matches the original tool call, helping the model correlate results with requests.Forcing tool callsBy default, the model has the freedom to choose which bound tool to use based on the user’s input. However, you might want to force choosing a tool, ensuring the model uses either a particular tool or any tool from a given list:Force use of any toolForce use of specific toolsCopymodel_with_tools = model.bind_tools([tool_1], tool_choice="any") Parallel tool callsMany models support calling multiple tools in parallel when appropriate. This allows the model to gather information from different sources simultaneously.Parallel tool callsCopymodel_with_tools = model.bind_tools([get_weather]) response = model_with_tools.invoke( "What's the weather in Boston and Tokyo?" ) # The model may generate multiple tool calls print(response.tool_calls) # [ # {'name': 'get_weather', 'args': {'location': 'Boston'}, 'id': 'call_1'}, # {'name': 'get_weather', 'args': {'location': 'Tokyo'}, 'id': 'call_2'}, # ] # Execute all tools (can be done in parallel with async) results = [] for tool_call in response.tool_calls: if tool_call['name'] == 'get_weather': result = get_weather.invoke(tool_call) ... results.append(result) The model intelligently determines when parallel execution is appropriate based on the independence of the requested operations.Most models supporting tool calling enable parallel tool calls by default. Some (including OpenAI and Anthropic) allow you to disable this feature. To do this, set parallel_tool_calls=False:Copymodel.bind_tools([get_weather], parallel_tool_calls=False) Streaming tool callsWhen streaming responses, tool calls are progressively built through ToolCallChunk. This allows you to see tool calls as they’re being generated rather than waiting for the complete response.Streaming tool callsCopyfor chunk in model_with_tools.stream( "What's the weather in Boston and Tokyo?" ): # Tool call chunks arrive progressively for tool_chunk in chunk.tool_call_chunks: if name := tool_chunk.get("name"): print(f"Tool: {name}") if id_ := tool_chunk.get("id"): print(f"ID: {id_}") if args := tool_chunk.get("args"): print(f"Args: {args}") # Output: # Tool: get_weather # ID: call_SvMlU1TVIZugrFLckFE2ceRE # Args: {"lo # Args: catio # Args: n": "B # Args: osto # Args: n"} # Tool: get_weather # ID: call_QMZdy6qInx13oWKE7KhuhOLR # Args: {"lo # Args: catio # Args: n": "T # Args: okyo # Args: "} You can accumulate chunks to build complete tool calls:Accumulate tool callsCopygathered = None for chunk in model_with_tools.stream("What's the weather in Boston?"): gathered = chunk if gathered is None else gathered + chunk print(gathered.tool_calls) ​Structured output Models can be requested to provide their response in a format matching a given schema. This is useful for ensuring the output can be easily parsed and used in subsequent processing. LangChain supports multiple schema types and methods for enforcing structured output. To learn about structured output, see Structured output. Pydantic TypedDict JSON SchemaPydantic models provide the richest feature set with field validation, descriptions, and nested structures.Copyfrom pydantic import BaseModel, Field class Movie(BaseModel): """A movie with details.""" title: str = Field(..., description="The title of the movie") year: int = Field(..., description="The year the movie was released") director: str = Field(..., description="The director of the movie") rating: float = Field(..., description="The movie's rating out of 10") model_with_structure = model.with_structured_output(Movie) response = model_with_structure.invoke("Provide details about the movie Inception") print(response) # Movie(title="Inception", year=2010, director="Christopher Nolan", rating=8.8) Python’s TypedDict provides a simpler alternative to Pydantic models, ideal when you don’t need runtime validation.Copyfrom typing_extensions import TypedDict, Annotated class MovieDict(TypedDict): """A movie with details.""" title: Annotated[str, ..., "The title of the movie"] year: Annotated[int, ..., "The year the movie was released"] director: Annotated[str, ..., "The director of the movie"] rating: Annotated[float, ..., "The movie's rating out of 10"] model_with_structure = model.with_structured_output(MovieDict) response = model_with_structure.invoke("Provide details about the movie Inception") print(response) # {'title': 'Inception', 'year': 2010, 'director': 'Christopher Nolan', 'rating': 8.8} Provide a JSON Schema for maximum control and interoperability.Copyimport json json_schema = { "title": "Movie", "description": "A movie with details", "type": "object", "properties": { "title": { "type": "string", "description": "The title of the movie" }, "year": { "type": "integer", "description": "The year the movie was released" }, "director": { "type": "string", "description": "The director of the movie" }, "rating": { "type": "number", "description": "The movie's rating out of 10" } }, "required": ["title", "year", "director", "rating"] } model_with_structure = model.with_structured_output( json_schema, method="json_schema", ) response = model_with_structure.invoke("Provide details about the movie Inception") print(response) # {'title': 'Inception', 'year': 2010, ...} Key considerations for structured output Method parameter: Some providers support different methods for structured output: 'json_schema': Uses dedicated structured output features offered by the provider. 'function_calling': Derives structured output by forcing a tool call that follows the given schema. 'json_mode': A precursor to 'json_schema' offered by some providers. Generates valid JSON, but the schema must be described in the prompt. Include raw: Set include_raw=True to get both the parsed output and the raw AI message. Validation: Pydantic models provide automatic validation. TypedDict and JSON Schema require manual validation. See your provider’s integration page for supported methods and configuration options. Example: Message output alongside parsed structureIt can be useful to return the raw AIMessage object alongside the parsed representation to access response metadata such as token counts. To do this, set include_raw=True when calling with_structured_output:Copyfrom pydantic import BaseModel, Field class Movie(BaseModel): """A movie with details.""" title: str = Field(..., description="The title of the movie") year: int = Field(..., description="The year the movie was released") director: str = Field(..., description="The director of the movie") rating: float = Field(..., description="The movie's rating out of 10") model_with_structure = model.with_structured_output(Movie, include_raw=True) response = model_with_structure.invoke("Provide details about the movie Inception") response # { # "raw": AIMessage(...), # "parsed": Movie(title=..., year=..., ...), # "parsing_error": None, # } Example: Nested structuresSchemas can be nested:Pydantic BaseModelTypedDictCopyfrom pydantic import BaseModel, Field class Actor(BaseModel): name: str role: str class MovieDetails(BaseModel): title: str year: int cast: list[Actor] genres: list[str] budget: float | None = Field(None, description="Budget in millions USD") model_with_structure = model.with_structured_output(MovieDetails) ​Advanced topics ​Model profiles Model profiles require langchain>=1.1. LangChain chat models can expose a dictionary of supported features and capabilities through a .profile attribute: Copymodel.profile # { # "max_input_tokens": 400000, # "image_inputs": True, # "reasoning_output": True, # "tool_calling": True, # ... # } Refer to the full set of fields in the API reference. Much of the model profile data is powered by the models.dev project, an open source initiative that provides model capability data. These data are augmented with additional fields for purposes of use with LangChain. These augmentations are kept aligned with the upstream project as it evolves. Model profile data allow applications to work around model capabilities dynamically. For example: Summarization middleware can trigger summarization based on a model’s context window size. Structured output strategies in create_agent can be inferred automatically (e.g., by checking support for native structured output features). Model inputs can be gated based on supported modalities and maximum input tokens. Updating or overwriting profile dataModel profile data can be changed if it is missing, stale, or incorrect.Option 1 (quick fix)You can instantiate a chat model with any valid profile:Copycustom_profile = { "max_input_tokens": 100_000, "tool_calling": True, "structured_output": True, # ... } model = init_chat_model("...", profile=custom_profile) The profile is also a regular dict and can be updated in place. If the model instance is shared, consider using model_copy to avoid mutating shared state.Copynew_profile = model.profile | {"key": "value"} model.model_copy(update={"profile": new_profile}) Option 2 (fix data upstream)The primary source for the data is the models.dev project. This data is merged with additional fields and overrides in LangChain integration packages and are shipped with those packages.Model profile data can be updated through the following process: (If needed) update the source data at models.dev through a pull request to its repository on GitHub. (If needed) update additional fields and overrides in langchain_/data/profile_augmentations.toml through a pull request to the LangChain integration package`. Use the langchain-model-profiles CLI tool to pull the latest data from models.dev, merge in the augmentations and update the profile data: Copypip install langchain-model-profiles Copylangchain-profiles refresh --provider --data-dir <data_dir> This command: Downloads the latest data for from models.dev Merges augmentations from profile_augmentations.toml in <data_dir> Writes merged profiles to profiles.py in <data_dir> For example: from libs/partners/anthropic in the LangChain monorepo:Copyuv run --with langchain-model-profiles --provider anthropic --data-dir langchain_anthropic/data Model profiles are a beta feature. The format of a profile is subject to change. ​Multimodal Certain models can process and return non-textual data such as images, audio, and video. You can pass non-textual data to a model by providing content blocks. All LangChain chat models with underlying multimodal capabilities support: Data in the cross-provider standard format (see our messages guide) OpenAI chat completions format Any format that is native to that specific provider (e.g., Anthropic models accept Anthropic native format) See the multimodal section of the messages guide for details. Some models can return multimodal data as part of their response. If invoked to do so, the resulting AIMessage will have content blocks with multimodal types. Multimodal outputCopyresponse = model.invoke("Create a picture of a cat") print(response.content_blocks) # [ # {"type": "text", "text": "Here's a picture of a cat"}, # {"type": "image", "base64": "...", "mime_type": "image/jpeg"}, # ] See the integrations page for details on specific providers. ​Reasoning Many models are capable of performing multi-step reasoning to arrive at a conclusion. This involves breaking down complex problems into smaller, more manageable steps. If supported by the underlying model, you can surface this reasoning process to better understand how the model arrived at its final answer. Stream reasoning outputComplete reasoning outputCopyfor chunk in model.stream("Why do parrots have colorful feathers?"): reasoning_steps = [r for r in chunk.content_blocks if r["type"] == "reasoning"] print(reasoning_steps if reasoning_steps else chunk.text) Depending on the model, you can sometimes specify the level of effort it should put into reasoning. Similarly, you can request that the model turn off reasoning entirely. This may take the form of categorical “tiers” of reasoning (e.g., 'low' or 'high') or integer token budgets. For details, see the integrations page or reference for your respective chat model. ​Local models LangChain supports running models locally on your own hardware. This is useful for scenarios where either data privacy is critical, you want to invoke a custom model, or when you want to avoid the costs incurred when using a cloud-based model. Ollama is one of the easiest ways to run chat and embedding models locally. ​Prompt caching Many providers offer prompt caching features to reduce latency and cost on repeat processing of the same tokens. These features can be implicit or explicit: Implicit prompt caching: providers will automatically pass on cost savings if a request hits a cache. Examples: OpenAI and Gemini. Explicit caching: providers allow you to manually indicate cache points for greater control or to guarantee cost savings. Examples: ChatOpenAI (via prompt_cache_key) Anthropic’s AnthropicPromptCachingMiddleware Gemini. AWS Bedrock Prompt caching is often only engaged above a minimum input token threshold. See provider pages for details. Cache usage will be reflected in the usage metadata of the model response. ​Server-side tool use Some providers support server-side tool-calling loops: models can interact with web search, code interpreters, and other tools and analyze the results in a single conversational turn. If a model invokes a tool server-side, the content of the response message will include content representing the invocation and result of the tool. Accessing the content blocks of the response will return the server-side tool calls and results in a provider-agnostic format: Invoke with server-side tool useCopyfrom langchain.chat_models import init_chat_model model = init_chat_model("gpt-4.1-mini") tool = {"type": "web_search"} model_with_tools = model.bind_tools([tool]) response = model_with_tools.invoke("What was a positive news story from today?") response.content_blocks ResultCopy[ { "type": "server_tool_call", "name": "web_search", "args": { "query": "positive news stories today", "type": "search" }, "id": "ws_abc123" }, { "type": "server_tool_result", "tool_call_id": "ws_abc123", "status": "success" }, { "type": "text", "text": "Here are some positive news stories from today...", "annotations": [ { "end_index": 410, "start_index": 337, "title": "article title", "type": "citation", "url": "..." } ] } ] See all 29 lines This represents a single conversational turn; there are no associated ToolMessage objects that need to be passed in as in client-side tool-calling. See the integration page for your given provider for available tools and usage details. ​Rate limiting Many chat model providers impose a limit on the number of invocations that can be made in a given time period. If you hit a rate limit, you will typically receive a rate limit error response from the provider, and will need to wait before making more requests. To help manage rate limits, chat model integrations accept a rate_limiter parameter that can be provided during initialization to control the rate at which requests are made. Initialize and use a rate limiterLangChain in comes with (an optional) built-in InMemoryRateLimiter. This limiter is thread safe and can be shared by multiple threads in the same process.Define a rate limiterCopyfrom langchain_core.rate_limiters import InMemoryRateLimiter rate_limiter = InMemoryRateLimiter( requests_per_second=0.1, # 1 request every 10s check_every_n_seconds=0.1, # Check every 100ms whether allowed to make a request max_bucket_size=10, # Controls the maximum burst size. ) model = init_chat_model( model="gpt-5", model_provider="openai", rate_limiter=rate_limiter ) The provided rate limiter can only limit the number of requests per unit time. It will not help if you need to also limit based on the size of the requests. ​Base URL or proxy For many chat model integrations, you can configure the base URL for API requests, which allows you to use model providers that have OpenAI-compatible APIs or to use a proxy server. Base URLMany model providers offer OpenAI-compatible APIs (e.g., Together AI, vLLM). You can use init_chat_model with these providers by specifying the appropriate base_url parameter:Copymodel = init_chat_model( model="MODEL_NAME", model_provider="openai", base_url="BASE_URL", api_key="YOUR_API_KEY", ) When using direct chat model class instantiation, the parameter name may vary by provider. Check the respective reference for details. Proxy configurationFor deployments requiring HTTP proxies, some model integrations support proxy configuration:Copyfrom langchain_openai import ChatOpenAI model = ChatOpenAI( model="gpt-4o", openai_proxy="http://proxy.example.com:8080" ) Proxy support varies by integration. Check the specific model provider’s reference for proxy configuration options. ​Log probabilities Certain models can be configured to return token-level log probabilities representing the likelihood of a given token by setting the logprobs parameter when initializing the model: Copymodel = init_chat_model( model="gpt-4o", model_provider="openai" ).bind(logprobs=True) response = model.invoke("Why do parrots talk?") print(response.response_metadata["logprobs"]) ​Token usage A number of model providers return token usage information as part of the invocation response. When available, this information will be included on the AIMessage objects produced by the corresponding model. For more details, see the messages guide. Some provider APIs, notably OpenAI and Azure OpenAI chat completions, require users opt-in to receiving token usage data in streaming contexts. See the streaming usage metadata section of the integration guide for details. You can track aggregate token counts across models in an application using either a callback or context manager, as shown below: Callback handler Context managerCopyfrom langchain.chat_models import init_chat_model from langchain_core.callbacks import UsageMetadataCallbackHandler model_1 = init_chat_model(model="gpt-4o-mini") model_2 = init_chat_model(model="claude-haiku-4-5-20251001") callback = UsageMetadataCallbackHandler() result_1 = model_1.invoke("Hello", config={"callbacks": [callback]}) result_2 = model_2.invoke("Hello", config={"callbacks": [callback]}) callback.usage_metadata Copy{ 'gpt-4o-mini-2024-07-18': { 'input_tokens': 8, 'output_tokens': 10, 'total_tokens': 18, 'input_token_details': {'audio': 0, 'cache_read': 0}, 'output_token_details': {'audio': 0, 'reasoning': 0} }, 'claude-haiku-4-5-20251001': { 'input_tokens': 8, 'output_tokens': 21, 'total_tokens': 29, 'input_token_details': {'cache_read': 0, 'cache_creation': 0} } } Copyfrom langchain.chat_models import init_chat_model from langchain_core.callbacks import get_usage_metadata_callback model_1 = init_chat_model(model="gpt-4o-mini") model_2 = init_chat_model(model="claude-haiku-4-5-20251001") with get_usage_metadata_callback() as cb: model_1.invoke("Hello") model_2.invoke("Hello") print(cb.usage_metadata) Copy{ 'gpt-4o-mini-2024-07-18': { 'input_tokens': 8, 'output_tokens': 10, 'total_tokens': 18, 'input_token_details': {'audio': 0, 'cache_read': 0}, 'output_token_details': {'audio': 0, 'reasoning': 0} }, 'claude-haiku-4-5-20251001': { 'input_tokens': 8, 'output_tokens': 21, 'total_tokens': 29, 'input_token_details': {'cache_read': 0, 'cache_creation': 0} } } ​Invocation config When invoking a model, you can pass additional configuration through the config parameter using a RunnableConfig dictionary. This provides run-time control over execution behavior, callbacks, and metadata tracking. Common configuration options include: Invocation with configCopyresponse = model.invoke( "Tell me a joke", config={ "run_name": "joke_generation", # Custom name for this run "tags": ["humor", "demo"], # Tags for categorization "metadata": {"user_id": "123"}, # Custom metadata "callbacks": [my_callback_handler], # Callback handlers } ) These configuration values are particularly useful when: Debugging with LangSmith tracing Implementing custom logging or monitoring Controlling resource usage in production Tracking invocations across complex pipelines Key configuration attributes​run_namestringIdentifies this specific invocation in logs and traces. Not inherited by sub-calls.​tagsstring[]Labels inherited by all sub-calls for filtering and organization in debugging tools.​metadataobjectCustom key-value pairs for tracking additional context, inherited by all sub-calls.​max_concurrencynumberControls the maximum number of parallel calls when using batch() or batch_as_completed().​callbacksarrayHandlers for monitoring and responding to events during execution.​recursion_limitnumberMaximum recursion depth for chains to prevent infinite loops in complex pipelines. See full RunnableConfig reference for all supported attributes. ​Configurable models You can also create a runtime-configurable model by specifying configurable_fields. If you don’t specify a model value, then 'model' and 'model_provider' will be configurable by default. Copyfrom langchain.chat_models import init_chat_model configurable_model = init_chat_model(temperature=0) configurable_model.invoke( "what's your name", config={"configurable": {"model": "gpt-5-nano"}}, # Run with GPT-5-Nano ) configurable_model.invoke( "what's your name", config={"configurable": {"model": "claude-sonnet-4-5-20250929"}}, # Run with Claude ) Configurable model with default valuesWe can create a configurable model with default model values, specify which parameters are configurable, and add prefixes to configurable params:Copyfirst_model = init_chat_model( model="gpt-4.1-mini", temperature=0, configurable_fields=("model", "model_provider", "temperature", "max_tokens"), config_prefix="first", # Useful when you have a chain with multiple models ) first_model.invoke("what's your name") Copyfirst_model.invoke( "what's your name", config={ "configurable": { "first_model": "claude-sonnet-4-5-20250929", "first_temperature": 0.5, "first_max_tokens": 100, } }, ) See the init_chat_model reference for more details on configurable_fields and config_prefix. Using a configurable model declarativelyWe can call declarative operations like bind_tools, with_structured_output, with_configurable, etc. on a configurable model and chain a configurable model in the same way that we would a regularly instantiated chat model object.Copyfrom pydantic import BaseModel, Field class GetWeather(BaseModel): """Get the current weather in a given location""" location: str = Field(..., description="The city and state, e.g. San Francisco, CA") class GetPopulation(BaseModel): """Get the current population in a given location""" location: str = Field(..., description="The city and state, e.g. San Francisco, CA") model = init_chat_model(temperature=0) model_with_tools = model.bind_tools([GetWeather, GetPopulation]) model_with_tools.invoke( "what's bigger in 2024 LA or NYC", config={"configurable": {"model": "gpt-4.1-mini"}} ).tool_calls Copy[ { 'name': 'GetPopulation', 'args': {'location': 'Los Angeles, CA'}, 'id': 'call_Ga9m8FAArIyEjItHmztPYA22', 'type': 'tool_call' }, { 'name': 'GetPopulation', 'args': {'location': 'New York, NY'}, 'id': 'call_jh2dEvBaAHRaw5JUDthOs7rt', 'type': 'tool_call' } ] Copymodel_with_tools.invoke( "what's bigger in 2024 LA or NYC", config={"configurable": {"model": "claude-sonnet-4-5-20250929"}}, ).tool_calls Copy[ { 'name': 'GetPopulation', 'args': {'location': 'Los Angeles, CA'}, 'id': 'toolu_01JMufPf4F4t2zLj7miFeqXp', 'type': 'tool_call' }, { 'name': 'GetPopulation', 'args': {'location': 'New York City, NY'}, 'id': 'toolu_01RQBHcE8kEEbYTuuS8WqY1u', 'type': 'tool_call' } ] Edit this page on GitHub or file an issue. Connect these docs to Claude, VSCode, and more via MCP for real-time answers.Was this page helpful?YesNoAgentsPreviousMessagesNext⌘IDocs by LangChain home pagegithubxlinkedinyoutubeResourcesForumChangelogLangChain AcademyTrust CenterCompanyAboutCareersBloggithubxlinkedinyoutubePowered by

Copyinit_chat_model

Pattern 2: Docs by LangChain home pageLangChain + LangGraphSearch...⌘KSupportGitHubTry LangSmithTry LangSmithSearch...NavigationMulti-agentMulti-agentLangChainLangGraphDeep AgentsIntegrationsLearnReferenceContributePythonOverviewGet startedInstallQuickstartChangelogPhilosophyCore componentsAgentsModelsMessagesToolsShort-term memoryStreamingStructured outputMiddlewareOverviewBuilt-in middlewareCustom middlewareAdvanced usageGuardrailsRuntimeContext engineeringModel Context Protocol (MCP)Human-in-the-loopMulti-agentOverviewSubagentsHandoffsSkillsRouterCustom workflowRetrievalLong-term memoryAgent developmentLangSmith StudioTestAgent Chat UIDeploy with LangSmithDeploymentObservabilityOn this pageWhy multi-agent?PatternsChoosing a patternVisual overviewPerformance comparisonOne-shot requestRepeat requestMulti-domainSummaryAdvanced usageMulti-agentMulti-agentCopy pageCopy pageMulti-agent systems coordinate specialized components to tackle complex workflows. However, not every complex task requires this approach — a single agent with the right (sometimes dynamic) tools and prompt can often achieve similar results. ​Why multi-agent? When developers say they need “multi-agent,” they’re usually looking for one or more of these capabilities: Context management: Provide specialized knowledge without overwhelming the model’s context window. If context were infinite and latency zero, you could dump all knowledge into a single prompt — but since it’s not, you need patterns to selectively surface relevant information. Distributed development: Allow different teams to develop and maintain capabilities independently, composing them into a larger system with clear boundaries. Parallelization: Spawn specialized workers for subtasks and execute them concurrently for faster results. Multi-agent patterns are particularly valuable when a single agent has too many tools and makes poor decisions about which to use, when tasks require specialized knowledge with extensive context (long prompts and domain-specific tools), or when you need to enforce sequential constraints that unlock capabilities only after certain conditions are met. At the center of multi-agent design is context engineering—deciding what information each agent sees. The quality of your system depends on ensuring each agent has access to the right data for its task. ​Patterns Here are the main patterns for building multi-agent systems, each suited to different use cases: PatternHow it worksSubagentsA main agent coordinates subagents as tools. All routing passes through the main agent, which decides when and how to invoke each subagent.HandoffsBehavior changes dynamically based on state. Tool calls update a state variable that triggers routing or configuration changes, switching agents or adjusting the current agent’s tools and prompt.SkillsSpecialized prompts and knowledge loaded on-demand. A single agent stays in control while loading context from skills as needed.RouterA routing step classifies input and directs it to one or more specialized agents. Results are synthesized into a combined response.Custom workflowBuild bespoke execution flows with LangGraph, mixing deterministic logic and agentic behavior. Embed other patterns as nodes in your workflow. ​Choosing a pattern Use this table to match your requirements to the right pattern: PatternDistributed developmentParallelizationMulti-hopDirect user interactionSubagents⭐⭐⭐⭐⭐⭐⭐⭐⭐⭐⭐⭐⭐⭐⭐⭐Handoffs——⭐⭐⭐⭐⭐⭐⭐⭐⭐⭐Skills⭐⭐⭐⭐⭐⭐⭐⭐⭐⭐⭐⭐⭐⭐⭐⭐⭐⭐Router⭐⭐⭐⭐⭐⭐⭐⭐—⭐⭐⭐ Distributed development: Can different teams maintain components independently? Parallelization: Can multiple agents execute concurrently? Multi-hop: Does the pattern support calling multiple subagents in series? Direct user interaction: Can subagents converse directly with the user? You can mix patterns! For example, a subagents architecture can invoke tools that invoke custom workflows or router agents. Subagents can even use the skills pattern to load context on-demand. The possibilities are endless! ​Visual overview Subagents Handoffs Skills RouterA main agent coordinates subagents as tools. All routing passes through the main agent.Agents transfer control to each other via tool calls. Each agent can hand off to others or respond directly to the user.A single agent loads specialized prompts and knowledge on-demand while staying in control.A routing step classifies input and directs it to specialized agents. Results are synthesized. ​Performance comparison Different patterns have different performance characteristics. Understanding these tradeoffs helps you choose the right pattern for your latency and cost requirements. Key metrics: Model calls: Number of LLM invocations. More calls = higher latency (especially if sequential) and higher per-request API costs. Tokens processed: Total context window usage across all calls. More tokens = higher processing costs and potential context limits. ​One-shot request User: “Buy coffee” A specialized coffee agent/skill can call a buy_coffee tool. PatternModel callsBest fitSubagents4Handoffs3✅Skills3✅Router3✅ Subagents Handoffs Skills Router4 model calls:3 model calls:3 model calls:3 model calls: Key insight: Handoffs, Skills, and Router are most efficient for single tasks (3 calls each). Subagents adds one extra call because results flow back through the main agent—this overhead provides centralized control. ​Repeat request Turn 1: “Buy coffee” Turn 2: “Buy coffee again” The user repeats the same request in the same conversation. PatternTurn 2 callsTotal (both turns)Best fitSubagents48Handoffs25✅Skills25✅Router36 Subagents Handoffs Skills Router4 calls again → 8 total Subagents are stateless by design—each invocation follows the same flow The main agent maintains conversation context, but subagents start fresh each time This provides strong context isolation but repeats the full flow 2 calls → 5 total The coffee agent is still active from turn 1 (state persists) No handoff needed—agent directly calls buy_coffee tool (call 1) Agent responds to user (call 2) Saves 1 call by skipping the handoff 2 calls → 5 total The skill context is already loaded in conversation history No need to reload—agent directly calls buy_coffee tool (call 1) Agent responds to user (call 2) Saves 1 call by reusing loaded skill 3 calls again → 6 total Routers are stateless—each request requires an LLM routing call Turn 2: Router LLM call (1) → Milk agent calls buy_coffee (2) → Milk agent responds (3) Can be optimized by wrapping as a tool in a stateful agent Key insight: Stateful patterns (Handoffs, Skills) save 40-50% of calls on repeat requests. Subagents maintain consistent cost per request—this stateless design provides strong context isolation but at the cost of repeated model calls. ​Multi-domain User: “Compare Python, JavaScript, and Rust for web development” Each language agent/skill contains 2000 tokens of documentation. All patterns can make parallel tool calls. PatternModel callsTotal tokensBest fitSubagents59K✅Handoffs7+14K+Skills315KRouter5~9K✅ Subagents Handoffs Skills Router5 calls, ~9K tokensEach subagent works in isolation with only its relevant context. Total: 9K tokens.7+ calls, ~14K+ tokensHandoffs executes sequentially—can’t research all three languages in parallel. Growing conversation history adds overhead. Total: ~14K+ tokens.3 calls, ~15K tokensAfter loading, every subsequent call processes all 6K tokens of skill documentation. Subagents processes 67% fewer tokens overall due to context isolation. Total: 15K tokens.5 calls, ~9K tokensRouter uses an LLM for routing, then invokes agents in parallel. Similar to Subagents but with explicit routing step. Total: 9K tokens. Key insight: For multi-domain tasks, patterns with parallel execution (Subagents, Router) are most efficient. Skills has fewer calls but high token usage due to context accumulation. Handoffs is inefficient here—it must execute sequentially and can’t leverage parallel tool calling for consulting multiple domains simultaneously. ​Summary Here’s how patterns compare across all three scenarios: PatternOne-shotRepeat requestMulti-domainSubagents4 calls8 calls (4+4)5 calls, 9K tokensHandoffs3 calls5 calls (3+2)7+ calls, 14K+ tokensSkills3 calls5 calls (3+2)3 calls, 15K tokensRouter3 calls6 calls (3+3)5 calls, 9K tokens Choosing a pattern: Optimize forSubagentsHandoffsSkillsRouterSingle requests✅✅✅Repeat requests✅✅Parallel execution✅✅Large-context domains✅✅Simple, focused tasks✅ Edit this page on GitHub or file an issue. Connect these docs to Claude, VSCode, and more via MCP for real-time answers.Was this page helpful?YesNoHuman-in-the-loopPreviousSubagentsNext⌘IDocs by LangChain home pagegithubxlinkedinyoutubeResourcesForumChangelogLangChain AcademyTrust CenterCompanyAboutCareersBloggithubxlinkedinyoutubePowered by

Copybuy_coffee

Pattern 3: Docs by LangChain home pageLangChain + LangGraphSearch...⌘KSupportGitHubTry LangSmithTry LangSmithSearch...NavigationAdvanced usageRetrievalLangChainLangGraphDeep AgentsIntegrationsLearnReferenceContributePythonOverviewGet startedInstallQuickstartChangelogPhilosophyCore componentsAgentsModelsMessagesToolsShort-term memoryStreamingStructured outputMiddlewareOverviewBuilt-in middlewareCustom middlewareAdvanced usageGuardrailsRuntimeContext engineeringModel Context Protocol (MCP)Human-in-the-loopMulti-agentRetrievalLong-term memoryAgent developmentLangSmith StudioTestAgent Chat UIDeploy with LangSmithDeploymentObservabilityOn this pageBuilding a knowledge baseFrom retrieval to RAGRetrieval pipelineBuilding blocksRAG architectures2-step RAGAgentic RAGHybrid RAGAdvanced usageRetrievalCopy pageCopy pageLarge Language Models (LLMs) are powerful, but they have two key limitations: Finite context — they can’t ingest entire corpora at once. Static knowledge — their training data is frozen at a point in time. Retrieval addresses these problems by fetching relevant external knowledge at query time. This is the foundation of Retrieval-Augmented Generation (RAG): enhancing an LLM’s answers with context-specific information. ​Building a knowledge base A knowledge base is a repository of documents or structured data used during retrieval. If you need a custom knowledge base, you can use LangChain’s document loaders and vector stores to build one from your own data. If you already have a knowledge base (e.g., a SQL database, CRM, or internal documentation system), you do not need to rebuild it. You can: Connect it as a tool for an agent in Agentic RAG. Query it and supply the retrieved content as context to the LLM (2-Step RAG). See the following tutorial to build a searchable knowledge base and minimal RAG workflow: Tutorial: Semantic searchLearn how to create a searchable knowledge base from your own data using LangChain’s document loaders, embeddings, and vector stores. In this tutorial, you’ll build a search engine over a PDF, enabling retrieval of passages relevant to a query. You’ll also implement a minimal RAG workflow on top of this engine to see how external knowledge can be integrated into LLM reasoning.Learn more ​From retrieval to RAG Retrieval allows LLMs to access relevant context at runtime. But most real-world applications go one step further: they integrate retrieval with generation to produce grounded, context-aware answers. This is the core idea behind Retrieval-Augmented Generation (RAG). The retrieval pipeline becomes a foundation for a broader system that combines search with generation. ​Retrieval pipeline A typical retrieval workflow looks like this: Each component is modular: you can swap loaders, splitters, embeddings, or vector stores without rewriting the app’s logic. ​Building blocks Document loadersIngest data from external sources (Google Drive, Slack, Notion, etc.), returning standardized Document objects.Learn moreText splittersBreak large docs into smaller chunks that will be retrievable individually and fit within a model’s context window.Learn moreEmbedding modelsAn embedding model turns text into a vector of numbers so that texts with similar meaning land close together in that vector space.Learn moreVector storesSpecialized databases for storing and searching embeddings.Learn moreRetrieversA retriever is an interface that returns documents given an unstructured query.Learn more ​RAG architectures RAG can be implemented in multiple ways, depending on your system’s needs. We outline each type in the sections below. ArchitectureDescriptionControlFlexibilityLatencyExample Use Case2-Step RAGRetrieval always happens before generation. Simple and predictable✅ High❌ Low⚡ FastFAQs, documentation botsAgentic RAGAn LLM-powered agent decides when and how to retrieve during reasoning❌ Low✅ High⏳ VariableResearch assistants with access to multiple toolsHybridCombines characteristics of both approaches with validation steps⚖️ Medium⚖️ Medium⏳ VariableDomain-specific Q&A with quality validation Latency: Latency is generally more predictable in 2-Step RAG, as the maximum number of LLM calls is known and capped. This predictability assumes that LLM inference time is the dominant factor. However, real-world latency may also be affected by the performance of retrieval steps—such as API response times, network delays, or database queries—which can vary based on the tools and infrastructure in use. ​2-step RAG In 2-Step RAG, the retrieval step is always executed before the generation step. This architecture is straightforward and predictable, making it suitable for many applications where the retrieval of relevant documents is a clear prerequisite for generating an answer. Tutorial: Retrieval-Augmented Generation (RAG)See how to build a Q&A chatbot that can answer questions grounded in your data using Retrieval-Augmented Generation. This tutorial walks through two approaches: A RAG agent that runs searches with a flexible tool—great for general-purpose use. A 2-step RAG chain that requires just one LLM call per query—fast and efficient for simpler tasks. Learn more ​Agentic RAG Agentic Retrieval-Augmented Generation (RAG) combines the strengths of Retrieval-Augmented Generation with agent-based reasoning. Instead of retrieving documents before answering, an agent (powered by an LLM) reasons step-by-step and decides when and how to retrieve information during the interaction. The only thing an agent needs to enable RAG behavior is access to one or more tools that can fetch external knowledge — such as documentation loaders, web APIs, or database queries. Copyimport requests from langchain.tools import tool from langchain.chat_models import init_chat_model from langchain.agents import create_agent @tool def fetch_url(url: str) -> str: """Fetch text content from a URL""" response = requests.get(url, timeout=10.0) response.raise_for_status() return response.text system_prompt = """\ Use fetch_url when you need to fetch information from a web-page; quote relevant snippets. """ agent = create_agent( model="claude-sonnet-4-5-20250929", tools=[fetch_url], # A tool for retrieval system_prompt=system_prompt, ) Show Extended example: Agentic RAG for LangGraph's llms.txtThis example implements an Agentic RAG system to assist users in querying LangGraph documentation. The agent begins by loading llms.txt, which lists available documentation URLs, and can then dynamically use a fetch_documentation tool to retrieve and process the relevant content based on the user’s question.Copyimport requests from langchain.agents import create_agent from langchain.messages import HumanMessage from langchain.tools import tool from markdownify import markdownify ALLOWED_DOMAINS = ["https://langchain-ai.github.io/"] LLMS_TXT = 'https://langchain-ai.github.io/langgraph/llms.txt' @tool def fetch_documentation(url: str) -> str: """Fetch and convert documentation from a URL""" if not any(url.startswith(domain) for domain in ALLOWED_DOMAINS): return ( "Error: URL not allowed. " f"Must start with one of: {', '.join(ALLOWED_DOMAINS)}" ) response = requests.get(url, timeout=10.0) response.raise_for_status() return markdownify(response.text) # We will fetch the content of llms.txt, so this can # be done ahead of time without requiring an LLM request. llms_txt_content = requests.get(LLMS_TXT).text # System prompt for the agent system_prompt = f""" You are an expert Python developer and technical assistant. Your primary role is to help users with questions about LangGraph and related tools. Instructions: 1. If a user asks a question you're unsure about — or one that likely involves API usage, behavior, or configuration — you MUST use the fetch_documentation tool to consult the relevant docs. 2. When citing documentation, summarize clearly and include relevant context from the content. 3. Do not use any URLs outside of the allowed domain. 4. If a documentation fetch fails, tell the user and proceed with your best expert understanding. You can access official documentation from the following approved sources: {llms_txt_content} You MUST consult the documentation to get up to date documentation before answering a user's question about LangGraph. Your answers should be clear, concise, and technically accurate. """ tools = [fetch_documentation] model = init_chat_model("claude-sonnet-4-0", max_tokens=32_000) agent = create_agent( model=model, tools=tools, system_prompt=system_prompt, name="Agentic RAG", ) response = agent.invoke({ 'messages': [ HumanMessage(content=( "Write a short example of a langgraph agent using the " "prebuilt create react agent. the agent should be able " "to look up stock pricing information." )) ] }) print(response['messages'][-1].content) Tutorial: Retrieval-Augmented Generation (RAG)See how to build a Q&A chatbot that can answer questions grounded in your data using Retrieval-Augmented Generation. This tutorial walks through two approaches: A RAG agent that runs searches with a flexible tool—great for general-purpose use. A 2-step RAG chain that requires just one LLM call per query—fast and efficient for simpler tasks. Learn more ​Hybrid RAG Hybrid RAG combines characteristics of both 2-Step and Agentic RAG. It introduces intermediate steps such as query preprocessing, retrieval validation, and post-generation checks. These systems offer more flexibility than fixed pipelines while maintaining some control over execution. Typical components include: Query enhancement: Modify the input question to improve retrieval quality. This can involve rewriting unclear queries, generating multiple variations, or expanding queries with additional context. Retrieval validation: Evaluate whether retrieved documents are relevant and sufficient. If not, the system may refine the query and retrieve again. Answer validation: Check the generated answer for accuracy, completeness, and alignment with source content. If needed, the system can regenerate or revise the answer. The architecture often supports multiple iterations between these steps: This architecture is suitable for: Applications with ambiguous or underspecified queries Systems that require validation or quality control steps Workflows involving multiple sources or iterative refinement Tutorial: Agentic RAG with Self-CorrectionAn example of Hybrid RAG that combines agentic reasoning with retrieval and self-correction.Learn more Edit this page on GitHub or file an issue. Connect these docs to Claude, VSCode, and more via MCP for real-time answers.Was this page helpful?YesNoCustom workflowPreviousLong-term memoryNext⌘IDocs by LangChain home pagegithubxlinkedinyoutubeResourcesForumChangelogLangChain AcademyTrust CenterCompanyAboutCareersBloggithubxlinkedinyoutubePowered by

CopyDocument

Pattern 4: Docs by LangChain home pageLangChain + LangGraphSearch...⌘KSupportGitHubTry LangSmithTry LangSmithSearch...NavigationStreamingOverviewLangChainLangGraphDeep AgentsIntegrationsLearnReferenceContributePythonOverviewGet startedInstallQuickstartChangelogPhilosophyCore componentsAgentsModelsMessagesToolsShort-term memoryStreamingOverviewFrontendStructured outputMiddlewareOverviewBuilt-in middlewareCustom middlewareAdvanced usageGuardrailsRuntimeContext engineeringModel Context Protocol (MCP)Human-in-the-loopMulti-agentRetrievalLong-term memoryAgent developmentLangSmith StudioTestAgent Chat UIDeploy with LangSmithDeploymentObservabilityOn this pageOverviewSupported stream modesAgent progressLLM tokensCustom updatesStream multiple modesCommon patternsStreaming tool callsAccessing completed messagesStreaming with human-in-the-loopStreaming from sub-agentsDisable streamingRelatedCore componentsStreamingOverviewCopy pageStream real-time updates from agent runsCopy pageLangChain implements a streaming system to surface real-time updates. Streaming is crucial for enhancing the responsiveness of applications built on LLMs. By displaying output progressively, even before a complete response is ready, streaming significantly improves user experience (UX), particularly when dealing with the latency of LLMs. ​Overview LangChain’s streaming system lets you surface live feedback from agent runs to your application. What’s possible with LangChain streaming: Stream agent progress — get state updates after each agent step. Stream LLM tokens — stream language model tokens as they’re generated. Stream custom updates — emit user-defined signals (e.g., "Fetched 10/100 records"). Stream multiple modes — choose from updates (agent progress), messages (LLM tokens + metadata), or custom (arbitrary user data). See the common patterns section below for additional end-to-end examples. ​Supported stream modes Pass one or more of the following stream modes as a list to the stream or astream methods: ModeDescriptionupdatesStreams state updates after each agent step. If multiple updates are made in the same step (e.g., multiple nodes are run), those updates are streamed separately.messagesStreams tuples of (token, metadata) from any graph nodes where an LLM is invoked.customStreams custom data from inside your graph nodes using the stream writer. ​Agent progress To stream agent progress, use the stream or astream methods with stream_mode="updates". This emits an event after every agent step. For example, if you have an agent that calls a tool once, you should see the following updates: LLM node: AIMessage with tool call requests Tool node: ToolMessage with execution result LLM node: Final AI response Streaming agent progressCopyfrom langchain.agents import create_agent def get_weather(city: str) -> str: """Get weather for a given city.""" return f"It's always sunny in {city}!" agent = create_agent( model="gpt-5-nano", tools=[get_weather], ) for chunk in agent.stream( {"messages": [{"role": "user", "content": "What is the weather in SF?"}]}, stream_mode="updates", ): for step, data in chunk.items(): print(f"step: {step}") print(f"content: {data['messages'][-1].content_blocks}") OutputCopystep: model content: [{'type': 'tool_call', 'name': 'get_weather', 'args': {'city': 'San Francisco'}, 'id': 'call_OW2NYNsNSKhRZpjW0wm2Aszd'}] step: tools content: [{'type': 'text', 'text': "It's always sunny in San Francisco!"}] step: model content: [{'type': 'text', 'text': 'It's always sunny in San Francisco!'}] ​LLM tokens To stream tokens as they are produced by the LLM, use stream_mode="messages". Below you can see the output of the agent streaming tool calls and the final response. Streaming LLM tokensCopyfrom langchain.agents import create_agent def get_weather(city: str) -> str: """Get weather for a given city.""" return f"It's always sunny in {city}!" agent = create_agent( model="gpt-5-nano", tools=[get_weather], ) for token, metadata in agent.stream( {"messages": [{"role": "user", "content": "What is the weather in SF?"}]}, stream_mode="messages", ): print(f"node: {metadata['langgraph_node']}") print(f"content: {token.content_blocks}") print("\n") OutputCopynode: model content: [{'type': 'tool_call_chunk', 'id': 'call_vbCyBcP8VuneUzyYlSBZZsVa', 'name': 'get_weather', 'args': '', 'index': 0}] node: model content: [{'type': 'tool_call_chunk', 'id': None, 'name': None, 'args': '{"', 'index': 0}] node: model content: [{'type': 'tool_call_chunk', 'id': None, 'name': None, 'args': 'city', 'index': 0}] node: model content: [{'type': 'tool_call_chunk', 'id': None, 'name': None, 'args': '":"', 'index': 0}] node: model content: [{'type': 'tool_call_chunk', 'id': None, 'name': None, 'args': 'San', 'index': 0}] node: model content: [{'type': 'tool_call_chunk', 'id': None, 'name': None, 'args': ' Francisco', 'index': 0}] node: model content: [{'type': 'tool_call_chunk', 'id': None, 'name': None, 'args': '"}', 'index': 0}] node: model content: [] node: tools content: [{'type': 'text', 'text': "It's always sunny in San Francisco!"}] node: model content: [] node: model content: [{'type': 'text', 'text': 'Here'}] node: model content: [{'type': 'text', 'text': ''s'}] node: model content: [{'type': 'text', 'text': ' what'}] node: model content: [{'type': 'text', 'text': ' I'}] node: model content: [{'type': 'text', 'text': ' got'}] node: model content: [{'type': 'text', 'text': ':'}] node: model content: [{'type': 'text', 'text': ' "'}] node: model content: [{'type': 'text', 'text': "It's"}] node: model content: [{'type': 'text', 'text': ' always'}] node: model content: [{'type': 'text', 'text': ' sunny'}] node: model content: [{'type': 'text', 'text': ' in'}] node: model content: [{'type': 'text', 'text': ' San'}] node: model content: [{'type': 'text', 'text': ' Francisco'}] node: model content: [{'type': 'text', 'text': '!"\n\n'}] See all 94 lines ​Custom updates To stream updates from tools as they are executed, you can use get_stream_writer. Streaming custom updatesCopyfrom langchain.agents import create_agent from langgraph.config import get_stream_writer def get_weather(city: str) -> str: """Get weather for a given city.""" writer = get_stream_writer() # stream any arbitrary data writer(f"Looking up data for city: {city}") writer(f"Acquired data for city: {city}") return f"It's always sunny in {city}!" agent = create_agent( model="claude-sonnet-4-5-20250929", tools=[get_weather], ) for chunk in agent.stream( {"messages": [{"role": "user", "content": "What is the weather in SF?"}]}, stream_mode="custom" ): print(chunk) OutputCopyLooking up data for city: San Francisco Acquired data for city: San Francisco If you add get_stream_writer inside your tool, you won’t be able to invoke the tool outside of a LangGraph execution context. ​Stream multiple modes You can specify multiple streaming modes by passing stream mode as a list: stream_mode=["updates", "custom"]. The streamed outputs will be tuples of (mode, chunk) where mode is the name of the stream mode and chunk is the data streamed by that mode. Streaming multiple modesCopyfrom langchain.agents import create_agent from langgraph.config import get_stream_writer def get_weather(city: str) -> str: """Get weather for a given city.""" writer = get_stream_writer() writer(f"Looking up data for city: {city}") writer(f"Acquired data for city: {city}") return f"It's always sunny in {city}!" agent = create_agent( model="gpt-5-nano", tools=[get_weather], ) for stream_mode, chunk in agent.stream( {"messages": [{"role": "user", "content": "What is the weather in SF?"}]}, stream_mode=["updates", "custom"] ): print(f"stream_mode: {stream_mode}") print(f"content: {chunk}") print("\n") OutputCopystream_mode: updates content: {'model': {'messages': [AIMessage(content='', response_metadata={'token_usage': {'completion_tokens': 280, 'prompt_tokens': 132, 'total_tokens': 412, 'completion_tokens_details': {'accepted_prediction_tokens': 0, 'audio_tokens': 0, 'reasoning_tokens': 256, 'rejected_prediction_tokens': 0}, 'prompt_tokens_details': {'audio_tokens': 0, 'cached_tokens': 0}}, 'model_provider': 'openai', 'model_name': 'gpt-5-nano-2025-08-07', 'system_fingerprint': None, 'id': 'chatcmpl-C9tlgBzGEbedGYxZ0rTCz5F7OXpL7', 'service_tier': 'default', 'finish_reason': 'tool_calls', 'logprobs': None}, id='lc_run--480c07cb-e405-4411-aa7f-0520fddeed66-0', tool_calls=[{'name': 'get_weather', 'args': {'city': 'San Francisco'}, 'id': 'call_KTNQIftMrl9vgNwEfAJMVu7r', 'type': 'tool_call'}], usage_metadata={'input_tokens': 132, 'output_tokens': 280, 'total_tokens': 412, 'input_token_details': {'audio': 0, 'cache_read': 0}, 'output_token_details': {'audio': 0, 'reasoning': 256}})]}} stream_mode: custom content: Looking up data for city: San Francisco stream_mode: custom content: Acquired data for city: San Francisco stream_mode: updates content: {'tools': {'messages': [ToolMessage(content="It's always sunny in San Francisco!", name='get_weather', tool_call_id='call_KTNQIftMrl9vgNwEfAJMVu7r')]}} stream_mode: updates content: {'model': {'messages': [AIMessage(content='San Francisco weather: It's always sunny in San Francisco!\n\n', response_metadata={'token_usage': {'completion_tokens': 764, 'prompt_tokens': 168, 'total_tokens': 932, 'completion_tokens_details': {'accepted_prediction_tokens': 0, 'audio_tokens': 0, 'reasoning_tokens': 704, 'rejected_prediction_tokens': 0}, 'prompt_tokens_details': {'audio_tokens': 0, 'cached_tokens': 0}}, 'model_provider': 'openai', 'model_name': 'gpt-5-nano-2025-08-07', 'system_fingerprint': None, 'id': 'chatcmpl-C9tljDFVki1e1haCyikBptAuXuHYG', 'service_tier': 'default', 'finish_reason': 'stop', 'logprobs': None}, id='lc_run--acbc740a-18fe-4a14-8619-da92a0d0ee90-0', usage_metadata={'input_tokens': 168, 'output_tokens': 764, 'total_tokens': 932, 'input_token_details': {'audio': 0, 'cache_read': 0}, 'output_token_details': {'audio': 0, 'reasoning': 704}})]}} ​Common patterns Below are examples showing common use cases for streaming. ​Streaming tool calls You may want to stream both: Partial JSON as tool calls are generated The completed, parsed tool calls that are executed Specifying stream_mode="messages" will stream incremental message chunks generated by all LLM calls in the agent. To access the completed messages with parsed tool calls: If those messages are tracked in the state (as in the model node of create_agent), use stream_mode=["messages", "updates"] to access completed messages through state updates (demonstrated below). If those messages are not tracked in the state, use custom updates or aggregate the chunks during the streaming loop (next section). Refer to the section below on streaming from sub-agents if your agent includes multiple LLMs. Copyfrom typing import Any from langchain.agents import create_agent from langchain.messages import AIMessage, AIMessageChunk, AnyMessage, ToolMessage def get_weather(city: str) -> str: """Get weather for a given city.""" return f"It's always sunny in {city}!" agent = create_agent("openai:gpt-5.2", tools=[get_weather]) def _render_message_chunk(token: AIMessageChunk) -> None: if token.text: print(token.text, end="|") if token.tool_call_chunks: print(token.tool_call_chunks) # N.B. all content is available through token.content_blocks def _render_completed_message(message: AnyMessage) -> None: if isinstance(message, AIMessage) and message.tool_calls: print(f"Tool calls: {message.tool_calls}") if isinstance(message, ToolMessage): print(f"Tool response: {message.content_blocks}") input_message = {"role": "user", "content": "What is the weather in Boston?"} for stream_mode, data in agent.stream( {"messages": [input_message]}, stream_mode=["messages", "updates"], ): if stream_mode == "messages": token, metadata = data if isinstance(token, AIMessageChunk): _render_message_chunk(token) if stream_mode == "updates": for source, update in data.items(): if source in ("model", "tools"): # source captures node name _render_completed_message(update["messages"][-1]) OutputCopy[{'name': 'get_weather', 'args': '', 'id': 'call_D3Orjr89KgsLTZ9hTzYv7Hpf', 'index': 0, 'type': 'tool_call_chunk'}] [{'name': None, 'args': '{"', 'id': None, 'index': 0, 'type': 'tool_call_chunk'}] [{'name': None, 'args': 'city', 'id': None, 'index': 0, 'type': 'tool_call_chunk'}] [{'name': None, 'args': '":"', 'id': None, 'index': 0, 'type': 'tool_call_chunk'}] [{'name': None, 'args': 'Boston', 'id': None, 'index': 0, 'type': 'tool_call_chunk'}] [{'name': None, 'args': '"}', 'id': None, 'index': 0, 'type': 'tool_call_chunk'}] Tool calls: [{'name': 'get_weather', 'args': {'city': 'Boston'}, 'id': 'call_D3Orjr89KgsLTZ9hTzYv7Hpf', 'type': 'tool_call'}] Tool response: [{'type': 'text', 'text': "It's always sunny in Boston!"}] The| weather| in| Boston| is| |sun|ny||.| See all 9 lines ​Accessing completed messages If completed messages are tracked in an agent’s state, you can use stream_mode=["messages", "updates"] as demonstrated above to access completed messages during streaming. In some cases, completed messages are not reflected in state updates. If you have access to the agent internals, you can use custom updates to access these messages during streaming. Otherwise, you can aggregate message chunks in the streaming loop (see below). Consider the below example, where we incorporate a stream writer into a simplified guardrail middleware. This middleware demonstrates tool calling to generate a structured “safe / unsafe” evaluation (one could also use structured outputs for this): Copyfrom typing import Any, Literal from langchain.agents.middleware import after_agent, AgentState from langgraph.runtime import Runtime from langchain.messages import AIMessage from langchain.chat_models import init_chat_model from langgraph.config import get_stream_writer from pydantic import BaseModel class ResponseSafety(BaseModel): """Evaluate a response as safe or unsafe.""" evaluation: Literal["safe", "unsafe"] safety_model = init_chat_model("openai:gpt-5.2") @after_agent(can_jump_to=["end"]) def safety_guardrail(state: AgentState, runtime: Runtime) -> dict[str, Any] | None: """Model-based guardrail: Use an LLM to evaluate response safety.""" stream_writer = get_stream_writer() # Get the model response if not state["messages"]: return None last_message = state["messages"][-1] if not isinstance(last_message, AIMessage): return None # Use another model to evaluate safety model_with_tools = safety_model.bind_tools([ResponseSafety], tool_choice="any") result = model_with_tools.invoke( [ { "role": "system", "content": "Evaluate this AI response as generally safe or unsafe." }, { "role": "user", "content": f"AI response: {last_message.text}" } ] ) stream_writer(result) tool_call = result.tool_calls[0] if tool_call["args"]["evaluation"] == "unsafe": last_message.content = "I cannot provide that response. Please rephrase your request." return None We can then incorporate this middleware into our agent and include its custom stream events: Copyfrom typing import Any from langchain.agents import create_agent from langchain.messages import AIMessageChunk, AIMessage, AnyMessage def get_weather(city: str) -> str: """Get weather for a given city.""" return f"It's always sunny in {city}!" agent = create_agent( model="openai:gpt-5.2", tools=[get_weather], middleware=[safety_guardrail], ) def _render_message_chunk(token: AIMessageChunk) -> None: if token.text: print(token.text, end="|") if token.tool_call_chunks: print(token.tool_call_chunks) def _render_completed_message(message: AnyMessage) -> None: if isinstance(message, AIMessage) and message.tool_calls: print(f"Tool calls: {message.tool_calls}") if isinstance(message, ToolMessage): print(f"Tool response: {message.content_blocks}") input_message = {"role": "user", "content": "What is the weather in Boston?"} for stream_mode, data in agent.stream( {"messages": [input_message]}, stream_mode=["messages", "updates", "custom"], ): if stream_mode == "messages": token, metadata = data if isinstance(token, AIMessageChunk): _render_message_chunk(token) if stream_mode == "updates": for source, update in data.items(): if source in ("model", "tools"): _render_completed_message(update["messages"][-1]) if stream_mode == "custom": # access completed message in stream print(f"Tool calls: {data.tool_calls}") OutputCopy[{'name': 'get_weather', 'args': '', 'id': 'call_je6LWgxYzuZ84mmoDalTYMJC', 'index': 0, 'type': 'tool_call_chunk'}] [{'name': None, 'args': '{"', 'id': None, 'index': 0, 'type': 'tool_call_chunk'}] [{'name': None, 'args': 'city', 'id': None, 'index': 0, 'type': 'tool_call_chunk'}] [{'name': None, 'args': '":"', 'id': None, 'index': 0, 'type': 'tool_call_chunk'}] [{'name': None, 'args': 'Boston', 'id': None, 'index': 0, 'type': 'tool_call_chunk'}] [{'name': None, 'args': '"}', 'id': None, 'index': 0, 'type': 'tool_call_chunk'}] Tool calls: [{'name': 'get_weather', 'args': {'city': 'Boston'}, 'id': 'call_je6LWgxYzuZ84mmoDalTYMJC', 'type': 'tool_call'}] Tool response: [{'type': 'text', 'text': "It's always sunny in Boston!"}] The| weather| in| |Boston|| is| |sun|ny||.|[{'name': 'ResponseSafety', 'args': '', 'id': 'call_O8VJIbOG4Q9nQF0T8ltVi58O', 'index': 0, 'type': 'tool_call_chunk'}] [{'name': None, 'args': '{"', 'id': None, 'index': 0, 'type': 'tool_call_chunk'}] [{'name': None, 'args': 'evaluation', 'id': None, 'index': 0, 'type': 'tool_call_chunk'}] [{'name': None, 'args': '":"', 'id': None, 'index': 0, 'type': 'tool_call_chunk'}] [{'name': None, 'args': 'safe', 'id': None, 'index': 0, 'type': 'tool_call_chunk'}] [{'name': None, 'args': '"}', 'id': None, 'index': 0, 'type': 'tool_call_chunk'}] Tool calls: [{'name': 'ResponseSafety', 'args': {'evaluation': 'safe'}, 'id': 'call_O8VJIbOG4Q9nQF0T8ltVi58O', 'type': 'tool_call'}] See all 15 lines Alternatively, if you aren’t able to add custom events to the stream, you can aggregate message chunks within the streaming loop: Copyinput_message = {"role": "user", "content": "What is the weather in Boston?"} full_message = None for stream_mode, data in agent.stream( {"messages": [input_message]}, stream_mode=["messages", "updates"], ): if stream_mode == "messages": token, metadata = data if isinstance(token, AIMessageChunk): _render_message_chunk(token) full_message = token if full_message is None else full_message + token if token.chunk_position == "last": if full_message.tool_calls: print(f"Tool calls: {full_message.tool_calls}") full_message = None if stream_mode == "updates": for source, update in data.items(): if source == "tools": _render_completed_message(update["messages"][-1]) ​Streaming with human-in-the-loop To handle human-in-the-loop interrupts, we build on the above example: We configure the agent with human-in-the-loop middleware and a checkpointer We collect interrupts generated during the "updates" stream mode We respond to those interrupts with a command Copyfrom typing import Any from langchain.agents import create_agent from langchain.agents.middleware import HumanInTheLoopMiddleware from langchain.messages import AIMessage, AIMessageChunk, AnyMessage, ToolMessage from langgraph.checkpoint.memory import InMemorySaver from langgraph.types import Command, Interrupt def get_weather(city: str) -> str: """Get weather for a given city.""" return f"It's always sunny in {city}!" checkpointer = InMemorySaver() agent = create_agent( "openai:gpt-5.2", tools=[get_weather], middleware=[ HumanInTheLoopMiddleware(interrupt_on={"get_weather": True}), ], checkpointer=checkpointer, ) def _render_message_chunk(token: AIMessageChunk) -> None: if token.text: print(token.text, end="|") if token.tool_call_chunks: print(token.tool_call_chunks) def _render_completed_message(message: AnyMessage) -> None: if isinstance(message, AIMessage) and message.tool_calls: print(f"Tool calls: {message.tool_calls}") if isinstance(message, ToolMessage): print(f"Tool response: {message.content_blocks}") def _render_interrupt(interrupt: Interrupt) -> None: interrupts = interrupt.value for request in interrupts["action_requests"]: print(request["description"]) input_message = { "role": "user", "content": ( "Can you look up the weather in Boston and San Francisco?" ), } config = {"configurable": {"thread_id": "some_id"}} interrupts = [] for stream_mode, data in agent.stream( {"messages": [input_message]}, config=config, stream_mode=["messages", "updates"], ): if stream_mode == "messages": token, metadata = data if isinstance(token, AIMessageChunk): _render_message_chunk(token) if stream_mode == "updates": for source, update in data.items(): if source in ("model", "tools"): _render_completed_message(update["messages"][-1]) if source == "interrupt": interrupts.extend(update) _render_interrupt(update[0]) OutputCopy[{'name': 'get_weather', 'args': '', 'id': 'call_GOwNaQHeqMixay2qy80padfE', 'index': 0, 'type': 'tool_call_chunk'}] [{'name': None, 'args': '{"ci', 'id': None, 'index': 0, 'type': 'tool_call_chunk'}] [{'name': None, 'args': 'ty": ', 'id': None, 'index': 0, 'type': 'tool_call_chunk'}] [{'name': None, 'args': '"Bosto', 'id': None, 'index': 0, 'type': 'tool_call_chunk'}] [{'name': None, 'args': 'n"}', 'id': None, 'index': 0, 'type': 'tool_call_chunk'}] [{'name': 'get_weather', 'args': '', 'id': 'call_Ndb4jvWm2uMA0JDQXu37wDH6', 'index': 1, 'type': 'tool_call_chunk'}] [{'name': None, 'args': '{"ci', 'id': None, 'index': 1, 'type': 'tool_call_chunk'}] [{'name': None, 'args': 'ty": ', 'id': None, 'index': 1, 'type': 'tool_call_chunk'}] [{'name': None, 'args': '"San F', 'id': None, 'index': 1, 'type': 'tool_call_chunk'}] [{'name': None, 'args': 'ranc', 'id': None, 'index': 1, 'type': 'tool_call_chunk'}] [{'name': None, 'args': 'isco"', 'id': None, 'index': 1, 'type': 'tool_call_chunk'}] [{'name': None, 'args': '}', 'id': None, 'index': 1, 'type': 'tool_call_chunk'}] Tool calls: [{'name': 'get_weather', 'args': {'city': 'Boston'}, 'id': 'call_GOwNaQHeqMixay2qy80padfE', 'type': 'tool_call'}, {'name': 'get_weather', 'args': {'city': 'San Francisco'}, 'id': 'call_Ndb4jvWm2uMA0JDQXu37wDH6', 'type': 'tool_call'}] Tool execution requires approval Tool: get_weather Args: {'city': 'Boston'} Tool execution requires approval Tool: get_weather Args: {'city': 'San Francisco'} See all 21 lines We next collect a decision for each interrupt. Importantly, the order of decisions must match the order of actions we collected. To illustrate, we will edit one tool call and accept the other: Copydef _get_interrupt_decisions(interrupt: Interrupt) -> list[dict]: return [ { "type": "edit", "edited_action": { "name": "get_weather", "args": {"city": "Boston, U.K."}, }, } if "boston" in request["description"].lower() else {"type": "approve"} for request in interrupt.value["action_requests"] ] decisions = {} for interrupt in interrupts: decisions[interrupt.id] = { "decisions": _get_interrupt_decisions(interrupt) } decisions OutputCopy{ 'a96c40474e429d661b5b32a8d86f0f3e': { 'decisions': [ { 'type': 'edit', 'edited_action': { 'name': 'get_weather', 'args': {'city': 'Boston, U.K.'} } }, {'type': 'approve'}, ] } } We can then resume by passing a command into the same streaming loop: Copyinterrupts = [] for stream_mode, data in agent.stream( Command(resume=decisions), config=config, stream_mode=["messages", "updates"], ): # Streaming loop is unchanged if stream_mode == "messages": token, metadata = data if isinstance(token, AIMessageChunk): _render_message_chunk(token) if stream_mode == "updates": for source, update in data.items(): if source in ("model", "tools"): _render_completed_message(update["messages"][-1]) if source == "interrupt": interrupts.extend(update) _render_interrupt(update[0]) OutputCopyTool response: [{'type': 'text', 'text': "It's always sunny in Boston, U.K.!"}] Tool response: [{'type': 'text', 'text': "It's always sunny in San Francisco!"}] -| |Boston||:| It|'s| always| sunny| in| Boston|,| U|.K|.| |-| |San| Francisco||:| It|'s| always| sunny| in| San| Francisco|!| ​Streaming from sub-agents When there are multiple LLMs at any point in an agent, it’s often necessary to disambiguate the source of messages as they are generated. To do this, pass a name to each agent when creating it. This name is then available in metadata via the lc_agent_name key when streaming in "messages" mode. Below, we update the streaming tool calls example: We replace our tool with a call_weather_agent tool that invokes an agent internally We add a name to each agent We specify subgraphs=True when creating the stream Our stream processing is identical to before, but we add logic to keep track of what agent is active using create_agent’s name parameter When you set a name on an agent, that name is also attached to any AIMessages generated by that agent. First we construct the agent: Copyfrom typing import Any from langchain.agents import create_agent from langchain.chat_models import init_chat_model from langchain.messages import AIMessage, AnyMessage def get_weather(city: str) -> str: """Get weather for a given city.""" return f"It's always sunny in {city}!" weather_model = init_chat_model("openai:gpt-5.2") weather_agent = create_agent( model=weather_model, tools=[get_weather], name="weather_agent", ) def call_weather_agent(query: str) -> str: """Query the weather agent.""" result = weather_agent.invoke({ "messages": [{"role": "user", "content": query}] }) return result["messages"][-1].text supervisor_model = init_chat_model("openai:gpt-5.2") agent = create_agent( model=supervisor_model, tools=[call_weather_agent], name="supervisor", ) Next, we add logic to the streaming loop to report which agent is emitting tokens: Copydef _render_message_chunk(token: AIMessageChunk) -> None: if token.text: print(token.text, end="|") if token.tool_call_chunks: print(token.tool_call_chunks) def _render_completed_message(message: AnyMessage) -> None: if isinstance(message, AIMessage) and message.tool_calls: print(f"Tool calls: {message.tool_calls}") if isinstance(message, ToolMessage): print(f"Tool response: {message.content_blocks}") input_message = {"role": "user", "content": "What is the weather in Boston?"} current_agent = None for _, stream_mode, data in agent.stream( {"messages": [input_message]}, stream_mode=["messages", "updates"], subgraphs=True, ): if stream_mode == "messages": token, metadata = data if agent_name := metadata.get("lc_agent_name"): if agent_name != current_agent: print(f"🤖 {agent_name}: ") current_agent = agent_name if isinstance(token, AIMessage): _render_message_chunk(token) if stream_mode == "updates": for source, update in data.items(): if source in ("model", "tools"): _render_completed_message(update["messages"][-1]) OutputCopy🤖 supervisor: [{'name': 'call_weather_agent', 'args': '', 'id': 'call_asorzUf0mB6sb7MiKfgojp7I', 'index': 0, 'type': 'tool_call_chunk'}] [{'name': None, 'args': '{"', 'id': None, 'index': 0, 'type': 'tool_call_chunk'}] [{'name': None, 'args': 'query', 'id': None, 'index': 0, 'type': 'tool_call_chunk'}] [{'name': None, 'args': '":"', 'id': None, 'index': 0, 'type': 'tool_call_chunk'}] [{'name': None, 'args': 'Boston', 'id': None, 'index': 0, 'type': 'tool_call_chunk'}] [{'name': None, 'args': ' weather', 'id': None, 'index': 0, 'type': 'tool_call_chunk'}] [{'name': None, 'args': ' right', 'id': None, 'index': 0, 'type': 'tool_call_chunk'}] [{'name': None, 'args': ' now', 'id': None, 'index': 0, 'type': 'tool_call_chunk'}] [{'name': None, 'args': ' and', 'id': None, 'index': 0, 'type': 'tool_call_chunk'}] [{'name': None, 'args': " today's", 'id': None, 'index': 0, 'type': 'tool_call_chunk'}] [{'name': None, 'args': ' forecast', 'id': None, 'index': 0, 'type': 'tool_call_chunk'}] [{'name': None, 'args': '"}', 'id': None, 'index': 0, 'type': 'tool_call_chunk'}] Tool calls: [{'name': 'call_weather_agent', 'args': {'query': "Boston weather right now and today's forecast"}, 'id': 'call_asorzUf0mB6sb7MiKfgojp7I', 'type': 'tool_call'}] 🤖 weather_agent: [{'name': 'get_weather', 'args': '', 'id': 'call_LZ89lT8fW6w8vqck5pZeaDIx', 'index': 0, 'type': 'tool_call_chunk'}] [{'name': None, 'args': '{"', 'id': None, 'index': 0, 'type': 'tool_call_chunk'}] [{'name': None, 'args': 'city', 'id': None, 'index': 0, 'type': 'tool_call_chunk'}] [{'name': None, 'args': '":"', 'id': None, 'index': 0, 'type': 'tool_call_chunk'}] [{'name': None, 'args': 'Boston', 'id': None, 'index': 0, 'type': 'tool_call_chunk'}] [{'name': None, 'args': '"}', 'id': None, 'index': 0, 'type': 'tool_call_chunk'}] Tool calls: [{'name': 'get_weather', 'args': {'city': 'Boston'}, 'id': 'call_LZ89lT8fW6w8vqck5pZeaDIx', 'type': 'tool_call'}] Tool response: [{'type': 'text', 'text': "It's always sunny in Boston!"}] Boston| weather| right| now|:| |Sunny||. |Today|'s| forecast| for| Boston|:| |Sunny| all| day||.|Tool response: [{'type': 'text', 'text': 'Boston weather right now: Sunny.\n\nToday's forecast for Boston: Sunny all day.'}] 🤖 supervisor: Boston| weather| right| now|:| |Sunny||. |Today|'s| forecast| for| Boston|:| |Sunny| all| day||.| See all 30 lines ​Disable streaming In some applications you might need to disable streaming of individual tokens for a given model. This is useful when: Working with multi-agent systems to control which agents stream their output Mixing models that support streaming with those that do not Deploying to LangSmith and wanting to prevent certain model outputs from being streamed to the client Set streaming=False when initializing the model. Copyfrom langchain_openai import ChatOpenAI model = ChatOpenAI( model="gpt-4o", streaming=False ) When deploying to LangSmith, set streaming=False on any models whose output you don’t want streamed to the client. This is configured in your graph code before deployment. Not all chat model integrations support the streaming parameter. If your model doesn’t support it, use disable_streaming=True instead. This parameter is available on all chat models via the base class. See the LangGraph streaming guide for more details. ​Related Frontend streaming — Build React UIs with useStream for real-time agent interactions Streaming with chat models — Stream tokens directly from a chat model without using an agent or graph Streaming with human-in-the-loop — Stream agent progress while handling interrupts for human review LangGraph streaming — Advanced streaming options including values, debug modes, and subgraph streaming Edit this page on GitHub or file an issue. Connect these docs to Claude, VSCode, and more via MCP for real-time answers.Was this page helpful?YesNoShort-term memoryPreviousFrontendNext⌘IDocs by LangChain home pagegithubxlinkedinyoutubeResourcesForumChangelogLangChain AcademyTrust CenterCompanyAboutCareersBloggithubxlinkedinyoutubePowered by

Copy"Fetched 10/100 records"

Pattern 5: Docs by LangChain home pageLangChain + LangGraphSearch...⌘KSupportGitHubTry LangSmithTry LangSmithSearch...NavigationCore componentsShort-term memoryLangChainLangGraphDeep AgentsIntegrationsLearnReferenceContributePythonOverviewGet startedInstallQuickstartChangelogPhilosophyCore componentsAgentsModelsMessagesToolsShort-term memoryStreamingStructured outputMiddlewareOverviewBuilt-in middlewareCustom middlewareAdvanced usageGuardrailsRuntimeContext engineeringModel Context Protocol (MCP)Human-in-the-loopMulti-agentRetrievalLong-term memoryAgent developmentLangSmith StudioTestAgent Chat UIDeploy with LangSmithDeploymentObservabilityOn this pageOverviewUsageIn productionCustomizing agent memoryCommon patternsTrim messagesDelete messagesSummarize messagesAccess memoryToolsRead short-term memory in a toolWrite short-term memory from toolsPromptBefore modelAfter modelCore componentsShort-term memoryCopy pageCopy page​Overview Memory is a system that remembers information about previous interactions. For AI agents, memory is crucial because it lets them remember previous interactions, learn from feedback, and adapt to user preferences. As agents tackle more complex tasks with numerous user interactions, this capability becomes essential for both efficiency and user satisfaction. Short term memory lets your application remember previous interactions within a single thread or conversation. A thread organizes multiple interactions in a session, similar to the way email groups messages in a single conversation. Conversation history is the most common form of short-term memory. Long conversations pose a challenge to today’s LLMs; a full history may not fit inside an LLM’s context window, resulting in an context loss or errors. Even if your model supports the full context length, most LLMs still perform poorly over long contexts. They get “distracted” by stale or off-topic content, all while suffering from slower response times and higher costs. Chat models accept context using messages, which include instructions (a system message) and inputs (human messages). In chat applications, messages alternate between human inputs and model responses, resulting in a list of messages that grows longer over time. Because context windows are limited, many applications can benefit from using techniques to remove or “forget” stale information. ​Usage To add short-term memory (thread-level persistence) to an agent, you need to specify a checkpointer when creating an agent. LangChain’s agent manages short-term memory as a part of your agent’s state.By storing these in the graph’s state, the agent can access the full context for a given conversation while maintaining separation between different threads.State is persisted to a database (or memory) using a checkpointer so the thread can be resumed at any time.Short-term memory updates when the agent is invoked or a step (like a tool call) is completed, and the state is read at the start of each step. Copyfrom langchain.agents import create_agent from langgraph.checkpoint.memory import InMemorySaver agent = create_agent( "gpt-5", tools=[get_user_info], checkpointer=InMemorySaver(), ) agent.invoke( {"messages": [{"role": "user", "content": "Hi! My name is Bob."}]}, {"configurable": {"thread_id": "1"}}, ) ​In production In production, use a checkpointer backed by a database: Copypip install langgraph-checkpoint-postgres Copyfrom langchain.agents import create_agent from langgraph.checkpoint.postgres import PostgresSaver DB_URI = "postgresql://postgres:postgres@localhost:5442/postgres?sslmode=disable" with PostgresSaver.from_conn_string(DB_URI) as checkpointer: checkpointer.setup() # auto create tables in PostgresSql agent = create_agent( "gpt-5", tools=[get_user_info], checkpointer=checkpointer, ) ​Customizing agent memory By default, agents use AgentState to manage short term memory, specifically the conversation history via a messages key. You can extend AgentState to add additional fields. Custom state schemas are passed to create_agent using the state_schema parameter. Copyfrom langchain.agents import create_agent, AgentState from langgraph.checkpoint.memory import InMemorySaver class CustomAgentState(AgentState): user_id: str preferences: dict agent = create_agent( "gpt-5", tools=[get_user_info], state_schema=CustomAgentState, checkpointer=InMemorySaver(), ) # Custom state can be passed in invoke result = agent.invoke( { "messages": [{"role": "user", "content": "Hello"}], "user_id": "user_123", "preferences": {"theme": "dark"} }, {"configurable": {"thread_id": "1"}}) ​Common patterns With short-term memory enabled, long conversations can exceed the LLM’s context window. Common solutions are: Trim messagesRemove first or last N messages (before calling LLM)Delete messagesDelete messages from LangGraph state permanentlySummarize messagesSummarize earlier messages in the history and replace them with a summaryCustom strategiesCustom strategies (e.g., message filtering, etc.) This allows the agent to keep track of the conversation without exceeding the LLM’s context window. ​Trim messages Most LLMs have a maximum supported context window (denominated in tokens). One way to decide when to truncate messages is to count the tokens in the message history and truncate whenever it approaches that limit. If you’re using LangChain, you can use the trim messages utility and specify the number of tokens to keep from the list, as well as the strategy (e.g., keep the last max_tokens) to use for handling the boundary. To trim message history in an agent, use the @before_model middleware decorator: Copyfrom langchain.messages import RemoveMessage from langgraph.graph.message import REMOVE_ALL_MESSAGES from langgraph.checkpoint.memory import InMemorySaver from langchain.agents import create_agent, AgentState from langchain.agents.middleware import before_model from langgraph.runtime import Runtime from langchain_core.runnables import RunnableConfig from typing import Any @before_model def trim_messages(state: AgentState, runtime: Runtime) -> dict[str, Any] | None: """Keep only the last few messages to fit context window.""" messages = state["messages"] if len(messages) <= 3: return None # No changes needed first_msg = messages[0] recent_messages = messages[-3:] if len(messages) % 2 == 0 else messages[-4:] new_messages = [first_msg] + recent_messages return { "messages": [ RemoveMessage(id=REMOVE_ALL_MESSAGES), *new_messages ] } agent = create_agent( your_model_here, tools=your_tools_here, middleware=[trim_messages], checkpointer=InMemorySaver(), ) config: RunnableConfig = {"configurable": {"thread_id": "1"}} agent.invoke({"messages": "hi, my name is bob"}, config) agent.invoke({"messages": "write a short poem about cats"}, config) agent.invoke({"messages": "now do the same but for dogs"}, config) final_response = agent.invoke({"messages": "what's my name?"}, config) final_response["messages"][-1].pretty_print() """ ================================== Ai Message ================================== Your name is Bob. You told me that earlier. If you'd like me to call you a nickname or use a different name, just say the word. """ ​Delete messages You can delete messages from the graph state to manage the message history. This is useful when you want to remove specific messages or clear the entire message history. To delete messages from the graph state, you can use the RemoveMessage. For RemoveMessage to work, you need to use a state key with add_messages reducer. The default AgentState provides this. To remove specific messages: Copyfrom langchain.messages import RemoveMessage def delete_messages(state): messages = state["messages"] if len(messages) > 2: # remove the earliest two messages return {"messages": [RemoveMessage(id=m.id) for m in messages[:2]]} To remove all messages: Copyfrom langgraph.graph.message import REMOVE_ALL_MESSAGES def delete_messages(state): return {"messages": [RemoveMessage(id=REMOVE_ALL_MESSAGES)]} When deleting messages, make sure that the resulting message history is valid. Check the limitations of the LLM provider you’re using. For example: Some providers expect message history to start with a user message Most providers require assistant messages with tool calls to be followed by corresponding tool result messages. Copyfrom langchain.messages import RemoveMessage from langchain.agents import create_agent, AgentState from langchain.agents.middleware import after_model from langgraph.checkpoint.memory import InMemorySaver from langgraph.runtime import Runtime from langchain_core.runnables import RunnableConfig @after_model def delete_old_messages(state: AgentState, runtime: Runtime) -> dict | None: """Remove old messages to keep conversation manageable.""" messages = state["messages"] if len(messages) > 2: # remove the earliest two messages return {"messages": [RemoveMessage(id=m.id) for m in messages[:2]]} return None agent = create_agent( "gpt-5-nano", tools=[], system_prompt="Please be concise and to the point.", middleware=[delete_old_messages], checkpointer=InMemorySaver(), ) config: RunnableConfig = {"configurable": {"thread_id": "1"}} for event in agent.stream( {"messages": [{"role": "user", "content": "hi! I'm bob"}]}, config, stream_mode="values", ): print([(message.type, message.content) for message in event["messages"]]) for event in agent.stream( {"messages": [{"role": "user", "content": "what's my name?"}]}, config, stream_mode="values", ): print([(message.type, message.content) for message in event["messages"]]) Copy[('human', "hi! I'm bob")] [('human', "hi! I'm bob"), ('ai', 'Hi Bob! Nice to meet you. How can I help you today? I can answer questions, brainstorm ideas, draft text, explain things, or help with code.')] [('human', "hi! I'm bob"), ('ai', 'Hi Bob! Nice to meet you. How can I help you today? I can answer questions, brainstorm ideas, draft text, explain things, or help with code.'), ('human', "what's my name?")] [('human', "hi! I'm bob"), ('ai', 'Hi Bob! Nice to meet you. How can I help you today? I can answer questions, brainstorm ideas, draft text, explain things, or help with code.'), ('human', "what's my name?"), ('ai', 'Your name is Bob. How can I help you today, Bob?')] [('human', "what's my name?"), ('ai', 'Your name is Bob. How can I help you today, Bob?')] ​Summarize messages The problem with trimming or removing messages, as shown above, is that you may lose information from culling of the message queue. Because of this, some applications benefit from a more sophisticated approach of summarizing the message history using a chat model. To summarize message history in an agent, use the built-in SummarizationMiddleware: Copyfrom langchain.agents import create_agent from langchain.agents.middleware import SummarizationMiddleware from langgraph.checkpoint.memory import InMemorySaver from langchain_core.runnables import RunnableConfig checkpointer = InMemorySaver() agent = create_agent( model="gpt-4o", tools=[], middleware=[ SummarizationMiddleware( model="gpt-4o-mini", trigger=("tokens", 4000), keep=("messages", 20) ) ], checkpointer=checkpointer, ) config: RunnableConfig = {"configurable": {"thread_id": "1"}} agent.invoke({"messages": "hi, my name is bob"}, config) agent.invoke({"messages": "write a short poem about cats"}, config) agent.invoke({"messages": "now do the same but for dogs"}, config) final_response = agent.invoke({"messages": "what's my name?"}, config) final_response["messages"][-1].pretty_print() """ ================================== Ai Message ================================== Your name is Bob! """ See SummarizationMiddleware for more configuration options. ​Access memory You can access and modify the short-term memory (state) of an agent in several ways: ​Tools ​Read short-term memory in a tool Access short term memory (state) in a tool using the runtime parameter (typed as ToolRuntime). The runtime parameter is hidden from the tool signature (so the model doesn’t see it), but the tool can access the state through it. Copyfrom langchain.agents import create_agent, AgentState from langchain.tools import tool, ToolRuntime class CustomState(AgentState): user_id: str @tool def get_user_info( runtime: ToolRuntime ) -> str: """Look up user info.""" user_id = runtime.state["user_id"] return "User is John Smith" if user_id == "user_123" else "Unknown user" agent = create_agent( model="gpt-5-nano", tools=[get_user_info], state_schema=CustomState, ) result = agent.invoke({ "messages": "look up user information", "user_id": "user_123" }) print(result["messages"][-1].content) # > User is John Smith. ​Write short-term memory from tools To modify the agent’s short-term memory (state) during execution, you can return state updates directly from the tools. This is useful for persisting intermediate results or making information accessible to subsequent tools or prompts. Copyfrom langchain.tools import tool, ToolRuntime from langchain_core.runnables import RunnableConfig from langchain.messages import ToolMessage from langchain.agents import create_agent, AgentState from langgraph.types import Command from pydantic import BaseModel class CustomState(AgentState): user_name: str class CustomContext(BaseModel): user_id: str @tool def update_user_info( runtime: ToolRuntime[CustomContext, CustomState], ) -> Command: """Look up and update user info.""" user_id = runtime.context.user_id name = "John Smith" if user_id == "user_123" else "Unknown user" return Command(update={ "user_name": name, # update the message history "messages": [ ToolMessage( "Successfully looked up user information", tool_call_id=runtime.tool_call_id ) ] }) @tool def greet( runtime: ToolRuntime[CustomContext, CustomState] ) -> str | Command: """Use this to greet the user once you found their info.""" user_name = runtime.state.get("user_name", None) if user_name is None: return Command(update={ "messages": [ ToolMessage( "Please call the 'update_user_info' tool it will get and update the user's name.", tool_call_id=runtime.tool_call_id ) ] }) return f"Hello {user_name}!" agent = create_agent( model="gpt-5-nano", tools=[update_user_info, greet], state_schema=CustomState, context_schema=CustomContext, ) agent.invoke( {"messages": [{"role": "user", "content": "greet the user"}]}, context=CustomContext(user_id="user_123"), ) ​Prompt Access short term memory (state) in middleware to create dynamic prompts based on conversation history or custom state fields. Copyfrom langchain.agents import create_agent from typing import TypedDict from langchain.agents.middleware import dynamic_prompt, ModelRequest class CustomContext(TypedDict): user_name: str def get_weather(city: str) -> str: """Get the weather in a city.""" return f"The weather in {city} is always sunny!" @dynamic_prompt def dynamic_system_prompt(request: ModelRequest) -> str: user_name = request.runtime.context["user_name"] system_prompt = f"You are a helpful assistant. Address the user as {user_name}." return system_prompt agent = create_agent( model="gpt-5-nano", tools=[get_weather], middleware=[dynamic_system_prompt], context_schema=CustomContext, ) result = agent.invoke( {"messages": [{"role": "user", "content": "What is the weather in SF?"}]}, context=CustomContext(user_name="John Smith"), ) for msg in result["messages"]: msg.pretty_print() OutputCopy================================ Human Message ================================= What is the weather in SF? ================================== Ai Message ================================== Tool Calls: get_weather (call_WFQlOGn4b2yoJrv7cih342FG) Call ID: call_WFQlOGn4b2yoJrv7cih342FG Args: city: San Francisco ================================= Tool Message ================================= Name: get_weather The weather in San Francisco is always sunny! ================================== Ai Message ================================== Hi John Smith, the weather in San Francisco is always sunny! ​Before model Access short term memory (state) in @before_model middleware to process messages before model calls. Copyfrom langchain.messages import RemoveMessage from langgraph.graph.message import REMOVE_ALL_MESSAGES from langgraph.checkpoint.memory import InMemorySaver from langchain.agents import create_agent, AgentState from langchain.agents.middleware import before_model from langchain_core.runnables import RunnableConfig from langgraph.runtime import Runtime from typing import Any @before_model def trim_messages(state: AgentState, runtime: Runtime) -> dict[str, Any] | None: """Keep only the last few messages to fit context window.""" messages = state["messages"] if len(messages) <= 3: return None # No changes needed first_msg = messages[0] recent_messages = messages[-3:] if len(messages) % 2 == 0 else messages[-4:] new_messages = [first_msg] + recent_messages return { "messages": [ RemoveMessage(id=REMOVE_ALL_MESSAGES), *new_messages ] } agent = create_agent( "gpt-5-nano", tools=[], middleware=[trim_messages], checkpointer=InMemorySaver() ) config: RunnableConfig = {"configurable": {"thread_id": "1"}} agent.invoke({"messages": "hi, my name is bob"}, config) agent.invoke({"messages": "write a short poem about cats"}, config) agent.invoke({"messages": "now do the same but for dogs"}, config) final_response = agent.invoke({"messages": "what's my name?"}, config) final_response["messages"][-1].pretty_print() """ ================================== Ai Message ================================== Your name is Bob. You told me that earlier. If you'd like me to call you a nickname or use a different name, just say the word. """ ​After model Access short term memory (state) in @after_model middleware to process messages after model calls. Copyfrom langchain.messages import RemoveMessage from langgraph.checkpoint.memory import InMemorySaver from langchain.agents import create_agent, AgentState from langchain.agents.middleware import after_model from langgraph.runtime import Runtime @after_model def validate_response(state: AgentState, runtime: Runtime) -> dict | None: """Remove messages containing sensitive words.""" STOP_WORDS = ["password", "secret"] last_message = state["messages"][-1] if any(word in last_message.content for word in STOP_WORDS): return {"messages": [RemoveMessage(id=last_message.id)]} return None agent = create_agent( model="gpt-5-nano", tools=[], middleware=[validate_response], checkpointer=InMemorySaver(), ) Edit this page on GitHub or file an issue. Connect these docs to Claude, VSCode, and more via MCP for real-time answers.Was this page helpful?YesNoToolsPreviousOverviewNext⌘IDocs by LangChain home pagegithubxlinkedinyoutubeResourcesForumChangelogLangChain AcademyTrust CenterCompanyAboutCareersBloggithubxlinkedinyoutubePowered by

Copycheckpointer

Pattern 6: Docs by LangChain home pageLangChain + LangGraphSearch...⌘KSupportGitHubTry LangSmithTry LangSmithSearch...NavigationIntegrations by componentChat modelsLangChainLangGraphDeep AgentsIntegrationsLearnReferenceContributePythonLangChain integrationsAll providersPopular ProvidersOpenAIAnthropic (Claude)GoogleAWS (Amazon)Hugging FaceMicrosoftOllamaGroqIntegrations by componentChat modelsTools and toolkitsMiddlewareRetrieversText splittersEmbedding modelsVector storesDocument loadersKey-value storesIntegrations by componentChat modelsCopy pageCopy pageChat models are language models that use a sequence of messages as inputs and return messages as outputs (as opposed to traditional, plaintext LLMs). ​Featured models While these LangChain classes support the indicated advanced feature, you may need to refer to provider-specific documentation to learn which hosted models or backends support the feature. ModelTool callingStructured outputMultimodalChatOpenAI✅✅✅ChatAnthropic✅✅✅ChatVertexAI (deprecated)✅✅✅ChatGoogleGenerativeAI✅✅✅AzureChatOpenAI✅✅✅ChatGroq✅✅❌ChatBedrock✅✅❌ChatAmazonNova✅❌✅ChatHuggingFace✅✅❌ChatOllama✅✅❌ChatWatsonx✅✅✅ChatXAI✅✅❌ChatNVIDIA✅✅✅ChatCohere✅✅❌ChatMistralAI✅✅❌ChatTogether✅✅❌ChatFireworks✅✅❌ChatLlamaCpp✅✅❌ChatDatabricks✅✅❌ChatPerplexity❌✅✅ ​Chat completions API Certain model providers offer endpoints that are compatible with OpenAI’s (legacy) Chat Completions API. In such case, you can use ChatOpenAI with a custom base_url to connect to these endpoints. Note that features built on top of the Chat Completions API may not be fully supported by ChatOpenAI; in such cases, consider using a provider-specific class if available (e.g. ChatLiteLLM (community-maintained) for LiteLLM). Example: OpenRouterTo use OpenRouter, you will need to sign up for an account and obtain an API key.Copyfrom langchain_openai import ChatOpenAI model = ChatOpenAI( model="...", # Specify a model available on OpenRouter api_key="OPENROUTER_API_KEY", base_url="https://openrouter.ai/api/v1", ) Refer to the OpenRouter documentation for more details.To capture reasoning tokens, Switch imports from langchain_openai to langchain_deepseek Use ChatDeepSeek instead of ChatOpenAI. You will need to change param base_url to api_base. Adjust reasoning parameters as needed under extra_body, e.g.: Copymodel = ChatDeepSeek( model="...", api_key="...", api_base="https://openrouter.ai/api/v1", extra_body={"reasoning": {"enabled": True}}, ) This is a known limitation with ChatOpenAI and will be addressed in a future release. ​All chat models AbsoView guideAI21 LabsView guideAI/ML APIView guideAlibaba Cloud PAI EASView guideAmazon NovaView guideAnthropicView guideAzureAIChatCompletionsModelView guideAzure OpenAIView guideAzure ML EndpointView guideBaichuan ChatView guideBaidu QianfanView guideBasetenView guideAWS BedrockView guideCerebrasView guideCloudflareWorkersAIView guideCohereView guideContextualAIView guideCoze ChatView guideDappier AIView guideDatabricksView guideDeepInfraView guideDeepSeekView guideEden AIView guideEverlyAIView guideFeatherless AIView guideFireworksView guideChatFriendliView guideGoogle GeminiView guideGoogle Cloud Vertex AIView guideGPTRouterView guideDigitalOcean GradientView guideGreenNodeView guideGroqView guideChatHuggingFaceView guideIBM watsonx.aiView guideJinaChatView guideKineticaView guideKonkoView guideLiteLLMView guideLlama 2 ChatView guideLlama APIView guideLlamaEdgeView guideLlama.cppView guidemaritalkView guideMiniMaxView guideMistralAIView guideMLXView guideModelScopeView guideMoonshotView guideNaverView guideNebiusView guideNetmindView guideNVIDIA AI EndpointsView guideChatOCIModelDeploymentView guideOCIGenAIView guideChatOctoAIView guideOllamaView guideOpenAIView guideOutlinesView guidePerplexityView guidePipeshiftView guideChatPredictionGuardView guidePremAIView guidePromptLayer ChatOpenAIView guideQwen QwQView guideQwenView guideRekaView guideRunPod Chat ModelView guideSambaNovaView guideChatSeekrFlowView guideSnowflake CortexView guideSparkLLM ChatView guideNebula (Symbl.ai)View guideTencent HunyuanView guideTogetherView guideTongyi QwenView guideUpstageView guidevLLM ChatView guideVolc Engine MaasView guideChatWriterView guidexAIView guideXinferenceView guideYandexGPTView guideChatYIView guideYuan2.0View guideZHIPU AIView guide If you’d like to contribute an integration, see Contributing integrations. Edit this page on GitHub or file an issue. Connect these docs to Claude, VSCode, and more via MCP for real-time answers.Was this page helpful?YesNoChatGroqPreviousTools and toolkitsNext⌘IDocs by LangChain home pagegithubxlinkedinyoutubeResourcesForumChangelogLangChain AcademyTrust CenterCompanyAboutCareersBloggithubxlinkedinyoutubePowered by

CopyChatOpenAI

Pattern 7: Docs by LangChain home pageLangChain + LangGraphSearch...⌘KSupportGitHubTry LangSmithTry LangSmithSearch...NavigationMiddlewareCustom middlewareLangChainLangGraphDeep AgentsIntegrationsLearnReferenceContributePythonOverviewGet startedInstallQuickstartChangelogPhilosophyCore componentsAgentsModelsMessagesToolsShort-term memoryStreamingStructured outputMiddlewareOverviewBuilt-in middlewareCustom middlewareAdvanced usageGuardrailsRuntimeContext engineeringModel Context Protocol (MCP)Human-in-the-loopMulti-agentRetrievalLong-term memoryAgent developmentLangSmith StudioTestAgent Chat UIDeploy with LangSmithDeploymentObservabilityOn this pageHooksNode-style hooksWrap-style hooksCreate middlewareDecorator-based middlewareClass-based middlewareCustom state schemaExecution orderAgent jumpsBest practicesExamplesDynamic model selectionTool call monitoringDynamically selecting toolsWorking with system messagesAdditional resourcesMiddlewareCustom middlewareCopy pageCopy pageBuild custom middleware by implementing hooks that run at specific points in the agent execution flow. ​Hooks Middleware provides two styles of hooks to intercept agent execution: Node-style hooksRun sequentially at specific execution points.Wrap-style hooksRun around each model or tool call. ​Node-style hooks Run sequentially at specific execution points. Use for logging, validation, and state updates. Available hooks: before_agent - Before agent starts (once per invocation) before_model - Before each model call after_model - After each model response after_agent - After agent completes (once per invocation) Example: Decorator ClassCopyfrom langchain.agents.middleware import before_model, after_model, AgentState from langchain.messages import AIMessage from langgraph.runtime import Runtime from typing import Any @before_model(can_jump_to=["end"]) def check_message_limit(state: AgentState, runtime: Runtime) -> dict[str, Any] | None: if len(state["messages"]) >= 50: return { "messages": [AIMessage("Conversation limit reached.")], "jump_to": "end" } return None @after_model def log_response(state: AgentState, runtime: Runtime) -> dict[str, Any] | None: print(f"Model returned: {state['messages'][-1].content}") return None Copyfrom langchain.agents.middleware import AgentMiddleware, AgentState, hook_config from langchain.messages import AIMessage from langgraph.runtime import Runtime from typing import Any class MessageLimitMiddleware(AgentMiddleware): def init(self, max_messages: int = 50): super().init() self.max_messages = max_messages @hook_config(can_jump_to=["end"]) def before_model(self, state: AgentState, runtime: Runtime) -> dict[str, Any] | None: if len(state["messages"]) == self.max_messages: return { "messages": [AIMessage("Conversation limit reached.")], "jump_to": "end" } return None def after_model(self, state: AgentState, runtime: Runtime) -> dict[str, Any] | None: print(f"Model returned: {state['messages'][-1].content}") return None ​Wrap-style hooks Intercept execution and control when the handler is called. Use for retries, caching, and transformation. You decide if the handler is called zero times (short-circuit), once (normal flow), or multiple times (retry logic). Available hooks: wrap_model_call - Around each model call wrap_tool_call - Around each tool call Example: Decorator ClassCopyfrom langchain.agents.middleware import wrap_model_call, ModelRequest, ModelResponse from typing import Callable @wrap_model_call def retry_model( request: ModelRequest, handler: Callable[[ModelRequest], ModelResponse], ) -> ModelResponse: for attempt in range(3): try: return handler(request) except Exception as e: if attempt == 2: raise print(f"Retry {attempt + 1}/3 after error: {e}") Copyfrom langchain.agents.middleware import AgentMiddleware, ModelRequest, ModelResponse from typing import Callable class RetryMiddleware(AgentMiddleware): def init(self, max_retries: int = 3): super().init() self.max_retries = max_retries def wrap_model_call( self, request: ModelRequest, handler: Callable[[ModelRequest], ModelResponse], ) -> ModelResponse: for attempt in range(self.max_retries): try: return handler(request) except Exception as e: if attempt == self.max_retries - 1: raise print(f"Retry {attempt + 1}/{self.max_retries} after error: {e}") ​Create middleware You can create middleware in two ways: Decorator-based middlewareQuick and simple for single-hook middleware. Use decorators to wrap individual functions.Class-based middlewareMore powerful for complex middleware with multiple hooks or configuration. ​Decorator-based middleware Quick and simple for single-hook middleware. Use decorators to wrap individual functions. Available decorators: Node-style: @before_agent - Runs before agent starts (once per invocation) @before_model - Runs before each model call @after_model - Runs after each model response @after_agent - Runs after agent completes (once per invocation) Wrap-style: @wrap_model_call - Wraps each model call with custom logic @wrap_tool_call - Wraps each tool call with custom logic Convenience: @dynamic_prompt - Generates dynamic system prompts Example: Copyfrom langchain.agents.middleware import ( before_model, wrap_model_call, AgentState, ModelRequest, ModelResponse, ) from langchain.agents import create_agent from langgraph.runtime import Runtime from typing import Any, Callable @before_model def log_before_model(state: AgentState, runtime: Runtime) -> dict[str, Any] | None: print(f"About to call model with {len(state['messages'])} messages") return None @wrap_model_call def retry_model( request: ModelRequest, handler: Callable[[ModelRequest], ModelResponse], ) -> ModelResponse: for attempt in range(3): try: return handler(request) except Exception as e: if attempt == 2: raise print(f"Retry {attempt + 1}/3 after error: {e}") agent = create_agent( model="gpt-4o", middleware=[log_before_model, retry_model], tools=[...], ) When to use decorators: Single hook needed No complex configuration Quick prototyping ​Class-based middleware More powerful for complex middleware with multiple hooks or configuration. Use classes when you need to define both sync and async implementations for the same hook, or when you want to combine multiple hooks in a single middleware. Example: Copyfrom langchain.agents.middleware import ( AgentMiddleware, AgentState, ModelRequest, ModelResponse, ) from langgraph.runtime import Runtime from typing import Any, Callable class LoggingMiddleware(AgentMiddleware): def before_model(self, state: AgentState, runtime: Runtime) -> dict[str, Any] | None: print(f"About to call model with {len(state['messages'])} messages") return None def after_model(self, state: AgentState, runtime: Runtime) -> dict[str, Any] | None: print(f"Model returned: {state['messages'][-1].content}") return None agent = create_agent( model="gpt-4o", middleware=[LoggingMiddleware()], tools=[...], ) When to use classes: Defining both sync and async implementations for the same hook Multiple hooks needed in a single middleware Complex configuration required (e.g., configurable thresholds, custom models) Reuse across projects with init-time configuration ​Custom state schema Middleware can extend the agent’s state with custom properties. This enables middleware to: Track state across execution: Maintain counters, flags, or other values that persist throughout the agent’s execution lifecycle Share data between hooks: Pass information from before_model to after_model or between different middleware instances Implement cross-cutting concerns: Add functionality like rate limiting, usage tracking, user context, or audit logging without modifying the core agent logic Make conditional decisions: Use accumulated state to determine whether to continue execution, jump to different nodes, or modify behavior dynamically Decorator ClassCopyfrom langchain.agents import create_agent from langchain.messages import HumanMessage from langchain.agents.middleware import AgentState, before_model, after_model from typing_extensions import NotRequired from typing import Any from langgraph.runtime import Runtime class CustomState(AgentState): model_call_count: NotRequired[int] user_id: NotRequired[str] @before_model(state_schema=CustomState, can_jump_to=["end"]) def check_call_limit(state: CustomState, runtime: Runtime) -> dict[str, Any] | None: count = state.get("model_call_count", 0) if count > 10: return {"jump_to": "end"} return None @after_model(state_schema=CustomState) def increment_counter(state: CustomState, runtime: Runtime) -> dict[str, Any] | None: return {"model_call_count": state.get("model_call_count", 0) + 1} agent = create_agent( model="gpt-4o", middleware=[check_call_limit, increment_counter], tools=[], ) # Invoke with custom state result = agent.invoke({ "messages": [HumanMessage("Hello")], "model_call_count": 0, "user_id": "user-123", }) Copyfrom langchain.agents import create_agent from langchain.messages import HumanMessage from langchain.agents.middleware import AgentState, AgentMiddleware from typing_extensions import NotRequired from typing import Any class CustomState(AgentState): model_call_count: NotRequired[int] user_id: NotRequired[str] class CallCounterMiddleware(AgentMiddleware[CustomState]): state_schema = CustomState def before_model(self, state: CustomState, runtime) -> dict[str, Any] | None: count = state.get("model_call_count", 0) if count > 10: return {"jump_to": "end"} return None def after_model(self, state: CustomState, runtime) -> dict[str, Any] | None: return {"model_call_count": state.get("model_call_count", 0) + 1} agent = create_agent( model="gpt-4o", middleware=[CallCounterMiddleware()], tools=[], ) # Invoke with custom state result = agent.invoke({ "messages": [HumanMessage("Hello")], "model_call_count": 0, "user_id": "user-123", }) ​Execution order When using multiple middleware, understand how they execute: Copyagent = create_agent( model="gpt-4o", middleware=[middleware1, middleware2, middleware3], tools=[...], ) Execution flowBefore hooks run in order: middleware1.before_agent() middleware2.before_agent() middleware3.before_agent() Agent loop starts middleware1.before_model() middleware2.before_model() middleware3.before_model() Wrap hooks nest like function calls: middleware1.wrap_model_call() → middleware2.wrap_model_call() → middleware3.wrap_model_call() → model After hooks run in reverse order: middleware3.after_model() middleware2.after_model() middleware1.after_model() Agent loop ends middleware3.after_agent() middleware2.after_agent() middleware1.after_agent() Key rules: before_* hooks: First to last after_* hooks: Last to first (reverse) wrap_* hooks: Nested (first middleware wraps all others) ​Agent jumps To exit early from middleware, return a dictionary with jump_to: Available jump targets: 'end': Jump to the end of the agent execution (or the first after_agent hook) 'tools': Jump to the tools node 'model': Jump to the model node (or the first before_model hook) Decorator ClassCopyfrom langchain.agents.middleware import after_model, hook_config, AgentState from langchain.messages import AIMessage from langgraph.runtime import Runtime from typing import Any @after_model @hook_config(can_jump_to=["end"]) def check_for_blocked(state: AgentState, runtime: Runtime) -> dict[str, Any] | None: last_message = state["messages"][-1] if "BLOCKED" in last_message.content: return { "messages": [AIMessage("I cannot respond to that request.")], "jump_to": "end" } return None Copyfrom langchain.agents.middleware import AgentMiddleware, hook_config, AgentState from langchain.messages import AIMessage from langgraph.runtime import Runtime from typing import Any class BlockedContentMiddleware(AgentMiddleware): @hook_config(can_jump_to=["end"]) def after_model(self, state: AgentState, runtime: Runtime) -> dict[str, Any] | None: last_message = state["messages"][-1] if "BLOCKED" in last_message.content: return { "messages": [AIMessage("I cannot respond to that request.")], "jump_to": "end" } return None ​Best practices Keep middleware focused - each should do one thing well Handle errors gracefully - don’t let middleware errors crash the agent Use appropriate hook types: Node-style for sequential logic (logging, validation) Wrap-style for control flow (retry, fallback, caching) Clearly document any custom state properties Unit test middleware independently before integrating Consider execution order - place critical middleware first in the list Use built-in middleware when possible ​Examples ​Dynamic model selection Decorator ClassCopyfrom langchain.agents.middleware import wrap_model_call, ModelRequest, ModelResponse from langchain.chat_models import init_chat_model from typing import Callable complex_model = init_chat_model("gpt-4o") simple_model = init_chat_model("gpt-4o-mini") @wrap_model_call def dynamic_model( request: ModelRequest, handler: Callable[[ModelRequest], ModelResponse], ) -> ModelResponse: # Use different model based on conversation length if len(request.messages) > 10: model = complex_model else: model = simple_model return handler(request.override(model=model)) Copyfrom langchain.agents.middleware import AgentMiddleware, ModelRequest, ModelResponse from langchain.chat_models import init_chat_model from typing import Callable complex_model = init_chat_model("gpt-4o") simple_model = init_chat_model("gpt-4o-mini") class DynamicModelMiddleware(AgentMiddleware): def wrap_model_call( self, request: ModelRequest, handler: Callable[[ModelRequest], ModelResponse], ) -> ModelResponse: # Use different model based on conversation length if len(request.messages) > 10: model = complex_model else: model = simple_model return handler(request.override(model=model)) ​Tool call monitoring Decorator ClassCopyfrom langchain.agents.middleware import wrap_tool_call from langchain.tools.tool_node import ToolCallRequest from langchain.messages import ToolMessage from langgraph.types import Command from typing import Callable @wrap_tool_call def monitor_tool( request: ToolCallRequest, handler: Callable[[ToolCallRequest], ToolMessage | Command], ) -> ToolMessage | Command: print(f"Executing tool: {request.tool_call['name']}") print(f"Arguments: {request.tool_call['args']}") try: result = handler(request) print(f"Tool completed successfully") return result except Exception as e: print(f"Tool failed: {e}") raise Copyfrom langchain.tools.tool_node import ToolCallRequest from langchain.agents.middleware import AgentMiddleware from langchain.messages import ToolMessage from langgraph.types import Command from typing import Callable class ToolMonitoringMiddleware(AgentMiddleware): def wrap_tool_call( self, request: ToolCallRequest, handler: Callable[[ToolCallRequest], ToolMessage | Command], ) -> ToolMessage | Command: print(f"Executing tool: {request.tool_call['name']}") print(f"Arguments: {request.tool_call['args']}") try: result = handler(request) print(f"Tool completed successfully") return result except Exception as e: print(f"Tool failed: {e}") raise ​Dynamically selecting tools Select relevant tools at runtime to improve performance and accuracy. Benefits: Shorter prompts - Reduce complexity by exposing only relevant tools Better accuracy - Models choose correctly from fewer options Permission control - Dynamically filter tools based on user access Decorator ClassCopyfrom langchain.agents import create_agent from langchain.agents.middleware import wrap_model_call, ModelRequest, ModelResponse from typing import Callable @wrap_model_call def select_tools( request: ModelRequest, handler: Callable[[ModelRequest], ModelResponse], ) -> ModelResponse: """Middleware to select relevant tools based on state/context.""" # Select a small, relevant subset of tools based on state/context relevant_tools = select_relevant_tools(request.state, request.runtime) return handler(request.override(tools=relevant_tools)) agent = create_agent( model="gpt-4o", tools=all_tools, # All available tools need to be registered upfront middleware=[select_tools], ) Copyfrom langchain.agents import create_agent from langchain.agents.middleware import AgentMiddleware, ModelRequest, ModelResponse from typing import Callable class ToolSelectorMiddleware(AgentMiddleware): def wrap_model_call( self, request: ModelRequest, handler: Callable[[ModelRequest], ModelResponse], ) -> ModelResponse: """Middleware to select relevant tools based on state/context.""" # Select a small, relevant subset of tools based on state/context relevant_tools = select_relevant_tools(request.state, request.runtime) return handler(request.override(tools=relevant_tools)) agent = create_agent( model="gpt-4o", tools=all_tools, # All available tools need to be registered upfront middleware=[ToolSelectorMiddleware()], ) ​Working with system messages Modify system messages in middleware using the system_message field on ModelRequest. The system_message field contains a SystemMessage object (even if the agent was created with a string system_prompt). Example: Adding context to system message Decorator ClassCopyfrom langchain.agents.middleware import wrap_model_call, ModelRequest, ModelResponse from langchain.messages import SystemMessage from typing import Callable @wrap_model_call def add_context( request: ModelRequest, handler: Callable[[ModelRequest], ModelResponse], ) -> ModelResponse: # Always work with content blocks new_content = list(request.system_message.content_blocks) + [ {"type": "text", "text": "Additional context."} ] new_system_message = SystemMessage(content=new_content) return handler(request.override(system_message=new_system_message)) Copyfrom langchain.agents.middleware import AgentMiddleware, ModelRequest, ModelResponse from langchain.messages import SystemMessage from typing import Callable class ContextMiddleware(AgentMiddleware): def wrap_model_call( self, request: ModelRequest, handler: Callable[[ModelRequest], ModelResponse], ) -> ModelResponse: # Always work with content blocks new_content = list(request.system_message.content_blocks) + [ {"type": "text", "text": "Additional context."} ] new_system_message = SystemMessage(content=new_content) return handler(request.override(system_message=new_system_message)) Example: Working with cache control (Anthropic) When working with Anthropic models, you can use structured content blocks with cache control directives to cache large system prompts: Decorator ClassCopyfrom langchain.agents.middleware import wrap_model_call, ModelRequest, ModelResponse from langchain.messages import SystemMessage from typing import Callable @wrap_model_call def add_cached_context( request: ModelRequest, handler: Callable[[ModelRequest], ModelResponse], ) -> ModelResponse: # Always work with content blocks new_content = list(request.system_message.content_blocks) + [ { "type": "text", "text": "Here is a large document to analyze:\n\n...", # content up until this point is cached "cache_control": {"type": "ephemeral"} } ] new_system_message = SystemMessage(content=new_content) return handler(request.override(system_message=new_system_message)) Copyfrom langchain.agents.middleware import AgentMiddleware, ModelRequest, ModelResponse from langchain.messages import SystemMessage from typing import Callable class CachedContextMiddleware(AgentMiddleware): def wrap_model_call( self, request: ModelRequest, handler: Callable[[ModelRequest], ModelResponse], ) -> ModelResponse: # Always work with content blocks new_content = list(request.system_message.content_blocks) + [ { "type": "text", "text": "Here is a large document to analyze:\n\n...", "cache_control": {"type": "ephemeral"} # This content will be cached } ] new_system_message = SystemMessage(content=new_content) return handler(request.override(system_message=new_system_message)) Notes: ModelRequest.system_message is always a SystemMessage object, even if the agent was created with system_prompt="string" Use SystemMessage.content_blocks to access content as a list of blocks, regardless of whether the original content was a string or list When modifying system messages, use content_blocks and append new blocks to preserve existing structure You can pass SystemMessage objects directly to create_agent’s system_prompt parameter for advanced use cases like cache control ​Additional resources Middleware API reference Built-in middleware Testing agents Edit this page on GitHub or file an issue. Connect these docs to Claude, VSCode, and more via MCP for real-time answers.Was this page helpful?YesNoBuilt-in middlewarePreviousGuardrailsNext⌘IDocs by LangChain home pagegithubxlinkedinyoutubeResourcesForumChangelogLangChain AcademyTrust CenterCompanyAboutCareersBloggithubxlinkedinyoutubePowered by

Copybefore_agent

Pattern 8: Docs by LangChain home pageLangChain + LangGraphSearch...⌘KSupportGitHubTry LangSmithTry LangSmithSearch...NavigationCore componentsMessagesLangChainLangGraphDeep AgentsIntegrationsLearnReferenceContributePythonOverviewGet startedInstallQuickstartChangelogPhilosophyCore componentsAgentsModelsMessagesToolsShort-term memoryStreamingStructured outputMiddlewareOverviewBuilt-in middlewareCustom middlewareAdvanced usageGuardrailsRuntimeContext engineeringModel Context Protocol (MCP)Human-in-the-loopMulti-agentRetrievalLong-term memoryAgent developmentLangSmith StudioTestAgent Chat UIDeploy with LangSmithDeploymentObservabilityOn this pageBasic usageText promptsMessage promptsDictionary formatMessage typesSystem messageHuman messageText contentMessage metadataAI messageTool callsToken usageStreaming and chunksTool messageMessage contentStandard content blocksMultimodalContent block referenceUse with chat modelsCore componentsMessagesCopy pageCopy pageMessages are the fundamental unit of context for models in LangChain. They represent the input and output of models, carrying both the content and metadata needed to represent the state of a conversation when interacting with an LLM. Messages are objects that contain: Role - Identifies the message type (e.g. system, user) Content - Represents the actual content of the message (like text, images, audio, documents, etc.) Metadata - Optional fields such as response information, message IDs, and token usage LangChain provides a standard message type that works across all model providers, ensuring consistent behavior regardless of the model being called. ​Basic usage The simplest way to use messages is to create message objects and pass them to a model when invoking. Copyfrom langchain.chat_models import init_chat_model from langchain.messages import HumanMessage, AIMessage, SystemMessage model = init_chat_model("gpt-5-nano") system_msg = SystemMessage("You are a helpful assistant.") human_msg = HumanMessage("Hello, how are you?") # Use with chat models messages = [system_msg, human_msg] response = model.invoke(messages) # Returns AIMessage ​Text prompts Text prompts are strings - ideal for straightforward generation tasks where you don’t need to retain conversation history. Copyresponse = model.invoke("Write a haiku about spring") Use text prompts when: You have a single, standalone request You don’t need conversation history You want minimal code complexity ​Message prompts Alternatively, you can pass in a list of messages to the model by providing a list of message objects. Copyfrom langchain.messages import SystemMessage, HumanMessage, AIMessage messages = [ SystemMessage("You are a poetry expert"), HumanMessage("Write a haiku about spring"), AIMessage("Cherry blossoms bloom...") ] response = model.invoke(messages) Use message prompts when: Managing multi-turn conversations Working with multimodal content (images, audio, files) Including system instructions ​Dictionary format You can also specify messages directly in OpenAI chat completions format. Copymessages = [ {"role": "system", "content": "You are a poetry expert"}, {"role": "user", "content": "Write a haiku about spring"}, {"role": "assistant", "content": "Cherry blossoms bloom..."} ] response = model.invoke(messages) ​Message types System message - Tells the model how to behave and provide context for interactions Human message - Represents user input and interactions with the model AI message - Responses generated by the model, including text content, tool calls, and metadata Tool message - Represents the outputs of tool calls ​System message A SystemMessage represent an initial set of instructions that primes the model’s behavior. You can use a system message to set the tone, define the model’s role, and establish guidelines for responses. Basic instructionsCopysystem_msg = SystemMessage("You are a helpful coding assistant.") messages = [ system_msg, HumanMessage("How do I create a REST API?") ] response = model.invoke(messages) Detailed personaCopyfrom langchain.messages import SystemMessage, HumanMessage system_msg = SystemMessage(""" You are a senior Python developer with expertise in web frameworks. Always provide code examples and explain your reasoning. Be concise but thorough in your explanations. """) messages = [ system_msg, HumanMessage("How do I create a REST API?") ] response = model.invoke(messages) ​Human message A HumanMessage represents user input and interactions. They can contain text, images, audio, files, and any other amount of multimodal content. ​Text content Message objectString shortcutCopyresponse = model.invoke([ HumanMessage("What is machine learning?") ]) ​Message metadata Add metadataCopyhuman_msg = HumanMessage( content="Hello!", name="alice", # Optional: identify different users id="msg_123", # Optional: unique identifier for tracing ) The name field behavior varies by provider – some use it for user identification, others ignore it. To check, refer to the model provider’s reference. ​AI message An AIMessage represents the output of a model invocation. They can include multimodal data, tool calls, and provider-specific metadata that you can later access. Copyresponse = model.invoke("Explain AI") print(type(response)) # <class 'langchain.messages.AIMessage'> AIMessage objects are returned by the model when calling it, which contains all of the associated metadata in the response. Providers weigh/contextualize types of messages differently, which means it is sometimes helpful to manually create a new AIMessage object and insert it into the message history as if it came from the model. Copyfrom langchain.messages import AIMessage, SystemMessage, HumanMessage # Create an AI message manually (e.g., for conversation history) ai_msg = AIMessage("I'd be happy to help you with that question!") # Add to conversation history messages = [ SystemMessage("You are a helpful assistant"), HumanMessage("Can you help me?"), ai_msg, # Insert as if it came from the model HumanMessage("Great! What's 2+2?") ] response = model.invoke(messages) Attributes​textstringThe text content of the message.​contentstring | dict[]The raw content of the message.​content_blocksContentBlock[]The standardized content blocks of the message.​tool_callsdict[] | NoneThe tool calls made by the model.Empty if no tools are called.​idstringA unique identifier for the message (either automatically generated by LangChain or returned in the provider response)​usage_metadatadict | NoneThe usage metadata of the message, which can contain token counts when available.​response_metadataResponseMetadata | NoneThe response metadata of the message. ​Tool calls When models make tool calls, they’re included in the AIMessage: Copyfrom langchain.chat_models import init_chat_model model = init_chat_model("gpt-5-nano") def get_weather(location: str) -> str: """Get the weather at a location.""" ... model_with_tools = model.bind_tools([get_weather]) response = model_with_tools.invoke("What's the weather in Paris?") for tool_call in response.tool_calls: print(f"Tool: {tool_call['name']}") print(f"Args: {tool_call['args']}") print(f"ID: {tool_call['id']}") Other structured data, such as reasoning or citations, can also appear in message content. ​Token usage An AIMessage can hold token counts and other usage metadata in its usage_metadata field: Copyfrom langchain.chat_models import init_chat_model model = init_chat_model("gpt-5-nano") response = model.invoke("Hello!") response.usage_metadata Copy{'input_tokens': 8, 'output_tokens': 304, 'total_tokens': 312, 'input_token_details': {'audio': 0, 'cache_read': 0}, 'output_token_details': {'audio': 0, 'reasoning': 256}} See UsageMetadata for details. ​Streaming and chunks During streaming, you’ll receive AIMessageChunk objects that can be combined into a full message object: Copychunks = [] full_message = None for chunk in model.stream("Hi"): chunks.append(chunk) print(chunk.text) full_message = chunk if full_message is None else full_message + chunk Learn more: Streaming tokens from chat models Streaming tokens and/or steps from agents ​Tool message For models that support tool calling, AI messages can contain tool calls. Tool messages are used to pass the results of a single tool execution back to the model. Tools can generate ToolMessage objects directly. Below, we show a simple example. Read more in the tools guide. Copyfrom langchain.messages import AIMessage from langchain.messages import ToolMessage # After a model makes a tool call # (Here, we demonstrate manually creating the messages for brevity) ai_message = AIMessage( content=[], tool_calls=[{ "name": "get_weather", "args": {"location": "San Francisco"}, "id": "call_123" }] ) # Execute tool and create result message weather_result = "Sunny, 72°F" tool_message = ToolMessage( content=weather_result, tool_call_id="call_123" # Must match the call ID ) # Continue conversation messages = [ HumanMessage("What's the weather in San Francisco?"), ai_message, # Model's tool call tool_message, # Tool execution result ] response = model.invoke(messages) # Model processes the result Attributes​contentstringrequiredThe stringified output of the tool call.​tool_call_idstringrequiredThe ID of the tool call that this message is responding to. Must match the ID of the tool call in the AIMessage.​namestringrequiredThe name of the tool that was called.​artifactdictAdditional data not sent to the model but can be accessed programmatically. The artifact field stores supplementary data that won’t be sent to the model but can be accessed programmatically. This is useful for storing raw results, debugging information, or data for downstream processing without cluttering the model’s context.Example: Using artifact for retrieval metadataFor example, a retrieval tool could retrieve a passage from a document for reference by a model. Where message content contains text that the model will reference, an artifact can contain document identifiers or other metadata that an application can use (e.g., to render a page). See example below:Copyfrom langchain.messages import ToolMessage # Sent to model message_content = "It was the best of times, it was the worst of times." # Artifact available downstream artifact = {"document_id": "doc_123", "page": 0} tool_message = ToolMessage( content=message_content, tool_call_id="call_123", name="search_books", artifact=artifact, ) See the RAG tutorial for an end-to-end example of building retrieval agents with LangChain. ​Message content You can think of a message’s content as the payload of data that gets sent to the model. Messages have a content attribute that is loosely-typed, supporting strings and lists of untyped objects (e.g., dictionaries). This allows support for provider-native structures directly in LangChain chat models, such as multimodal content and other data. Separately, LangChain provides dedicated content types for text, reasoning, citations, multi-modal data, server-side tool calls, and other message content. See content blocks below. LangChain chat models accept message content in the content attribute. This may contain either: A string A list of content blocks in a provider-native format A list of LangChain’s standard content blocks See below for an example using multimodal inputs: Copyfrom langchain.messages import HumanMessage # String content human_message = HumanMessage("Hello, how are you?") # Provider-native format (e.g., OpenAI) human_message = HumanMessage(content=[ {"type": "text", "text": "Hello, how are you?"}, {"type": "image_url", "image_url": {"url": "https://example.com/image.jpg"}} ]) # List of standard content blocks human_message = HumanMessage(content_blocks=[ {"type": "text", "text": "Hello, how are you?"}, {"type": "image", "url": "https://example.com/image.jpg"}, ]) Specifying content_blocks when initializing a message will still populate message content, but provides a type-safe interface for doing so. ​Standard content blocks LangChain provides a standard representation for message content that works across providers. Message objects implement a content_blocks property that will lazily parse the content attribute into a standard, type-safe representation. For example, messages generated from ChatAnthropic or ChatOpenAI will include thinking or reasoning blocks in the format of the respective provider, but can be lazily parsed into a consistent ReasoningContentBlock representation: Anthropic OpenAICopyfrom langchain.messages import AIMessage message = AIMessage( content=[ {"type": "thinking", "thinking": "...", "signature": "WaUjzkyp..."}, {"type": "text", "text": "..."}, ], response_metadata={"model_provider": "anthropic"} ) message.content_blocks Copy[{'type': 'reasoning', 'reasoning': '...', 'extras': {'signature': 'WaUjzkyp...'}}, {'type': 'text', 'text': '...'}] Copyfrom langchain.messages import AIMessage message = AIMessage( content=[ { "type": "reasoning", "id": "rs_abc123", "summary": [ {"type": "summary_text", "text": "summary 1"}, {"type": "summary_text", "text": "summary 2"}, ], }, {"type": "text", "text": "...", "id": "msg_abc123"}, ], response_metadata={"model_provider": "openai"} ) message.content_blocks Copy[{'type': 'reasoning', 'id': 'rs_abc123', 'reasoning': 'summary 1'}, {'type': 'reasoning', 'id': 'rs_abc123', 'reasoning': 'summary 2'}, {'type': 'text', 'text': '...', 'id': 'msg_abc123'}] See the integrations guides to get started with the inference provider of your choice. Serializing standard contentIf an application outside of LangChain needs access to the standard content block representation, you can opt-in to storing content blocks in message content.To do this, you can set the LC_OUTPUT_VERSION environment variable to v1. Or, initialize any chat model with output_version="v1":Copyfrom langchain.chat_models import init_chat_model model = init_chat_model("gpt-5-nano", output_version="v1") ​Multimodal Multimodality refers to the ability to work with data that comes in different forms, such as text, audio, images, and video. LangChain includes standard types for these data that can be used across providers. Chat models can accept multimodal data as input and generate it as output. Below we show short examples of input messages featuring multimodal data. Extra keys can be included top-level in the content block or nested in "extras": {"key": value}.OpenAI and AWS Bedrock Converse, for example, require a filename for PDFs. See the provider page for your chosen model for specifics. Image inputPDF document inputAudio inputVideo inputCopy# From URL message = { "role": "user", "content": [ {"type": "text", "text": "Describe the content of this image."}, {"type": "image", "url": "https://example.com/path/to/image.jpg"}, ] } # From base64 data message = { "role": "user", "content": [ {"type": "text", "text": "Describe the content of this image."}, { "type": "image", "base64": "AAAAIGZ0eXBtcDQyAAAAAGlzb21tcDQyAAACAGlzb2...", "mime_type": "image/jpeg", }, ] } # From provider-managed File ID message = { "role": "user", "content": [ {"type": "text", "text": "Describe the content of this image."}, {"type": "image", "file_id": "file-abc123"}, ] } Not all models support all file types. Check the model provider’s reference for supported formats and size limits. ​Content block reference Content blocks are represented (either when creating a message or accessing the content_blocks property) as a list of typed dictionaries. Each item in the list must adhere to one of the following block types: CoreTextContentBlockPurpose: Standard text output​typestringrequiredAlways "text"​textstringrequiredThe text content​annotationsobject[]List of annotations for the text​extrasobjectAdditional provider-specific dataExample:Copy{ "type": "text", "text": "Hello world", "annotations": [] } ReasoningContentBlockPurpose: Model reasoning steps​typestringrequiredAlways "reasoning"​reasoningstringThe reasoning content​extrasobjectAdditional provider-specific dataExample:Copy{ "type": "reasoning", "reasoning": "The user is asking about...", "extras": {"signature": "abc123"}, } MultimodalImageContentBlockPurpose: Image data​typestringrequiredAlways "image"​urlstringURL pointing to the image location.​base64stringBase64-encoded image data.​idstringUnique identifier for this content block (either generated by the provider or by LangChain).​mime_typestringImage MIME type (e.g., image/jpeg, image/png). Required for base64 data.AudioContentBlockPurpose: Audio data​typestringrequiredAlways "audio"​urlstringURL pointing to the audio location.​base64stringBase64-encoded audio data.​idstringUnique identifier for this content block (either generated by the provider or by LangChain).​mime_typestringAudio MIME type (e.g., audio/mpeg, audio/wav). Required for base64 data.VideoContentBlockPurpose: Video data​typestringrequiredAlways "video"​urlstringURL pointing to the video location.​base64stringBase64-encoded video data.​idstringUnique identifier for this content block (either generated by the provider or by LangChain).​mime_typestringVideo MIME type (e.g., video/mp4, video/webm). Required for base64 data.FileContentBlockPurpose: Generic files (PDF, etc)​typestringrequiredAlways "file"​urlstringURL pointing to the file location.​base64stringBase64-encoded file data.​idstringUnique identifier for this content block (either generated by the provider or by LangChain).​mime_typestringFile MIME type (e.g., application/pdf). Required for base64 data.PlainTextContentBlockPurpose: Document text (.txt, .md)​typestringrequiredAlways "text-plain"​textstringThe text content​mime_typestringMIME type of the text (e.g., text/plain, text/markdown)Tool CallingToolCallPurpose: Function calls​typestringrequiredAlways "tool_call"​namestringrequiredName of the tool to call​argsobjectrequiredArguments to pass to the tool​idstringrequiredUnique identifier for this tool callExample:Copy{ "type": "tool_call", "name": "search", "args": {"query": "weather"}, "id": "call_123" } ToolCallChunkPurpose: Streaming tool call fragments​typestringrequiredAlways "tool_call_chunk"​namestringName of the tool being called​argsstringPartial tool arguments (may be incomplete JSON)​idstringTool call identifier​indexnumber | stringPosition of this chunk in the streamInvalidToolCallPurpose: Malformed calls, intended to catch JSON parsing errors.​typestringrequiredAlways "invalid_tool_call"​namestringName of the tool that failed to be called​argsobjectArguments to pass to the tool​errorstringDescription of what went wrongServer-Side Tool ExecutionServerToolCallPurpose: Tool call that is executed server-side.​typestringrequiredAlways "server_tool_call"​idstringrequiredAn identifier associated with the tool call.​namestringrequiredThe name of the tool to be called.​argsstringrequiredPartial tool arguments (may be incomplete JSON)ServerToolCallChunkPurpose: Streaming server-side tool call fragments​typestringrequiredAlways "server_tool_call_chunk"​idstringAn identifier associated with the tool call.​namestringName of the tool being called​argsstringPartial tool arguments (may be incomplete JSON)​indexnumber | stringPosition of this chunk in the streamServerToolResultPurpose: Search results​typestringrequiredAlways "server_tool_result"​tool_call_idstringrequiredIdentifier of the corresponding server tool call.​idstringIdentifier associated with the server tool result.​statusstringrequiredExecution status of the server-side tool. "success" or "error".​outputOutput of the executed tool.Provider-Specific BlocksNonStandardContentBlockPurpose: Provider-specific escape hatch​typestringrequiredAlways "non_standard"​valueobjectrequiredProvider-specific data structureUsage: For experimental or provider-unique featuresAdditional provider-specific content types may be found within the reference documentation of each model provider. View the canonical type definitions in the API reference. Content blocks were introduced as a new property on messages in LangChain v1 to standardize content formats across providers while maintaining backward compatibility with existing code.Content blocks are not a replacement for the content property, but rather a new property that can be used to access the content of a message in a standardized format. ​Use with chat models Chat models accept a sequence of message objects as input and return an AIMessage as output. Interactions are often stateless, so that a simple conversational loop involves invoking a model with a growing list of messages. Refer to the below guides to learn more: Built-in features for persisting and managing conversation histories Strategies for managing context windows, including trimming and summarizing messages Edit this page on GitHub or file an issue. Connect these docs to Claude, VSCode, and more via MCP for real-time answers.Was this page helpful?YesNoModelsPreviousToolsNext⌘IDocs by LangChain home pagegithubxlinkedinyoutubeResourcesForumChangelogLangChain AcademyTrust CenterCompanyAboutCareersBloggithubxlinkedinyoutubePowered by

Copysystem

Example Code Patterns

Example 1 (markdown):

Copypip install -U langchain
# Requires Python 3.10+

Example 2 (markdown):

Copy# Installing the OpenAI integration
pip install -U langchain-openai

# Installing the Anthropic integration
pip install -U langchain-anthropic

Example 3 (python):

Copydef create_agent(
    ...
    response_format: Union[
        ToolStrategy[StructuredResponseT],
        ProviderStrategy[StructuredResponseT],
        type[StructuredResponseT],
        None,
    ]

Example 4 (json):

Copycustom_profile = {
    "structured_output": True,
    # ...
}
model = init_chat_model("...", profile=custom_profile)

Reference Files

This skill includes comprehensive documentation in references/:

• chains.md - Chains documentation
• getting_started.md - Getting Started documentation

Use view to read specific reference files when detailed information is needed.

Working with This Skill

For Beginners

Start with the getting_started or tutorials reference files for foundational concepts.

For Specific Features

Use the appropriate category reference file (api, guides, etc.) for detailed information.

For Code Examples

The quick reference section above contains common patterns extracted from the official docs.

Resources

references/

Organized documentation extracted from official sources. These files contain:

• Detailed explanations
• Code examples with language annotations
• Links to original documentation
• Table of contents for quick navigation

scripts/

Add helper scripts here for common automation tasks.

assets/

Add templates, boilerplate, or example projects here.

Notes

• This skill was automatically generated from official documentation
• Reference files preserve the structure and examples from source docs
• Code examples include language detection for better syntax highlighting
• Quick reference patterns are extracted from common usage examples in the docs

Updating

To refresh this skill with updated documentation:

1. Re-run the scraper with the same configuration
2. The skill will be rebuilt with the latest information