🎉 学 LangChain 免费领 openAI GPT key 限额1000名 →
新版文档(v1.x)
langchain
Streaming

Streaming

import StreamingAgentProgressJs from '/snippets/code-samples/streaming-agent-progress-js.mdx'; import StreamingAgentProgressPy from '/snippets/code-samples/streaming-agent-progress-py.mdx'; import StreamingReasoningTokensPy from '/snippets/code-samples/streaming-reasoning-tokens-py.mdx'; import StreamingReasoningTokensJs from '/snippets/code-samples/streaming-reasoning-tokens-js.mdx';

For new applications, we recommend event streaming (opens in a new tab)—the typed-projection API introduced in LangChain v1.3. Event streaming gives you separate iterators per projection (messages, values, tool calls, subgraphs) so you can consume them independently instead of branching on stream_mode chunks.

LangChain 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:

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][CompiledStateGraph.stream] or @[astream][CompiledStateGraph.astream] methods:

ModeDescription
updatesStreams 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][CompiledStateGraph.stream] or @[astream][CompiledStateGraph.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

Pass a thread_id via config so the conversation is checkpointed and follow-up turns can resume the same history. thread_id is independent of stream_mode; you can also pass context alongside it for per-run data your tools read from runtime.context.

Output

Persisting conversation history with thread_id requires the agent to be configured with a checkpointer (opens in a new tab). On LangSmith deployments a checkpointer is provisioned automatically. Locally, pass one explicitly, for example create_agent(..., checkpointer=InMemorySaver()). The remaining snippets on this page omit thread_id for brevity, but you should pass it in production.

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.

Output
**Wrapping an agent as a node in a parent `StateGraph`?** @[`create_agent`] returns a compiled graph, so using it as a node makes it a subgraph. `stream_mode="messages"` on the parent graph will not emit token chunks from the inner agent's LLM calls unless you pass `subgraphs=True`. See [Subgraph outputs](https://docs.langchain.com/oss/langgraph/streaming#subgraph-outputs).

Custom updates

To stream updates from tools as they are executed, you can use @[get_stream_writer].

Output
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"].

Each streamed chunk is a StreamPart dict with type, ns, and data keys. Use chunk["type"] to determine the stream mode and chunk["data"] to access the payload.

Output

Common patterns

Below are examples showing common use cases for streaming.

Streaming thinking / reasoning tokens

Some models perform internal reasoning before producing a final answer. You can stream these thinking / reasoning tokens as they're generated by filtering standard content blocks (opens in a new tab) for the type "reasoning".

Reasoning output must be enabled on the model.

See the [reasoning section](https://docs.langchain.com/oss/langchain/models#reasoning) and your [provider's integration page](https://docs.langchain.com/oss/integrations/providers/overview) for configuration details.

To quickly check a model's reasoning support, see [models.dev](https://models.dev).

To stream thinking tokens from an agent, use stream_mode="messages" and filter for reasoning content blocks:

Output

This works the same way regardless of the model provider—LangChain normalizes provider-specific formats (Anthropic thinking blocks, OpenAI reasoning summaries, etc.) into a standard "reasoning" content block type via the content_blocks (opens in a new tab) property.

To stream reasoning tokens directly from a chat model (without an agent), see streaming with chat models (opens in a new tab).

Streaming tool calls

You may want to stream both:

  1. Partial JSON as tool calls (opens in a new tab) are generated
  2. The completed, parsed tool calls that are executed

Specifying stream_mode="messages" will stream incremental message chunks (opens in a new tab) generated by all LLM calls in the agent. To access the completed messages with parsed tool calls:

  1. If those messages are tracked in the state (opens in a new tab) (as in the model node of create_agent (opens in a new tab)), use stream_mode=["messages", "updates"] to access completed messages through state updates (demonstrated below).
  2. 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.

from 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.5", 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 chunk in agent.stream(
    {"messages": [input_message]},
    stream_mode=["messages", "updates"],  # [!code highlight]
    version="v2",  # [!code highlight]
):
    if chunk["type"] == "messages":  # [!code highlight]
        token, metadata = chunk["data"]  # [!code highlight]
        if isinstance(token, AIMessageChunk):
            _render_message_chunk(token)  # [!code highlight]
    elif chunk["type"] == "updates":  # [!code highlight]
        for source, update in chunk["data"].items():  # [!code highlight]
            if source in ("model", "tools"):  # `source` captures node name
                _render_completed_message(update["messages"][-1])  # [!code highlight]
Output

Accessing completed messages

If completed messages are tracked in an agent's state (opens in a new tab), you can use stream_mode=["messages", "updates"] as demonstrated in the Streaming tool calls section 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 (opens in a new tab). This middleware demonstrates tool calling to generate a structured "safe / unsafe" evaluation (one could also use structured outputs (opens in a new tab) for this):

from 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  # [!code highlight]
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.5")
 
@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()  # [!code highlight]
    # 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)  # [!code highlight]
 
    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:

from 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.5",
    tools=[get_weather],
    middleware=[safety_guardrail],  # [!code highlight]
)
 
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 chunk in agent.stream(
    {"messages": [input_message]},
    stream_mode=["messages", "updates", "custom"],  # [!code highlight]
    version="v2",  # [!code highlight]
):
    if chunk["type"] == "messages":  # [!code highlight]
        token, metadata = chunk["data"]  # [!code highlight]
        if isinstance(token, AIMessageChunk):
            _render_message_chunk(token)
    elif chunk["type"] == "updates":  # [!code highlight]
        for source, update in chunk["data"].items():  # [!code highlight]
            if source in ("model", "tools"):
                _render_completed_message(update["messages"][-1])
    elif chunk["type"] == "custom":  # [!code highlight]
        # access completed message in stream
        print(f"Tool calls: {chunk['data'].tool_calls}")  # [!code highlight]
