核心组件事件流复制页面LangChain 智能体构建在 LangGraph 之上因此它们支持相同的流式传输栈并提供了针对消息、工具调用、状态和自定义更新的智能体聚焦投影。对于大多数应用和前端用例请通过stream_events(..., versionv3)使用事件流。事件流返回一个带有类型化投影的运行对象因此每个投影都可以独立消费而无需解析流模式元组。fromlangchain.agentsimportcreate_agentdefget_weather(city:str)-str:获取城市天气。returnf{city}的天气总是晴朗agentcreate_agent(modelgpt-5-nano,tools[get_weather],)streamagent.stream_events({messages:[{role:user,content:旧金山的天气怎么样}],},versionv3)formessageinstream.messages:fordeltainmessage.text:print(delta,end,flushTrue)final_statestream.output你可以流式传输的内容投影用途for event in stream原始协议事件包含完整信封并可访问每个通道。stream.messages模型消息流每次 LLM 调用对应一个。message.text消息的文本增量和最终文本。message.reasoning针对暴露推理内容的模型的推理增量。message.tool_calls工具调用参数块和最终确定的工具调用。message.output模型调用完成后的最终消息对象。stream.values智能体状态快照。stream.output最终智能体状态。stream.subgraphs嵌套图运行子智能体和普通子图。stream.extensions自定义转换器投影。stream.tool_calls工具执行生命周期、输入、输出增量、最终输出和错误。stream.messages产生ChatModelStream对象。每个消息流都暴露了.text、.reasoning、.tool_calls和.output。同步投影可迭代以获取实时增量并可耗尽以获取最终值使用str(message.text)获取最终文本使用message.tool_calls.get()获取最终确定的工具调用。智能体消息当你需要来自每次 LLM 调用的模型输出时请使用stream.messages。streamagent.stream_events(input,versionv3)formessageinstream.messages:print(f[{message.node}] ,end)fordeltainmessage.text:print(delta,end,flushTrue)full_messagemessage.output usagefull_message.usage_metadataifusage:print(usage)message.output提供最终确定 AI 消息包括特定于提供者的内容块。在 TypeScript 中当你只需要令牌计数或其他使用情况元数据时请使用message.usage在 Python 中请从message.output.usage_metadata读取使用情况。推理内容推理内容与文本内容的形状相同但仅当所选模型发出推理块时才可用。streamagent.stream_events(input,versionv3)formessageinstream.messages:fordeltainmessage.reasoning:print(f[思考]{delta},end,flushTrue)fordeltainmessage.text:print(delta,end,flushTrue)请参阅推理指南和你提供商的集成页面以获取模型配置详细信息。工具调用有两个有用的工具调用投影message.tool_calls在模型生成工具调用时流式传输工具调用参数块。stream.tool_calls在工具调用开始后流式传输工具执行的生命周期。streamagent.stream_events(input,versionv3)formessageinstream.messages:forchunkinmessage.tool_calls:print(f工具调用块:{chunk})finalizedmessage.tool_calls.get()iffinalized:print(f最终确定的工具调用:{finalized})forcallinstream.tool_calls:print(f{call.tool_name}({call.input}))fordeltaincall.output_deltas:print(delta,end,flushTrue)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_agentcreate_agent(modelinit_chat_model(openai:gpt-5.5),tools[get_weather],nameweather_agent,)defcall_weather(query:str)-str:查询天气智能体。resultweather_agent.invoke({messages:[{role:user,content:query}]})returnresult[messages][-1].text supervisorcreate_agent(modelinit_chat_model(openai:gpt-5.5),tools[call_weather],namesupervisor,)streamsupervisor.stream_events({messages:[{role:user,content:波士顿的天气怎么样}]},versionv3,)forsubagentinstream.subagents:print(f{subagent.name}: ,end)formessageinsubagent.messages:fortokeninmessage.text:print(token,end,flushTrue)print()从工具调用的普通StateGraph子图也会出现在stream.subgraphs上——在.compile(name...)上设置name以在subagent.graph_name中获取标签。stream.subagents是命名create_agent子智能体的聚焦视图而stream.subgraphs涵盖每个嵌套图。根据你的 UI 使用任何一个。状态和最终输出使用stream.values获取状态快照使用stream.output获取最终智能体状态。streamagent.stream_events(input,versionv3)forsnapshotinstream.values:print(snapshot)final_statestream.output多个投影在异步代码中并发消费时请将astream_events与asyncio.gather结合使用importasyncio streamawaitagent.astream_events(input,versionv3)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(...)streamagent.stream_events(input,versionv3)forname,iteminstream.interleave(messages,tool_calls,values):ifnamemessages:print(item.text)elifnametool_calls:print(item.tool_name,item.input)elifnamevalues:print(item)要访问未作为类型化投影公开的通道或检查完整事件信封请迭代原始协议事件foreventinstream:print(event[method],event[params][namespace],event[params][data])自定义更新当你的应用程序需要非内置的投影时请使用自定义流转换器例如检索进度、工件或领域特定事件。streamagent.stream_events(input,versionv3,transformers[ToolActivityTransformer],)foractivityinstream.extensions[tool_activity]:print(activity)在中间件上注册转换器中间件注册的转换器需要langchain1.3.2。中间件可以与其钩子和工具一起声明流转换器工厂。不同语言之间的工厂形状不同在AgentMiddleware子类上将transformers属性设置为工厂序列。每个工厂的形状为Callable[[tuple[str, ...]], StreamTransformer]并作为factory(scope)调用其中scope是迷你复用器作用域元组根复用器为()子图非空。每次调用返回一个新的转换器可保持每个子图隔离。fromlangchain.agentsimportcreate_agentfromlangchain.agents.middlewareimportAgentMiddlewareclassToolActivityMiddleware(AgentMiddleware):transformers(ToolActivityTransformer,)agentcreate_agent(modelgpt-5-nano,tools[get_weather],middleware[ToolActivityMiddleware()],)在编译时create_agent会合并中间件注册的工厂以及传递给其自身transformers参数的任何内容。编译后的图上的最终顺序是内置的ToolCallTransformer。中间件注册的工厂按中间件顺序。调用者通过create_agent提供的transformers。这使内置的工具调用投影保持在消费者转换器之前并给予调用者提供的条目最终决定权。内置的PIIMiddleware使用此钩子从流式传输的线路输出中编辑 PII。当apply_to_outputTrue时其注册的转换器会在文本增量、工具调用参数、工具输出和状态快照离开运行之前从其中清除检测到的 PII从而关闭了否则after_model状态级编辑会让原始 PII 泄露给stream_events(versionv3)的实时读者的窗口。fromlangchain.agentsimportcreate_agentfromlangchain.agents.middlewareimportPIIMiddleware agentcreate_agent(modelgpt-5-nano,tools[],middleware[PIIMiddleware(email,strategyredact,apply_to_outputTrue),],)请参阅 PII 检测以获取完整配置界面。请参阅构建你自己的投影以获取转换器契约。相关文档流式传输涵盖低级 Pregel 流模式。构建你自己的投影涵盖编写特定于应用程序的投影。前端流模式展示基于流式状态构建的 UI 用例。