> ## Documentation Index
> Fetch the complete documentation index at: https://cndoc-langchain.site/llms.txt
> Use this file to discover all available pages before exploring further.
# 流式处理
> 从代理运行中流式传输实时更新
LangChain 实现了一个流式处理系统，用于呈现实时更新。
流式处理对于增强基于大语言模型构建的应用程序的响应能力至关重要。通过渐进式地显示输出，甚至在完整响应准备好之前，流式处理显著提升了用户体验（UX），尤其是在处理大语言模型的延迟时。
## 概述
LangChain 的流式处理系统允许你将代理运行的实时反馈呈现到你的应用程序中。
使用 LangChain 流式处理可以实现：
* [**流式传输代理进度**](#代理进度)—在每个代理步骤后获取状态更新。
* [**流式传输大语言模型令牌**](#大语言模型令牌)—在生成时流式传输语言模型令牌。
* [**流式传输思考/推理令牌**](#流式传输思考/-推理令牌)—在生成时呈现模型推理过程。
* [**流式传输自定义更新**](#自定义更新)—发出用户定义的信号（例如，`"已获取 10/100 条记录"`）。
* [**流式传输多种模式**](#流式传输多种模式)—从 `updates`（代理进度）、`messages`（大语言模型令牌 + 元数据）或 `custom`（任意用户数据）中选择。
有关更多端到端示例，请参阅下面的[常见模式](#常见模式)部分。
## 支持的流式模式
将以下一种或多种流式模式作为列表传递给 [`stream`](https://reference.langchain.com/python/langgraph/graphs/#langgraph.graph.state.CompiledStateGraph.stream) 或 [`astream`](https://reference.langchain.com/python/langgraph/graphs/#langgraph.graph.state.CompiledStateGraph.astream) 方法：
| 模式 | 描述 |
| ---------- | --------------------------------------------------------- |
| `updates` | 在每个代理步骤后流式传输状态更新。如果在同一步骤中进行了多次更新（例如，运行了多个节点），这些更新将分别流式传输。 |
| `messages` | 从任何调用了大语言模型的图节点流式传输 `(token, metadata)` 元组。 |
| `custom` | 使用流式写入器从图节点内部流式传输自定义数据。 |
## 代理进度
要流式传输代理进度，请使用 [`stream`](https://reference.langchain.com/python/langgraph/graphs/#langgraph.graph.state.CompiledStateGraph.stream) 或 [`astream`](https://reference.langchain.com/python/langgraph/graphs/#langgraph.graph.state.CompiledStateGraph.astream) 方法，并设置 `stream_mode="updates"`。这会在每个代理步骤后发出一个事件。
例如，如果你有一个调用一次工具的代理，你应该看到以下更新：
* **大语言模型节点**：包含工具调用请求的 [`AIMessage`](https://reference.langchain.com/python/langchain-core/messages/ai/AIMessage)
* **工具节点**：包含执行结果的 [`ToolMessage`](https://reference.langchain.com/python/langchain-core/messages/tool/ToolMessage)
* **大语言模型节点**：最终的 AI 响应
```python title="流式传输代理进度" theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
from langchain.agents import create_agent
def get_weather(city: str) -> str:
"""获取给定城市的天气。"""
return f"It's always sunny in {city}!"
agent = create_agent(
model="gpt-5-nano",
tools=[get_weather],
)
for chunk in agent.stream( # [!code highlight]
{"messages": [{"role": "user", "content": "What is the weather in SF?"}]},
stream_mode="updates",
version="v2", # [!code highlight]
):
if chunk["type"] == "updates": # [!code highlight]
for step, data in chunk["data"].items(): # [!code highlight]
print(f"step: {step}")
print(f"content: {data['messages'][-1].content_blocks}")
```
```shell title="输出" theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
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!'}]
```
## 大语言模型令牌
要流式传输大语言模型生成的令牌，请使用 `stream_mode="messages"`。下面你可以看到代理流式传输工具调用和最终响应的输出。
```python title="流式传输大语言模型令牌" theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
from langchain.agents import create_agent
def get_weather(city: str) -> str:
"""获取给定城市的天气。"""
return f"It's always sunny in {city}!"
agent = create_agent(
model="gpt-5-nano",
tools=[get_weather],
)
for chunk in agent.stream( # [!code highlight]
{"messages": [{"role": "user", "content": "What is the weather in SF?"}]},
stream_mode="messages",
version="v2", # [!code highlight]
):
if chunk["type"] == "messages": # [!code highlight]
token, metadata = chunk["data"] # [!code highlight]
print(f"node: {metadata['langgraph_node']}")
print(f"content: {token.content_blocks}")
print("\n")
```
```shell title="输出" expandable theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
node: 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'}]
```
## 自定义更新
要在工具执行时流式传输更新，你可以使用 [`get_stream_writer`](https://reference.langchain.com/python/langgraph/config/get_stream_writer)。
```python title="流式传输自定义更新" theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
from langchain.agents import create_agent
from langgraph.config import get_stream_writer # [!code highlight]
def get_weather(city: str) -> str:
"""获取给定城市的天气。"""
writer = get_stream_writer() # [!code highlight]
# 流式传输任意数据
writer(f"正在查找城市数据: {city}")
writer(f"已获取城市数据: {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", # [!code highlight]
version="v2", # [!code highlight]
):
if chunk["type"] == "custom": # [!code highlight]
print(chunk["data"]) # [!code highlight]
```
```shell title="输出" theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
正在查找城市数据: San Francisco
已获取城市数据: San Francisco
```
如果你在工具内部添加了 [`get_stream_writer`](https://reference.langchain.com/python/langgraph/config/get_stream_writer)，你将无法在 LangGraph 执行上下文之外调用该工具。
## 流式传输多种模式
你可以通过将流式模式作为列表传递来指定多种流式模式：`stream_mode=["updates", "custom"]`。
每个流式传输的块都是一个 `StreamPart` 字典，包含 `type`、`ns` 和 `data` 键。使用 `chunk["type"]` 来确定流式模式，使用 `chunk["data"]` 来访问有效负载。
```python title="流式传输多种模式" theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
from langchain.agents import create_agent
from langgraph.config import get_stream_writer
def get_weather(city: str) -> str:
"""获取给定城市的天气。"""
writer = get_stream_writer()
writer(f"正在查找城市数据: {city}")
writer(f"已获取城市数据: {city}")
return f"It's always sunny in {city}!"
agent = create_agent(
model="gpt-5-nano",
tools=[get_weather],
)
for chunk in agent.stream( # [!code highlight]
{"messages": [{"role": "user", "content": "What is the weather in SF?"}]},
stream_mode=["updates", "custom"],
version="v2", # [!code highlight]
):
print(f"stream_mode: {chunk['type']}") # [!code highlight]
print(f"content: {chunk['data']}") # [!code highlight]
print("\n")
```
```shell title="输出" theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
stream_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: 正在查找城市数据: San Francisco
stream_mode: custom
content: 已获取城市数据: 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}})]}}
```
## 常见模式
以下是展示流式处理常见用例的示例。
### 流式传输思考/推理令牌
一些模型在生成最终答案之前会执行内部推理。你可以通过过滤 [标准内容块](/oss/python/langchain/messages#standard-content-blocks) 中 `type` 为 `"reasoning"` 的内容，在生成时流式传输这些思考/推理令牌。
必须在模型上启用推理输出。
有关配置详情，请参阅[推理部分](/oss/python/langchain/models#reasoning)和你的[提供商集成页面](/oss/python/integrations/providers/overview)。
要快速检查模型的推理支持情况，请参阅 [models.dev](https://models.dev)。
要从代理流式传输思考令牌，请使用 `stream_mode="messages"` 并过滤推理内容块：
```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
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:
"""获取给定城市的天气。"""
return f"{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": "旧金山的天气怎么样？"}]},
stream_mode="messages", # [!code highlight]
):
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"[思考] {reasoning[0]['reasoning']}", end="")
if text:
print(text[0]["text"], end="")
```
```shell title="输出" theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
[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`](/oss/python/langchain/messages#standard-content-blocks) 属性将提供商特定的格式（Anthropic `thinking` 块、OpenAI `reasoning` 摘要等）标准化为标准的 `"reasoning"` 内容块类型。
要直接从聊天模型（不使用代理）流式传输推理令牌，请参阅[使用聊天模型进行流式传输](/oss/python/langchain/models#reasoning)。
### 流式传输工具调用
你可能希望同时流式传输：
1. 在生成 [工具调用](/oss/python/langchain/models#tool-calling) 时的部分 JSON
2. 已执行的、已解析的工具调用
指定 [`stream_mode="messages"`](#大语言模型令牌) 将流式传输代理中所有大语言模型调用生成的增量[消息块](/oss/python/langchain/messages#streaming-and-chunks)。要访问包含已解析工具调用的完整消息：
1. 如果这些消息在[状态](/oss/python/langchain/agents#memory)中被跟踪（如 [`create_agent`](/oss/python/langchain/agents) 的模型节点中），请使用 `stream_mode=["messages", "updates"]` 通过[状态更新](#代理进度)访问完整消息（如下所示）。
2. 如果这些消息未在状态中跟踪，请使用[自定义更新](#自定义更新)或在流式循环中聚合块（[下一节](#访问完整消息)）。
如果你的代理包含多个大语言模型，请参阅下面关于[从子代理流式传输](#从子代理流式传输)的部分。
```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
from typing import Any
from langchain.agents import create_agent
from langchain.messages import AIMessage, AIMessageChunk, AnyMessage, ToolMessage
def get_weather(city: str) -> str:
"""获取给定城市的天气。"""
return f"It's always sunny in {city}!"
agent = create_agent("openai:gpt-5.4", 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)
# 注意：所有内容均可通过 token.content_blocks 访问
def _render_completed_message(message: AnyMessage) -> None:
if isinstance(message, AIMessage) and message.tool_calls:
print(f"工具调用: {message.tool_calls}")
if isinstance(message, ToolMessage):
print(f"工具响应: {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` 捕获节点名称
_render_completed_message(update["messages"][-1]) # [!code highlight]
```
```shell title="输出" expandable theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
[{'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'}]
工具调用: [{'name': 'get_weather', 'args': {'city': 'Boston'}, 'id': 'call_D3Orjr89KgsLTZ9hTzYv7Hpf', 'type': 'tool_call'}]
工具响应: [{'type': 'text', 'text': "It's always sunny in Boston!"}]
The| weather| in| Boston| is| **|sun|ny|**|.|
```
#### 访问完整消息
如果完整消息在代理的[状态](/oss/python/langchain/agents#memory)中被跟踪，你可以使用 `stream_mode=["messages", "updates"]`，如[流式传输工具调用](#流式传输工具调用)部分所示，在流式传输期间访问完整消息。
在某些情况下，完整消息不会反映在[状态更新](#代理进度)中。如果你可以访问代理内部，你可以使用[自定义更新](#自定义更新)在流式传输期间访问这些消息。否则，你可以在流式循环中聚合消息块（见下文）。
考虑下面的示例，我们将一个[流式写入器](#自定义更新)集成到一个简化的[护栏中间件](/oss/python/langchain/guardrails#after-agent-guardrails)中。这个中间件演示了工具调用以生成结构化的“安全/不安全”评估（也可以使用[结构化输出](/oss/python/langchain/models#structured-output)来实现）：
```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
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):
"""将响应评估为安全或不安全。"""
evaluation: Literal["safe", "unsafe"]
safety_model = init_chat_model("openai:gpt-5.4")
@after_agent(can_jump_to=["end"])
def safety_guardrail(state: AgentState, runtime: Runtime) -> dict[str, Any] | None:
"""基于模型的护栏：使用大语言模型评估响应安全性。"""
stream_writer = get_stream_writer() # [!code highlight]
# 获取模型响应
if not state["messages"]:
return None
last_message = state["messages"][-1]
if not isinstance(last_message, AIMessage):
return None
# 使用另一个模型评估安全性
model_with_tools = safety_model.bind_tools([ResponseSafety], tool_choice="any")
result = model_with_tools.invoke(
[
{
"role": "system",
"content": "将此 AI 响应评估为通常安全或不安全。"
},
{
"role": "user",
"content": f"AI 响应: {last_message.text}"
}
]
)
stream_writer(result) # [!code highlight]
tool_call = result.tool_calls[0]
if tool_call["args"]["evaluation"] == "unsafe":
last_message.content = "我无法提供该响应。请重新表述你的请求。"
return None
```
然后我们可以将这个中间件集成到我们的代理中，并包含其自定义流事件：
```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
from typing import Any
from langchain.agents import create_agent
from langchain.messages import AIMessageChunk, AIMessage, AnyMessage
def get_weather(city: str) -> str:
"""获取给定城市的天气。"""
return f"It's always sunny in {city}!"
agent = create_agent(
model="openai:gpt-5.4",
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"工具调用: {message.tool_calls}")
if isinstance(message, ToolMessage):
print(f"工具响应: {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]
# 在流中访问完整消息
print(f"工具调用: {chunk['data'].tool_calls}") # [!code highlight]
```
```shell title="输出" expandable theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
[{'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'}]
工具调用: [{'name': 'get_weather', 'args': {'city': 'Boston'}, 'id': 'call_je6LWgxYzuZ84mmoDalTYMJC', 'type': 'tool_call'}]
工具响应: [{'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'}]
工具调用: [{'name': 'ResponseSafety', 'args': {'evaluation': 'safe'}, 'id': 'call_O8VJIbOG4Q9nQF0T8ltVi58O', 'type': 'tool_call'}]
```
或者，如果你无法向流添加自定义事件，你可以在流式循环中聚合消息块：
```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
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"工具调用: {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])
```
### 与人机交互一起流式传输
为了处理人机交互[中断](/oss/python/langchain/human-in-the-loop)，我们基于[上面的示例](#流式传输工具调用)进行构建：
1. 我们使用[人机交互中间件和检查点](/oss/python/langchain/human-in-the-loop#configuring-interrupts)配置代理
2. 我们收集在 `"updates"` 流式模式期间生成的中断
3. 我们使用[命令](/oss/python/langchain/human-in-the-loop#responding-to-interrupts)响应这些中断
```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
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:
"""获取给定城市的天气。"""
return f"It's always sunny in {city}!"
checkpointer = InMemorySaver()
agent = create_agent(
"openai:gpt-5.4",
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"工具调用: {message.tool_calls}")
if isinstance(message, ToolMessage):
print(f"工具响应: {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": (
"你能查一下波士顿和旧金山的天气吗？"
),
}
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]
```
```shell title="输出" expandable theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
[{'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'}]
工具调用: [{'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'}]
工具执行需要批准
工具: get_weather
参数: {'city': 'Boston'}
工具执行需要批准
工具: get_weather
参数: {'city': 'San Francisco'}
```
接下来，我们为每个中断收集一个[决策](/oss/python/langchain/human-in-the-loop#interrupt-decision-types)。重要的是，决策的顺序必须与我们收集的操作顺序相匹配。
为了说明，我们将编辑一个工具调用并接受另一个：
```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
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
```
```shell title="输出" theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
{
'a96c40474e429d661b5b32a8d86f0f3e': {
'decisions': [
{
'type': 'edit',
'edited_action': {
'name': 'get_weather',
'args': {'city': 'Boston, U.K.'}
}
},
{'type': 'approve'},
]
}
}
```
然后我们可以通过将[命令](/oss/python/langchain/human-in-the-loop#responding-to-interrupts)传递到同一个流式循环中来恢复：
```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
interrupts = []
for chunk in agent.stream(
Command(resume=decisions), # [!code highlight]
config=config,
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__":
interrupts.extend(update)
_render_interrupt(update[0])
```
```shell title="输出" theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
工具响应: [{'type': 'text', 'text': "It's always sunny in Boston, U.K.!"}]
工具响应: [{'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|!|
```
### 从子代理流式传输
当代理中的任何点存在多个大语言模型时，通常需要在生成消息时区分消息的来源。
为此，在创建每个代理时传递一个 [`name`](https://reference.langchain.com/python/langchain/agents/#langchain.agents.create_agent\(name\))。然后，在以 `"messages"` 模式流式传输时，该名称可通过元数据中的 `lc_agent_name` 键获得。
下面，我们更新[流式传输工具调用](#流式传输工具调用)示例：
1. 我们将工具替换为一个 `call_weather_agent` 工具，该工具内部调用一个代理
2. 我们为每个代理添加一个 `name`
3. 我们在创建流时指定 [`subgraphs=True`](/oss/python/langgraph/use-subgraphs#stream-subgraph-outputs)
4. 我们的流处理与之前相同，但我们添加了逻辑以使用 `create_agent` 的 `name` 参数跟踪哪个代理处于活动状态
当你为代理设置 `name` 时，该名称也会附加到该代理生成的任何 `AIMessage` 上。
首先我们构建代理：
```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
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:
"""获取给定城市的天气。"""
return f"It's always sunny in {city}!"
weather_model = init_chat_model("openai:gpt-5.4")
weather_agent = create_agent(
model=weather_model,
tools=[get_weather],
name="weather_agent", # [!code highlight]
)
def call_weather_agent(query: str) -> str:
"""查询天气代理。"""
result = weather_agent.invoke({
"messages": [{"role": "user", "content": query}]
})
return result["messages"][-1].text
supervisor_model = init_chat_model("openai:gpt-5.4")
agent = create_agent(
model=supervisor_model,
tools=[call_weather_agent],
name="supervisor", # [!code highlight]
)
```
接下来，我们向流式循环添加逻辑以报告哪个代理正在发出令牌：
```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
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"工具调用: {message.tool_calls}")
if isinstance(message, ToolMessage):
print(f"工具响应: {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, AIMessage):
_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])
```
```shell title="输出" expandable theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
🤖 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'}]
工具调用: [{'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'}]
工具调用: [{'name': 'get_weather', 'args': {'city': 'Boston'}, 'id': 'call_LZ89lT8fW6w8vqck5pZeaDIx', 'type': 'tool_call'}]
工具响应: [{'type': 'text', 'text': "It's always sunny in Boston!"}]
Boston| weather| right| now|:| **|Sunny|**|.
|Today|'s| forecast| for| Boston|:| **|Sunny| all| day|**|.|工具响应: [{'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|**|.|
```
## 禁用流式传输
在某些应用程序中，你可能需要禁用给定模型的单个令牌流式传输。这在以下情况下很有用：
* 使用[多代理](/oss/python/langchain/multi-agent)系统时，控制哪些代理流式传输其输出
* 混合支持流式传输和不支持流式传输的模型
* 部署到 [LangSmith](/langsmith/home) 并希望防止某些模型输出流式传输到客户端
在初始化模型时设置 `streaming=False`。
```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
from langchain_openai import ChatOpenAI
model = ChatOpenAI(
model="gpt-5.4",
streaming=False # [!code highlight]
)
```
部署到 LangSmith 时，在任何不希望其输出流式传输到客户端的模型上设置 `streaming=False`。这在部署前在你的图代码中配置。
并非所有聊天模型集成都支持 `streaming` 参数。如果你的模型不支持，请改用 `disable_streaming=True`。此参数通过基类在所有聊天模型上可用。
有关更多详细信息，请参阅 [LangGraph 流式传输指南](/oss/python/langgraph/streaming#disable-streaming-for-specific-chat-models)。
## v2 流式格式
需要 LangGraph >= 1.1。
将 `version="v2"` 传递给 `stream()` 或 `astream()` 以获得统一的输出格式。每个块都是一个 `StreamPart` 字典，包含 `type`、`ns` 和 `data` 键——无论流式模式或模式数量如何，形状都相同：
```python v2 (new) theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
# 统一格式 - 不再需要元组解包
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"]) # 有效负载
```
```python v1 (current default) theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
# 必须解包 (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()`——它返回一个 `GraphOutput` 对象，具有 `.value` 和 `.interrupts` 属性，清晰地将状态与中断元数据分开：
```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
result = agent.invoke(
{"messages": [{"role": "user", "content": "Hello"}]},
version="v2",
)
print(result.value) # 状态（字典、Pydantic 模型或 dataclass）
print(result.interrupts) # Interrupt 对象的元组（如果没有则为空）
```
有关 v2 格式的更多详细信息，包括类型收窄、Pydantic/dataclass 强制转换和子图流式传输，请参阅 [LangGraph 流式传输文档](/oss/python/langgraph/streaming#stream-output-format-v2)。
## 相关内容
* [前端流式传输](/oss/python/langchain/streaming/frontend)—使用 `useStream` 构建 React UI 以实现实时代理交互
* [使用聊天模型进行流式传输](/oss/python/langchain/models#stream)—直接从聊天模型流式传输令牌，无需使用代理或图
* [使用聊天模型进行推理](/oss/python/langchain/models#reasoning)—配置和访问聊天模型的推理输出
* [标准内容块](/oss/python/langchain/messages#standard-content-blocks)—了解用于推理、文本和其他内容类型的标准化内容块格式
* [与人机交互一起流式传输](/oss/python/langchain/human-in-the-loop#streaming-with-human-in-the-loop)—在处理中断以供人工审查时流式传输代理进度
* [LangGraph 流式传输](/oss/python/langgraph/streaming)—高级流式传输选项，包括 `values`、`debug` 模式和子图流式传输
***
[将这些文档](/use-these-docs)通过 MCP 连接到 Claude、VSCode 等，以获取实时答案。
[在 GitHub 上编辑此页面](https://github.com/langchain-ai/docs/edit/main/src/oss/langchain/streaming.mdx) 或 [提交问题](https://github.com/langchain-ai/docs/issues/new/choose)。