Output

Alternatively, if you aren't able to add custom events to the stream, you can aggregate message chunks within the streaming loop:

input_message = {"role": "user", "content": "What is the weather in Boston?"}
full_message = None  # [!code highlight]
for chunk in agent.stream(
    {"messages": [input_message]},
    stream_mode=["messages", "updates"],
    version="v2",  # [!code highlight]
):
    if chunk["type"] == "messages":  # [!code highlight]
        token, metadata = chunk["data"]  # [!code highlight]
        if isinstance(token, AIMessageChunk):
            _render_message_chunk(token)
            full_message = token if full_message is None else full_message + token  # [!code highlight]
            if token.chunk_position == "last":  # [!code highlight]
                if full_message.tool_calls:  # [!code highlight]
                    print(f"Tool calls: {full_message.tool_calls}")  # [!code highlight]
                full_message = None  # [!code highlight]
    elif chunk["type"] == "updates":  # [!code highlight]
        for source, update in chunk["data"].items():  # [!code highlight]
            if source == "tools":
                _render_completed_message(update["messages"][-1])

Streaming with human-in-the-loop

To handle human-in-the-loop interrupts (opens in a new tab), we build on the above example:

  1. We configure the agent with human-in-the-loop middleware and a checkpointer (opens in a new tab)
  2. We collect interrupts generated during the "updates" stream mode
  3. We respond to those interrupts with a command (opens in a new tab)
from 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.5",
    tools=[get_weather],
    middleware=[  # [!code highlight]
        HumanInTheLoopMiddleware(interrupt_on={"get_weather": True}),  # [!code highlight]
    ],  # [!code highlight]
    checkpointer=checkpointer,  # [!code highlight]
)
 
 
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:  # [!code highlight]
    interrupts = interrupt.value  # [!code highlight]
    for request in interrupts["action_requests"]:  # [!code highlight]
        print(request["description"])  # [!code highlight]
 
 
input_message = {
    "role": "user",
    "content": (
        "Can you look up the weather in Boston and San Francisco?"
    ),
}
config = {"configurable": {"thread_id": "some_id"}}  # [!code highlight]
interrupts = []  # [!code highlight]
for chunk in agent.stream(
    {"messages": [input_message]},
    config=config,  # [!code highlight]
    stream_mode=["messages", "updates"],
    version="v2",  # [!code highlight]
):
    if chunk["type"] == "messages":  # [!code highlight]
        token, metadata = chunk["data"]  # [!code highlight]
        if isinstance(token, AIMessageChunk):
            _render_message_chunk(token)
    elif chunk["type"] == "updates":  # [!code highlight]
        for source, update in chunk["data"].items():  # [!code highlight]
            if source in ("model", "tools"):
                _render_completed_message(update["messages"][-1])
            if source == "__interrupt__":  # [!code highlight]
                interrupts.extend(update)  # [!code highlight]
                _render_interrupt(update[0])  # [!code highlight]
