第 10 章流式输出与 Streaming文章目录第 10 章流式输出与 Streaming10.1 本章目标10.2 核心概念stream_mode 七种模式对比v2 vs v3 流式 API10.3 实战实战 1Token 级打字机效果messages 模式实战 2多模式组合流式v2 统一格式实战 3自定义流式事件custom 模式实战 4v3 事件流式 API推荐10.4 API 速查10.5 错误与避坑指南坑 1同步 stream() 在 async 函数中调用坑 2messages 模式未过滤不需要的模型输出坑 3custom 模式在节点外调用 get_stream_writer()坑 4v2 格式未判断 chunk 类型10.6 最佳实践总结10.1 本章目标学完本章你将能够理解stream()和astream()的区别和使用场景掌握所有stream_mode模式values/updates/debug/messages/custom/checkpoints/tasks学会 v2 统一格式和 v3 事件流式 API实现 Token 级打字机效果和自定义流式事件10.2 核心概念stream_mode 七种模式对比模式输出内容典型用途values每步后完整 State 快照查看完整状态变化updates每步增量更新含节点名了解哪个节点改了什么messagesLLM Token 流式输出(token, metadata)打字机效果custom通过get_stream_writer()的自定义数据进度条、状态通知debug尽可能多的调试信息排查问题checkpointscheckpoint 创建事件监控持久化tasks任务启动/完成事件监控并行任务v2 vs v3 流式 API特性v2v3推荐APIstream(mode[...], versionv2)stream_events(versionv3)输出格式统一 StreamPart 格式类型化投影中断检测需手动处理stream.interrupts属性最终状态需手动收集stream.output属性10.3 实战实战 1Token 级打字机效果messages 模式fromtypingimportTypedDict,Annotatedfromlanggraph.graphimportStateGraph,START,END,add_messagesfromlangchain_core.messagesimportBaseMessage,HumanMessage,SystemMessagefromlangchain_openaiimportChatOpenAIclassState(TypedDict):messages:Annotated[list[BaseMessage],add_messages]llmChatOpenAI(modelgpt-4o,temperature0.7,streamingTrue)defchat_node(state:State)-dict:responsellm.invoke([SystemMessage(content你是一个友好的助手)]list(state[messages]))return{messages:[response]}builderStateGraph(State)builder.add_node(chat,chat_node)builder.add_edge(START,chat)builder.add_edge(chat,END)graphbuilder.compile()# # 流式输出Token 级打字机效果# print(AI: ,end,flushTrue)forchunkingraph.stream({messages:[HumanMessage(content写一首关于编程的短诗)]},stream_modemessages,):ifisinstance(chunk,tuple):msg_chunk,metadatachunkifhasattr(msg_chunk,content)andmsg_chunk.content:print(msg_chunk.content,end,flushTrue)实战 2多模式组合流式v2 统一格式# 同时使用多种 stream_modeforchunkingraph.stream({messages:[HumanMessage(content写一首诗)]},stream_mode[values,updates,messages],versionv2,):chunk_typechunk.get(type)ifchunk_typevalues:print(f[values] 消息数:{len(chunk[data].get(messages,[]))})elifchunk_typeupdates:fornode_name,updateinchunk[data].items():print(f[updates] 节点 {node_name} 更新了 State)elifchunk_typemessages:msg_chunk,metadatachunk[data]ifhasattr(msg_chunk,content)andmsg_chunk.content:print(msg_chunk.content,end,flushTrue)实战 3自定义流式事件custom 模式fromlanggraph.configimportget_stream_writerimporttimedeflong_running_node(state:State)-dict:模拟长时间运行的节点通过 custom 模式发送进度通知。writerget_stream_writer()# 获取流式写入器writer({progress:0,status:开始处理...})time.sleep(0.5)writer({progress:30,status:正在分析数据...})time.sleep(0.5)writer({progress:60,status:正在生成结果...})time.sleep(0.5)writer({progress:100,status:处理完成})return{messages:[AIMessage(content处理完成)]}# 接收自定义事件forchunkingraph.stream({messages:[HumanMessage(content开始处理)]},stream_mode[custom,messages],versionv2,):ifchunk[type]custom:eventchunk[data]print(f\n [{event[progress]}%]{event[status]},end)elifchunk[type]messages:msg_chunk,_chunk[data]ifhasattr(msg_chunk,content)andmsg_chunk.content:print(f\nAI:{msg_chunk.content},end)实战 4v3 事件流式 API推荐streamgraph.stream_events({messages:[HumanMessage(content写一首诗)]},versionv3,)# 类型化投影直接遍历 LLM 消息 tokenformsginstream.messages:fortokeninmsg.text:print(token,end,flushTrue)# 获取最终状态finalstream.outputprint(f\n\n最终消息数:{len(final[messages])})# 检查中断ifstream.interrupted:print(f图被中断:{stream.interrupts})10.4 API 速查API完整签名入参说明返回值说明.stream(input, stream_mode)stream(input, stream_mode, version)input: 初始 State;stream_mode: 模式或模式列表;version:v2Iterator[dict]同步流式.astream(input, stream_mode)astream(input, stream_mode, version)同 streamAsyncIterator[dict]异步流式stream_modevalues模式名无完整 State 快照每步后完整状态stream_modeupdates模式名无增量更新每个节点的变更stream_modemessages模式名无(token, metadata)LLM Token 流stream_modecustom模式名无自定义数据通过 get_stream_writerstream_modedebug模式名无调试信息节点执行详情get_stream_writer()get_stream_writer()无Writer 函数在节点中发送自定义事件.stream_events(input, v3)stream_events(input, versionv3)input: 初始 State;version:v3EventsStreamv3 事件流式 API推荐10.5 错误与避坑指南坑 1同步 stream() 在 async 函数中调用# ❌ 错误写法asyncdefbad_handler():forchunkingraph.stream(...):# 同步 stream 在 async 中阻塞事件循环print(chunk)# ✅ 正确写法asyncdefgood_handler():asyncforchunkingraph.astream(...):# 使用 astreamprint(chunk)坑 2messages 模式未过滤不需要的模型输出# ❌ 错误所有 LLM 输出都混在一起forchunkingraph.stream(...,stream_modemessages):print(chunk[0].content)# 可能包含内部模型的输出# ✅ 正确使用 tags 过滤modelChatOpenAI(modelgpt-4o).with_config({tags:[user_facing]})forchunkingraph.stream(...,stream_modemessages):msg_chunk,metadatachunkifuser_facinginmetadata.get(tags,[]):print(msg_chunk.content)坑 3custom 模式在节点外调用 get_stream_writer()# ❌ 错误写法writerget_stream_writer()# 在节点外调用 → 报错# ✅ 正确写法defmy_node(state:State)-dict:writerget_stream_writer()# 在节点内调用writer({status:ok})坑 4v2 格式未判断 chunk 类型# ❌ 错误假设所有 chunk 都是 messagesforchunkingraph.stream(...,stream_mode[values,messages],versionv2):print(chunk[data][0].content)# values 类型的 chunk 没有 .content# ✅ 正确先判断类型forchunkingraph.stream(...,stream_mode[values,messages],versionv2):ifchunk[type]messages:msg_chunk,_chunk[data]print(msg_chunk.content)10.6 最佳实践总结生产环境使用astream()异步流式不阻塞事件循环支持更多并发Token 流式用messages模式调试用debug模式各取所需使用 v2 统一格式处理多模式组合versionv2统一返回StreamPart新项目优先使用 v3 事件流式 API类型化投影更安全、更易用自定义进度通知用custom模式配合get_stream_writer()实现实时进度