流式输出

预览功能： 尝试对消息、工具调用、状态和最终输出进行事件流式输出的类型化智能体投影。从 LangChain 事件流式输出开始，或探索流式输出 cookbook 中的可运行示例。

LangChain 实现了一个流式输出系统来展示实时更新。流式输出对于提升基于 LLM 构建的应用程序的响应性至关重要。通过逐步显示输出，即使在完整响应准备好之前，流式输出也能显著改善用户体验（UX），特别是在处理 LLM 的延迟时。

概述

LangChain 的流式输出系统让你可以将智能体运行中的实时反馈展示给你的应用程序。 LangChain 流式输出的功能：

流式输出智能体进度 — 在每个智能体步骤后获取状态更新。
流式输出 LLM Token — 在语言模型生成 Token 时进行流式输出。
流式输出思考/推理 Token — 在模型推理生成时展示推理过程。
流式输出自定义更新 — 发出用户定义的信号（例如 "Fetched 10/100 records"）。
流式输出多种模式 — 从 updates（智能体进度）、messages（LLM Token + 元数据）或 custom（任意用户数据）中选择。

有关更多端到端示例，请参阅下面的常见模式部分。

支持的流式模式

将以下一个或多个流式模式作为列表传递给 stream 或 astream 方法：

模式	描述
`updates`	在每个智能体步骤后流式输出状态更新。如果在同一步骤中进行了多次更新（例如运行了多个节点），这些更新会分别流式输出。
`messages`	从调用了 LLM 的任何图节点中流式输出 `(token, metadata)` 元组。
`custom`	使用 stream writer 从图节点内部流式输出自定义数据。

智能体进度

要流式输出智能体进度，使用 stream 或 astream 方法并设置 stream_mode="updates"。这会在每个智能体步骤后发出一个事件。例如，如果你有一个调用了一次工具的智能体，你应该看到以下更新：

LLM 节点：包含工具调用请求的 AIMessage
工具节点：包含执行结果的 ToolMessage
LLM 节点：最终 AI 响应

流式输出智能体进度

from 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",
    version="v2",
):
    if chunk["type"] == "updates":
        for step, data in chunk["data"].items():
            print(f"step: {step}")
            print(f"content: {data['messages'][-1].content_blocks}")

输出

step: 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 Token

要在 LLM 生成 Token 时进行流式输出，使用 stream_mode="messages"。下面你可以看到智能体流式输出工具调用和最终响应的输出。

流式输出 LLM Token

from 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="messages",
    version="v2",
):
    if chunk["type"] == "messages":
        token, metadata = chunk["data"]
        print(f"node: {metadata['langgraph_node']}")
        print(f"content: {token.content_blocks}")
        print("\n")

自定义更新

要在工具执行期间流式输出更新，可以使用 get_stream_writer。

流式输出自定义更新

from 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="claude-sonnet-4-6",
    tools=[get_weather],
)

for chunk in agent.stream(
    {"messages": [{"role": "user", "content": "What is the weather in SF?"}]},
    stream_mode="custom",
    version="v2",
):
    if chunk["type"] == "custom":
        print(chunk["data"])

输出

Looking up data for city: San Francisco
Acquired data for city: San Francisco

如果你在工具内添加了 get_stream_writer，你将无法在 LangGraph 执行上下文之外调用该工具。

流式输出多种模式

你可以通过将流式模式作为列表传递来指定多种流式模式：stream_mode=["updates", "custom"]。每个流式块是一个包含 type、ns 和 data 键的 StreamPart 字典。使用 chunk["type"] 来确定流式模式，使用 chunk["data"] 来访问负载。

常见模式

以下是展示流式输出常见用例的示例。

流式输出思考/推理 Token

一些模型在产生最终答案之前会执行内部推理。你可以通过过滤标准内容块中 type 为 "reasoning" 的内容来流式输出这些思考/推理 Token。

必须在模型上启用推理输出。有关配置详情，请参阅推理部分和你的提供商集成页面。要快速检查模型的推理支持，请参阅 models.dev。

要从智能体流式输出思考 Token，使用 stream_mode="messages" 并过滤推理内容块：

from langchain.agents import create_agent
from langchain.messages import AIMessageChunk
from langchain_anthropic import ChatAnthropic
from langchain_core.runnables import Runnable


def get_weather(city: str) -> str:
    """Get weather for a given city."""
    return f"It's always sunny in {city}!"


model = ChatAnthropic(
    model_name="claude-sonnet-4-6",
    timeout=None,
    stop=None,
    thinking={"type": "enabled", "budget_tokens": 5000},
)
agent: Runnable = create_agent(
    model=model,
    tools=[get_weather],
)

for token, metadata in agent.stream(
    {"messages": [{"role": "user", "content": "What is the weather in SF?"}]},
    stream_mode="messages",
):
    if not isinstance(token, AIMessageChunk):
        continue
    reasoning = [b for b in token.content_blocks if b["type"] == "reasoning"]
    text = [b for b in token.content_blocks if b["type"] == "text"]
    if reasoning:
        print(f"[thinking] {reasoning[0]['reasoning']}", end="")
    if text:
        print(text[0]["text"], end="")