Output

We next collect a decision (opens in a new tab) 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:

def _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
Output

We can then resume by passing a command (opens in a new tab) into the same streaming loop:

interrupts = []
for chunk in agent.stream(
    Command(resume=decisions),  # [!code highlight]
    config=config,
    stream_mode=["messages", "updates"],
    version="v2",  # [!code highlight]
):
    # Streaming loop is unchanged
    if chunk["type"] == "messages":  # [!code highlight]
        token, metadata = chunk["data"]  # [!code highlight]
        if isinstance(token, AIMessageChunk):
            _render_message_chunk(token)
    elif chunk["type"] == "updates":  # [!code highlight]
        for source, update in chunk["data"].items():  # [!code highlight]
            if source in ("model", "tools"):
                _render_completed_message(update["messages"][-1])
            if source == "__interrupt__":
                interrupts.extend(update)
                _render_interrupt(update[0])
Output

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 (opens in a new tab) 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:

  1. We replace our tool with a call_weather_agent tool that invokes an agent internally
  2. We add a name to each agent
  3. We specify subgraphs=True (opens in a new tab) when creating the stream
  4. 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:

from 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.5")
weather_agent = create_agent(
    model=weather_model,
    tools=[get_weather],
    name="weather_agent",  # [!code highlight]
)
 
 
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.5")
agent = create_agent(
    model=supervisor_model,
    tools=[call_weather_agent],
    name="supervisor",  # [!code highlight]
)

Next, we add logic to the streaming loop to report which agent is emitting tokens:

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?"}
current_agent = None  # [!code highlight]
for chunk in agent.stream(
    {"messages": [input_message]},
    stream_mode=["messages", "updates"],
    subgraphs=True,  # [!code highlight]
    version="v2",  # [!code highlight]
):
    if chunk["type"] == "messages":  # [!code highlight]
        token, metadata = chunk["data"]  # [!code highlight]
        if agent_name := metadata.get("lc_agent_name"):  # [!code highlight]
            if agent_name != current_agent:  # [!code highlight]
                print(f"🤖 {agent_name}: ")  # [!code highlight]
                current_agent = agent_name  # [!code highlight]
        if isinstance(token, AIMessageChunk):
            _render_message_chunk(token)
    elif chunk["type"] == "updates":  # [!code highlight]
        for source, update in chunk["data"].items():  # [!code highlight]
            if source in ("model", "tools"):
                _render_completed_message(update["messages"][-1])
Output

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 (opens in a new tab) 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.

from langchain_openai import ChatOpenAI
 
model = ChatOpenAI(
    model="gpt-5.5",
    streaming=False  # [!code highlight]
)

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 (opens in a new tab) for more details.

v2 streaming format

Requires LangGraph 1.1 or later.

Pass version="v2" to stream() or astream() to get a unified output format. Every chunk is a StreamPart dict with type, ns, and data keys — the same shape regardless of stream mode or number of modes:

# Unified format — no more tuple unpacking
for chunk in agent.stream(
{"messages": [{"role": "user", "content": "What is the weather in SF?"}]},
stream_mode=["updates", "custom"],
version="v2",
):
print(chunk["type"])  # "updates" or "custom"
print(chunk["data"])  # payload

v1 (current default)

# Must unpack (mode, data) tuples
for mode, chunk in agent.stream(
{"messages": [{"role": "user", "content": "What is the weather in SF?"}]},
stream_mode=["updates", "custom"],
):
print(mode)   # "updates" or "custom"
print(chunk)  # payload

The v2 format also improves invoke() — it returns a GraphOutput object with .value and .interrupts attributes, cleanly separating state from interrupt metadata:

result = agent.invoke(
    {"messages": [{"role": "user", "content": "Hello"}]},
    version="v2",
)
print(result.value)       # state (dict, Pydantic model, or dataclass)
print(result.interrupts)  # tuple of Interrupt objects (empty if none)

See the LangGraph streaming docs (opens in a new tab) for more details on the v2 format, including type narrowing, Pydantic/dataclass coercion, and subgraph streaming.

Related

Build a SQL agentStructured output