核心组件
事件流

复制页面

LangChain 智能体构建在 LangGraph 之上，因此它们支持相同的流式传输栈，并提供了针对消息、工具调用、状态和自定义更新的智能体聚焦投影。
对于大多数应用和前端用例，请通过stream_events(..., version="v3")使用事件流。事件流返回一个带有类型化投影的运行对象，因此每个投影都可以独立消费，而无需解析流模式元组。

fromlangchain.agentsimportcreate_agentdefget_weather(city:str)->str:"""获取城市天气。"""returnf"{city}的天气总是晴朗！"agent=create_agent(model="gpt-5-nano",tools=[get_weather],)stream=agent.stream_events({"messages":[{"role":"user","content":"旧金山的天气怎么样？"}],},version="v3")formessageinstream.messages:fordeltainmessage.text:print(delta,end="",flush=True)final_state=stream.output

你可以流式传输的内容

stream.messages产生ChatModelStream对象。每个消息流都暴露了.text、.reasoning、.tool_calls和.output。同步投影可迭代以获取实时增量，并可耗尽以获取最终值：使用str(message.text)获取最终文本，使用message.tool_calls.get()获取最终确定的工具调用。

智能体消息

当你需要来自每次 LLM 调用的模型输出时，请使用stream.messages。

stream=agent.stream_events(input,version="v3")formessageinstream.messages:print(f"[{message.node}] ",end="")fordeltainmessage.text:print(delta,end="",flush=True)full_message=message.output usage=full_message.usage_metadataifusage:print(usage)

message.output提供最终确定 AI 消息，包括特定于提供者的内容块。在 TypeScript 中，当你只需要令牌计数或其他使用情况元数据时，请使用message.usage；在 Python 中，请从message.output.usage_metadata读取使用情况。

推理内容

推理内容与文本内容的形状相同，但仅当所选模型发出推理块时才可用。

stream=agent.stream_events(input,version="v3")formessageinstream.messages:fordeltainmessage.reasoning:print(f"[思考]{delta}",end="",flush=True)fordeltainmessage.text:print(delta,end="",flush=True)

请参阅推理指南和你提供商的集成页面以获取模型配置详细信息。

工具调用

有两个有用的工具调用投影：

• message.tool_calls在模型生成工具调用时流式传输工具调用参数块。
• stream.tool_calls在工具调用开始后流式传输工具执行的生命周期。

stream=agent.stream_events(input,version="v3")formessageinstream.messages:forchunkinmessage.tool_calls:print(f"工具调用块:{chunk}")finalized=message.tool_calls.get()iffinalized:print(f"最终确定的工具调用:{finalized}")forcallinstream.tool_calls:print(f"{call.tool_name}({call.input})")fordeltaincall.output_deltas:print(delta,end="",flush=True)print(call.output,call.error)

流式传输子智能体

当一个create_agent调用调用另一个命名的create_agent（通常通过包装工具）时，内部智能体的事件会在嵌套命名空间下流动。你传递给create_agent的name=会在流中标识该内部智能体，因此你可以按每个智能体进行筛选和标记。

命名的子智能体会出现在专用的stream.subagents投影上。每个句柄都暴露了内部智能体自身的.messages、.values、.tool_calls和.output，以及.name（你传递的name=）和.cause（分派子智能体的工具调用）。因为只有命名的create_agent运行会出现在这里，所以你无需筛选出普通子图。

fromlangchain.agentsimportcreate_agentfromlangchain.chat_modelsimportinit_chat_modeldefget_weather(city:str)->str:"""获取给定城市的天气。"""returnf"{city}的天气总是晴朗！"weather_agent=create_agent(model=init_chat_model("openai:gpt-5.5"),tools=[get_weather],name="weather_agent",)defcall_weather(query:str)->str:"""查询天气智能体。"""result=weather_agent.invoke({"messages":[{"role":"user","content":query}]})returnresult["messages"][-1].text supervisor=create_agent(model=init_chat_model("openai:gpt-5.5"),tools=[call_weather],name="supervisor",)stream=supervisor.stream_events({"messages":[{"role":"user","content":"波士顿的天气怎么样？"}]},version="v3",)forsubagentinstream.subagents:print(f"{subagent.name}: ",end="")formessageinsubagent.messages:fortokeninmessage.text:print(token,end="",flush=True)print()