输出

[thinking] The user is asking about the weather in San Francisco. I have a tool
[thinking]  available to get this information. Let me call the get_weather tool
[thinking]  with "San Francisco" as the city parameter.
The weather in San Francisco is: It's always sunny in San Francisco!

无论模型提供商如何，这都以相同方式工作——LangChain 通过 content_blocks 属性将提供商特定的格式（Anthropic thinking 块、OpenAI reasoning 摘要等）规范化为标准的 "reasoning" 内容块类型。要直接从聊天模型流式输出推理 Token（不使用智能体），请参阅聊天模型的流式输出。

流式输出工具调用

你可能希望同时流式输出：

工具调用生成时的部分 JSON
被执行的已完成、已解析的工具调用

指定 stream_mode="messages" 将流式输出智能体中所有 LLM 调用生成的增量消息块。要访问包含已解析工具调用的完整消息：

如果这些消息被跟踪在状态中（如 create_agent 的 model 节点），使用 stream_mode=["messages", "updates"] 通过状态更新访问完整消息（如下演示）。
如果这些消息未被跟踪在状态中，使用自定义更新或在流式循环中聚合块（下一节）。

带人机协作的流式输出

要处理人机协作中断，我们在上面的示例基础上构建：

我们使用人机协作中间件和检查点配置智能体
我们收集在 "updates" 流式模式期间生成的中断
我们使用命令响应这些中断

从子智能体流式输出

当智能体中的任何点有多个 LLM 时，通常需要消除消息生成来源的歧义。为此，在创建每个智能体时传递一个 name。在 "messages" 模式下流式输出时，此名称可通过 lc_agent_name 键在元数据中获取。

当你在智能体上设置 name 时，该名称也会附加到该智能体生成的任何 AIMessage 上。

禁用流式输出

在某些应用程序中，你可能需要为给定模型禁用单个 Token 的流式输出。这在以下场景中很有用：

使用多智能体系统来控制哪些智能体流式输出其输出
混合使用支持流式输出和不支持流式输出的模型
部署到 LangSmith 并希望防止某些模型输出被流式传输到客户端

在初始化模型时设置 streaming=False。

from langchain_openai import ChatOpenAI

model = ChatOpenAI(
    model="gpt-5.4",
    streaming=False
)

部署到 LangSmith 时，在你不希望将其输出流式传输到客户端的任何模型上设置 streaming=False。这在部署前在你的图代码中配置。

并非所有聊天模型集成都支持 streaming 参数。如果你的模型不支持它，请使用 disable_streaming=True 代替。此参数通过基类在所有聊天模型上可用。

有关更多详情，请参阅 LangGraph 流式输出指南。

v2 流式输出格式

需要 LangGraph >= 1.1。

向 stream() 或 astream() 传递 version="v2" 以获取统一的输出格式。每个块是一个包含 type、ns 和 data 键的 StreamPart 字典——无论流式模式或模式数量如何，形状都相同：

# 统一格式——不再需要元组解包
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" 或 "custom"
    print(chunk["data"])  # 负载

# 必须解包 (mode, data) 元组
for mode, chunk in agent.stream(
    {"messages": [{"role": "user", "content": "What is the weather in SF?"}]},
    stream_mode=["updates", "custom"],
):
    print(mode)   # "updates" 或 "custom"
    print(chunk)  # 负载

v2 格式还改进了 invoke() —— 它返回一个具有 .value 和 .interrupts 属性的 GraphOutput 对象，将状态与中断元数据清晰分离：

result = agent.invoke(
    {"messages": [{"role": "user", "content": "Hello"}]},
    version="v2",
)
print(result.value)       # 状态（dict、Pydantic 模型或 dataclass）
print(result.interrupts)  # Interrupt 对象的元组（如果没有则为空）

有关 v2 格式的更多详情，包括类型缩窄、Pydantic/dataclass 转换和子图流式输出，请参阅 LangGraph 流式输出文档。

Get started

Core components

Middleware

Frontend

Advanced usage

Agent development

Deploy with LangSmith

概述

支持的流式模式

智能体进度

LLM Token

自定义更新

流式输出多种模式

常见模式

流式输出思考/推理 Token

流式输出工具调用

带人机协作的流式输出

从子智能体流式输出

禁用流式输出

v2 流式输出格式

相关内容

​概述

​支持的流式模式

​智能体进度

​LLM Token

​自定义更新

​流式输出多种模式

​常见模式

​流式输出思考/推理 Token

​流式输出工具调用

​带人机协作的流式输出

​从子智能体流式输出

​禁用流式输出

​v2 流式输出格式

​相关内容

概述

支持的流式模式

智能体进度

LLM Token

自定义更新

流式输出多种模式

常见模式

流式输出思考/推理 Token

流式输出工具调用

带人机协作的流式输出

从子智能体流式输出

禁用流式输出

v2 流式输出格式

相关内容