【LangChain组件05：Streaming】—— LangChain 流式输出 Streaming：从逐 Token 打字到异步 SSE 实战

LangChain 流式输出 Streaming从逐 Token 打字到异步 SSE 实战如果使用invoke()用户需要等待 Agent 完成所有步骤多次模型调用 工具执行才能看到结果。对于复杂任务这可能耗时十几秒甚至更长。流式输出解决了这个问题每生成一个 Token 就立即返回用户可以实时看到进展——回复像打字一样逐字显示极大地提升用户体验。本文基于 LangChain 官方文档Python与菜鸟教程 LangChain 系列沿材料分类组件05Streaming的路径组织覆盖从逐 Token 流式、逐步查看执行过程、自定义事件、异步流式到 FastAPI SSE 集成以及 stream_mode 速查表。一、先厘清为什么需要流式输出方式用户体验适用场景invoke()等待 → 一次性看到完整结果脚本、API、批处理stream()实时看到每一个 Token聊天界面、实时展示一句话结论流式输出把等完全部再一次性返回变成边生成边返回核心价值是降低首 token 感知延迟让用户感觉 Agent 在活着工作而不是卡死。LangChain 的 Agent 内置了完善的流式输出支持主要通过stream()/astream()方法配合不同的stream_mode实现。二、stream_mode“messages”——逐 Token 流式这是最细粒度的流式模式每个 chunk 对应一个 Tokenfromdotenvimportload_dotenv load_dotenv()fromlangchain.agentsimportcreate_agentfromlangchain.chat_modelsimportinit_chat_modelfromlangchain.messagesimportHumanMessage modelinit_chat_model(deepseek:deepseek-v4-flash)agentcreate_agent(modelmodel,system_prompt你是菜鸟教程 RUNOOB 的助手。)# stream_modemessages 逐个 Token 返回print(实时流式输出)formsg_chunk,metadatainagent.stream({messages:[HumanMessage(content用一句话介绍菜鸟教程 RUNOOB)]},stream_modemessages,):# msg_chunk 是 AIMessageChunk每个 chunk 只包含一小段内容ifmsg_chunk.content:print(msg_chunk.content,end,flushTrue)print()运行结果打字效果实时流式输出 菜鸟教程RUNOOB是一个面向编程初学者的免费在线学习平台提供丰富的技术教程和实战示例。2.1 理解 metadatametadata包含这个 chunk 的来源信息哪个节点产生的print(查看 metadata 信息\n)formsg_chunk,metadatainagent.stream({messages:[HumanMessage(content你好介绍一下你自己)]},stream_modemessages,):ifmsg_chunk.contentandlen(msg_chunk.content)5:print(f内容:{msg_chunk.content})print(f来源节点:{metadata.get(langgraph_node)})print(f消息类型:{type(msg_chunk).__name__})break# 只看第一个有意义的 chunk运行结果查看 metadata 信息 内容: 你好我是菜 来源节点: model 消息类型: AIMessageChunk三、stream_mode“updates”——逐步查看执行过程这个模式在构建需要显示思考过程的界面时非常有用fromlangchain.toolsimporttoolfromlangchain.agentsimportcreate_agentfromlangchain.chat_modelsimportinit_chat_modelfromlangchain.messagesimportHumanMessagetooldefsearch_course(keyword:str)-str:在菜鸟教程 RUNOOB 搜索课程courses{python:Python3 基础教程30章20小时,html:HTML 基础教程25章15小时}returncourses.get(keyword.lower(),未找到相关课程)agentcreate_agent(modelinit_chat_model(deepseek:deepseek-v4-flash,temperature0),tools[search_course],system_prompt你是菜鸟教程 RUNOOB 的课程顾问。,)# 使用 updates 模式查看每一步print( Agent 执行过程 \n)forchunkinagent.stream({messages:[HumanMessage(content帮我查一下 Python 课程)]},stream_modeupdates,):fornode_name,updateinchunk.items():print(f[{node_name}],end )ifmessagesinupdate:formsginupdate[messages]:ifmsg.typeai:ifhasattr(msg,tool_calls)andmsg.tool_calls:calls[tc[name]fortcinmsg.tool_calls]print(f请求调用:{calls})elifmsg.content:print(f回复:{msg.content[:80]})elifmsg.typetool:print(f工具返回 [{msg.name}]:{msg.content})运行结果展示了完整的执行链路 Agent 执行过程 [model] 请求调用: [search_course] [tools] 工具返回 [search_course]: Python3 基础教程30章20小时 [model] 回复: 菜鸟教程 RUNOOB 中有 Python3 基础教程共30章学习时长约20小时非常适合Python初学者入门学习。四、stream_mode“custom”——发送自定义事件通过 Middleware 的runtime.stream_writer()你可以向流中发送自定义事件进度通知、状态推送等fromlangchain.agents.middlewareimportbefore_model,after_modelbefore_modeldefnotify_before(state,runtime):在模型调用前发送自定义事件runtime.stream_writer({type:status,message:正在思考...,})returnNoneafter_modeldefnotify_after(state,runtime):在模型调用后发送自定义事件last_msgstate[messages][-1]ifstate.get(messages)elseNonehas_toolshasattr(last_msg,tool_calls)andlast_msg.tool_callsifhas_tools:tool_names[tc[name]fortcinlast_msg.tool_calls]runtime.stream_writer({type:status,message:f正在调用工具:{, .join(tool_names)}...,})else:runtime.stream_writer({type:status,message:回答已完成,})returnNoneagentcreate_agent(modelinit_chat_model(deepseek:deepseek-v4-flash,temperature0),tools[search_course],middleware[notify_before,notify_after],system_prompt你是菜鸟教程 RUNOOB 的课程顾问。,)# 使用 stream_mode[updates, custom] 同时接收两种事件print( 混合流式输出 \n)formode,chunkinagent.stream({messages:[HumanMessage(content查一下 Python 课程)]},stream_mode[updates,custom],):ifmodecustom:print(f[自定义事件] 状态:{chunk[message]})elifmodeupdates:fornode_name,updateinchunk.items():ifmessagesinupdate:formsginupdate[messages]:ifmsg.typeaiandmsg.content:print(f[回复]{msg.content})运行结果 混合流式输出 [自定义事件] 状态: 正在思考... [自定义事件] 状态: 正在调用工具: search_course... [回复] 菜鸟教程 RUNOOB 中有 Python3 基础教程共30章学习时长约20小时。 [自定义事件] 状态: 正在思考... [自定义事件] 状态: 回答已完成五、异步流式输出在 Web 服务中使用异步流式可以避免阻塞事件循环importasynciofromlangchain.agentsimportcreate_agentfromlangchain.chat_modelsimportinit_chat_modelfromlangchain.messagesimportHumanMessageasyncdefstream_agent():异步流式运行 Agentagentcreate_agent(modelinit_chat_model(deepseek:deepseek-v4-flash),system_prompt你是菜鸟教程 RUNOOB 的助手。,)full_responseasyncformsg_chunk,metadatainagent.astream({messages:[HumanMessage(content一句话介绍菜鸟教程)]},stream_modemessages,):ifmsg_chunk.content:full_responsemsg_chunk.contentprint(msg_chunk.content,end,flushTrue)print(f\n\n完整回复长度:{len(full_response)}字)asyncio.run(stream_agent())5.1 FastAPI 集成示例SSE 流式响应fromfastapiimportFastAPIfromfastapi.responsesimportStreamingResponsefromlangchain.agentsimportcreate_agentfromlangchain.chat_modelsimportinit_chat_modelfromlangchain.messagesimportHumanMessage appFastAPI()agentcreate_agent(modelinit_chat_model(deepseek:deepseek-v4-flash),system_prompt你是菜鸟教程 RUNOOB 的助手。,)app.get(/chat)asyncdefchat(message:str):聊天接口返回 SSE 流式响应asyncdefgenerate():asyncformsg_chunk,metadatainagent.astream({messages:[HumanMessage(contentmessage)]},stream_modemessages,):ifmsg_chunk.content:# SSE 格式data: xxx\n\nyieldfdata:{msg_chunk.content}\n\nyielddata: [DONE]\n\nreturnStreamingResponse(generate(),media_typetext/event-stream)# 启动uvicorn main:app --reload✅ 成功标志前端能像聊天框一样逐字显示回复。生产环境中建议将 Agent 实例创建为全局单例避免每次请求都重新创建。Agent 的创建开销很小主要是编译图但复用实例更高效。六、stream_mode 速查表模式粒度迭代对象典型用途messagesToken 级(AIMessageChunk, metadata)打字效果、实时聊天updates节点级{node_name: state_update}展示思考过程values节点级全量完整 state状态快照、调试custom自定义任意 dict进度通知、状态推送debug详细调试信息开发阶段排查问题stream_mode可以组合使用如stream_mode[updates, custom, messages]。但过多的模式会增加流中的事件量建议按需选择。七、进阶v2 流式格式与工具调用流式7.1 v2 统一流式格式LangGraph 1.1传versionv2给stream()/astream()所有 chunk 都变成统一的StreamPartdict含type、ns、data键不再需要按模式解包# v2新统一格式所有 chunk 都是 StreamPart dictforchunkinagent.stream({messages:[{role:user,content:What is the weather in SF?}]},stream_mode[updates,custom],versionv2,):print(chunk[type])# updates 或 customprint(chunk[data])# payload对比 v1当前默认需要手动解包(mode, data)元组# v1默认需解包 (mode, data) 元组formode,chunkinagent.stream({messages:[{role:user,content:What is the weather in SF?}]},stream_mode[updates,custom],):print(mode)# updates 或 customprint(chunk)# payloadv2 还改进了invoke()——返回GraphOutput对象用.value和.interrupts干净地分离状态与中断元数据resultagent.invoke({messages:[{role:user,content:Hello}]},versionv2,)print(result.value)# statedict、Pydantic 模型或 dataclassprint(result.interrupts)# 中断对象元组无则为空7.2 流式工具调用你可能需要同时流式展示① 工具调用参数JSON如何逐步生成② 执行完成的已解析工具调用真正被执行的。用stream_mode[messages, updates]组合即可fromtypingimportAnyfromlangchain.agentsimportcreate_agentfromlangchain.messagesimportAIMessage,AIMessageChunk,AnyMessage,ToolMessagedefget_weather(city:str)-str:Get weather for a given city.returnfIts always sunny in{city}!agentcreate_agent(openai:gpt-5.5,tools[get_weather])def_render_message_chunk(token:AIMessageChunk)-None:iftoken.text:print(token.text,end|)iftoken.tool_call_chunks:print(token.tool_call_chunks)def_render_completed_message(message:AnyMessage)-None:ifisinstance(message,AIMessage)andmessage.tool_calls:print(fTool calls:{message.tool_calls})ifisinstance(message,ToolMessage):print(fTool response:{message.content_blocks})input_message{role:user,content:What is the weather in Boston?}forchunkinagent.stream({messages:[input_message]},stream_mode[messages,updates],versionv2,):ifchunk[type]messages:token,metadatachunk[data]ifisinstance(token,AIMessageChunk):_render_message_chunk(token)elifchunk[type]updates:forsource,updateinchunk[data].items():ifsourcein(model,tools):_render_completed_message(update[messages][-1])运行结果既能看到参数逐步生成的tool_call_chunk也能看到完成的Tool calls和Tool response。八、总结你真正需要记住的 N 件事流式输出核心价值 降低首 token 感知延迟让用户感觉 Agent 在活着工作。messages 模式做打字效果每个 chunk 一个 Tokenmetadata里有来源节点。updates 模式展示思考过程逐步看到 model 请求工具、tools 返回、model 回复。custom 模式发自定义事件用runtime.stream_writer()推进度/状态适合做状态条。异步流式用 astreamWeb 服务里避免阻塞事件循环FastAPI 用 StreamingResponse SSE。stream_mode 可组合但要克制事件量会随模式增加。v2 格式更统一所有 chunk 都是StreamPartdict用chunk[type]/chunk[data]。带工具调用时用 messagesupdates既流式看参数生成又能拿到完成的工具调用。验证清单我用stream_modemessages实现了逐 Token 打字效果我能从 metadata 读取来源节点langgraph_node我用stream_modeupdates展示过 Agent 执行过程需要进度反馈时我用 Middleware 的runtime.stream_writer()发 custom 事件生产 Web 服务里我用astream StreamingResponse 输出 SSE我按需组合 stream_mode没有一次性把所有模式都开我知道 v2 格式的StreamPartdict会读 chunk[“type”] / chunk[“data”]参考资源LangChain 官方文档Streaming——https://docs.langchain.com/oss/python/langchain/streamingLangChain ReferenceCompiledStateGraph.stream / astream——https://reference.langchain.com/python/langgraph/graphs菜鸟教程 LangChain 系列——https://www.runoob.com/langchain/