概述
LangChain 的流式系统允许您将智能体运行的实时反馈展示到您的应用程序中。 使用 LangChain 流式传输可以实现:- 流式传输智能体进度—在每次智能体步骤后获取状态更新。
- 流式传输 LLM 令牌—流式传输正在生成的语言模型令牌。
- 流式传输思考/推理令牌—展示正在生成的模型推理过程。
- 流式传输自定义更新—发射用户定义的信号(例如,
"Fetched 10/100 records")。 - 流式传输多种模式—从
updates(智能体进度)、messages(LLM 令牌 + 元数据)或custom(任意用户数据)中选择。
支持的流模式
将以下一个或多个流模式作为列表传递给stream 或 astream 方法:
| 模式 | 描述 |
|---|---|
updates | 在每个智能体步骤后流式传输状态更新。如果同一步骤中有多个更新(例如,运行了多个节点),这些更新将分别流式传输。 |
messages | 流式传输来自任何调用 LLM 的图节点的 (token, metadata) 元组。 |
custom | 使用流写入器从您的图节点内部流式传输自定义数据。 |
智能体进度
要流式传输智能体进度,请使用stream 或 astream 方法并设置 stream_mode="updates"。这会在每个智能体步骤后发射一个事件。
例如,如果您有一个调用一次工具的代理,您应该看到以下更新:
- LLM 节点:带有工具调用请求的
AIMessage - 工具节点:带有执行结果的
ToolMessage - LLM 节点:最终 AI 响应
Streaming agent progress
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}")
Output
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 令牌
要流式传输由 LLM 产生的令牌,请使用stream_mode="messages"。下面您可以看到代理流式传输工具调用和最终响应的输出。
Streaming LLM tokens
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")
Output
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。
Streaming custom updates
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()
# 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-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"])
Output
Looking up data for city: San Francisco
Acquired data for city: San Francisco
如果您在工具内部添加
get_stream_writer,您将无法在 LangGraph 执行上下文之外调用该工具。流式传输多种模式
您可以通过传递流模式列表来指定多个流模式:stream_mode=["updates", "custom"]。
每个流式传输的块都是一个 StreamPart 字典,包含 type、ns 和 data 键。使用 chunk["type"] 确定流模式,并使用 chunk["data"] 访问负载。
Streaming multiple modes
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="gpt-5-nano",
tools=[get_weather],
)
for chunk in agent.stream(
{"messages": [{"role": "user", "content": "What is the weather in SF?"}]},
stream_mode=["updates", "custom"],
version="v2",
):
print(f"stream_mode: {chunk['type']}")
print(f"content: {chunk['data']}")
print("\n")
Output
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: 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}})]}}
常见模式
以下是展示流式传输常见用例的示例。流式传输思考/推理令牌
某些模型在生成最终答案之前会进行内部推理。您可以通过过滤 标准内容块 的type 为 "reasoning" 来流式传输这些正在生成的思考/推理令牌。
要从智能体流式传输思考令牌,请使用
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="")
Output
[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!
thinking 块、OpenAI reasoning 摘要等)通过 content_blocks 属性标准化为标准的 "reasoning" 内容块类型。
要从聊天模型直接流式传输推理令牌(不使用智能体),请参阅 与聊天模型流式传输。
流式传输工具调用
您可能希望同时流式传输以下内容:- 随着 工具调用 生成而生成部分 JSON
- 已完成的、解析后的执行工具调用
stream_mode="messages" 将流式传输智能体中所有 LLM 调用生成的增量 消息块。要访问带有解析后工具调用的已完成消息:
- 如果这些消息在 状态 中被跟踪(如
create_agent的模型节点中),请使用stream_mode=["messages", "updates"]通过 状态更新 访问已完成的消息(如下所示演示)。 - 如果这些消息未在状态中跟踪,请使用 自定义更新 或在流式循环中聚合块(下一节)。
如果您的智能体包含多个 LLM,请参阅下面关于 从子智能体流式传输 的部分。
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.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 chunk in agent.stream(
{"messages": [input_message]},
stream_mode=["messages", "updates"],
version="v2",
):
if chunk["type"] == "messages":
token, metadata = chunk["data"]
if isinstance(token, AIMessageChunk):
_render_message_chunk(token)
elif chunk["type"] == "updates":
for source, update in chunk["data"].items():
if source in ("model", "tools"): # `source` captures node name
_render_completed_message(update["messages"][-1])
Output
[{'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|**|.|
访问已完成的消息
在某些情况下,已完成的消息不会反映在 状态更新 中。如果您可以访问智能体内部,您可以使用 自定义更新 在流式传输期间访问这些消息。否则,您可以在流式循环中聚合消息块(见下文)。 考虑以下示例,我们将 流写入器 纳入简化的 护栏中间件。此中间件演示了工具调用以生成结构化的“安全/不安全”评估(也可以使用 结构化输出 来完成此操作):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
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
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.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 chunk in agent.stream(
{"messages": [input_message]},
stream_mode=["messages", "updates", "custom"],
version="v2",
):
if chunk["type"] == "messages":
token, metadata = chunk["data"]
if isinstance(token, AIMessageChunk):
_render_message_chunk(token)
elif chunk["type"] == "updates":
for source, update in chunk["data"].items():
if source in ("model", "tools"):
_render_completed_message(update["messages"][-1])
elif chunk["type"] == "custom":
# access completed message in stream
print(f"Tool calls: {chunk['data'].tool_calls}")
Output
[{'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'}]
input_message = {"role": "user", "content": "What is the weather in Boston?"}
full_message = None
for chunk in agent.stream(
{"messages": [input_message]},
stream_mode=["messages", "updates"],
version="v2",
):
if chunk["type"] == "messages":
token, metadata = chunk["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
elif chunk["type"] == "updates":
for source, update in chunk["data"].items():
if source == "tools":
_render_completed_message(update["messages"][-1])
人机回环流式传输
为了处理人类参与循环的 中断,我们在 上面的示例 基础上构建:- 我们配置智能体使用 人类参与循环中间件和检查点器
- 我们收集在
"updates"流模式下生成的中断 - 我们使用 命令 响应这些中断
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.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 chunk in agent.stream(
{"messages": [input_message]},
config=config,
stream_mode=["messages", "updates"],
version="v2",
):
if chunk["type"] == "messages":
token, metadata = chunk["data"]
if isinstance(token, AIMessageChunk):
_render_message_chunk(token)
elif chunk["type"] == "updates":
for source, update in chunk["data"].items():
if source in ("model", "tools"):
_render_completed_message(update["messages"][-1])
if source == "__interrupt__":
interrupts.extend(update)
_render_interrupt(update[0])
Output
[{'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'}
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
{
'a96c40474e429d661b5b32a8d86f0f3e': {
'decisions': [
{
'type': 'edit',
'edited_action': {
'name': 'get_weather',
'args': {'city': 'Boston, U.K.'}
}
},
{'type': 'approve'},
]
}
}
interrupts = []
for chunk in agent.stream(
Command(resume=decisions),
config=config,
stream_mode=["messages", "updates"],
version="v2",
):
# Streaming loop is unchanged
if chunk["type"] == "messages":
token, metadata = chunk["data"]
if isinstance(token, AIMessageChunk):
_render_message_chunk(token)
elif chunk["type"] == "updates":
for source, update in chunk["data"].items():
if source in ("model", "tools"):
_render_completed_message(update["messages"][-1])
if source == "__interrupt__":
interrupts.extend(update)
_render_interrupt(update[0])
Output
Tool 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|!|
从子智能体流式传输
当智能体中的任何位置有多个 LLM 时,通常有必要区分生成消息的来源。 为此,在创建每个智能体时传递一个name。然后在 "messages" 模式下流式传输时,该名称可通过 lc_agent_name 键在元数据中获取。
下面,我们更新 流式传输工具调用 示例:
- 我们用
call_weather_agent工具替换我们的工具,该工具在内部调用智能体 - 我们为每个智能体添加一个
name - 创建流时指定
subgraphs=True - 我们的流处理与之前相同,但我们添加了逻辑以使用
create_agent的name参数跟踪当前活动的智能体
当您为智能体设置
name 时,该名称也会附加到该智能体生成的任何 AIMessage 上。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.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",
)
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
for chunk in agent.stream(
{"messages": [input_message]},
stream_mode=["messages", "updates"],
subgraphs=True,
version="v2",
):
if chunk["type"] == "messages":
token, metadata = chunk["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)
elif chunk["type"] == "updates":
for source, update in chunk["data"].items():
if source in ("model", "tools"):
_render_completed_message(update["messages"][-1])
Output
🤖 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|**|.|
禁用流式传输
在某些应用程序中,您可能需要禁用给定模型的单个令牌的流式传输。这在以下情况下很有用: 初始化模型时设置streaming=False。
from langchain_openai import ChatOpenAI
model = ChatOpenAI(
model="gpt-4.1",
streaming=False
)
部署到 LangSmith 时,对任何您不希望流式传输到客户端的模型设置
streaming=False。这是在部署前在您的图代码中配置的。并非所有聊天模型集成都支持
streaming 参数。如果您的模型不支持它,请使用 disable_streaming=True。此参数通过基类在所有聊天模型上可用。v2 流式格式
需要 LangGraph >= 1.1。
stream() 或 astream() 传递 version="v2" 以获得统一的输出格式。每个块都是一个 StreamPart 字典,包含 type、ns 和 data 键—无论流模式或模式数量如何,形状都相同:
# 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
invoke()—它返回一个带有 .value 和 .interrupts 属性的 GraphOutput 对象,清晰地将状态与中断元数据分离:
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)
相关资源
- 前端流式传输—使用
useStream构建 React UI 以实现实时智能体交互 - 与聊天模型流式传输—直接从聊天模型流式传输令牌,无需使用智能体或图
- 与聊天模型推理—配置和访问聊天模型的推理输出
- 标准内容块—了解用于推理、文本和其他内容类型的标准化内容块格式
- 人机回环流式传输—在处理人类审查的中断时流式传输智能体进度
- LangGraph 流式传输—高级流式传输选项,包括
values、debug模式和子图流式传输
Connect these docs to Claude, VSCode, and more via MCP for real-time answers.