从工具调用的普通StateGraph子图也会出现在stream.subgraphs上——在.compile(name=...)上设置name=以在subagent.graph_name中获取标签。

stream.subagents是命名create_agent子智能体的聚焦视图，而stream.subgraphs涵盖每个嵌套图。根据你的 UI 使用任何一个。

状态和最终输出

使用stream.values获取状态快照，使用stream.output获取最终智能体状态。

stream=agent.stream_events(input,version="v3")forsnapshotinstream.values:print(snapshot)final_state=stream.output

多个投影

在异步代码中并发消费时，请将astream_events与asyncio.gather结合使用：

importasyncio stream=awaitagent.astream_events(input,version="v3")asyncdefconsume_messages():asyncformessageinstream.messages:print(awaitmessage.text)asyncdefconsume_tool_calls():asyncforcallinstream.tool_calls:print(call.tool_name,call.input)awaitasyncio.gather(consume_messages(),consume_tool_calls())

对于同步代码，请改用stream.interleave(...)：

stream=agent.stream_events(input,version="v3")forname,iteminstream.interleave("messages","tool_calls","values"):ifname=="messages":print(item.text)elifname=="tool_calls":print(item.tool_name,item.input)elifname=="values":print(item)

要访问未作为类型化投影公开的通道，或检查完整事件信封，请迭代原始协议事件：

foreventinstream:print(event["method"],event["params"]["namespace"],event["params"]["data"])

自定义更新

当你的应用程序需要非内置的投影时，请使用自定义流转换器，例如检索进度、工件或领域特定事件。

stream=agent.stream_events(input,version="v3",transformers=[ToolActivityTransformer],)foractivityinstream.extensions["tool_activity"]:print(activity)

在中间件上注册转换器

中间件注册的转换器需要langchain>=1.3.2。

中间件可以与其钩子和工具一起声明流转换器工厂。不同语言之间的工厂形状不同：

在AgentMiddleware子类上将transformers属性设置为工厂序列。每个工厂的形状为Callable[[tuple[str, ...]], StreamTransformer]，并作为factory(scope)调用，其中scope是迷你复用器作用域元组（根复用器为()，子图非空）。每次调用返回一个新的转换器可保持每个子图隔离。

fromlangchain.agentsimportcreate_agentfromlangchain.agents.middlewareimportAgentMiddlewareclassToolActivityMiddleware(AgentMiddleware):transformers=(ToolActivityTransformer,)agent=create_agent(model="gpt-5-nano",tools=[get_weather],middleware=[ToolActivityMiddleware()],)

在编译时，create_agent会合并中间件注册的工厂以及传递给其自身transformers=参数的任何内容。编译后的图上的最终顺序是：

1. 内置的ToolCallTransformer。
2. 中间件注册的工厂，按中间件顺序。
3. 调用者通过create_agent提供的transformers=。

这使内置的工具调用投影保持在消费者转换器之前，并给予调用者提供的条目最终决定权。

内置的PIIMiddleware使用此钩子从流式传输的线路输出中编辑 PII。当apply_to_output=True时，其注册的转换器会在文本增量、工具调用参数、工具输出和状态快照离开运行之前从其中清除检测到的 PII，从而关闭了否则after_model状态级编辑会让原始 PII 泄露给stream_events(version="v3")的实时读者的窗口。

fromlangchain.agentsimportcreate_agentfromlangchain.agents.middlewareimportPIIMiddleware agent=create_agent(model="gpt-5-nano",tools=[],middleware=[PIIMiddleware("email",strategy="redact",apply_to_output=True),],)

请参阅 PII 检测以获取完整配置界面。
请参阅构建你自己的投影以获取转换器契约。

相关文档

• 流式传输涵盖低级 Pregel 流模式。
• 构建你自己的投影涵盖编写特定于应用程序的投影。
• 前端流模式展示基于流式状态构建的 UI 用例。