个人主页 for_ever_love__ 欢迎各位大佬莅临其他栏目: 大模型开发从0到1 其他栏目: iOS项目总结大全 其他栏目: 我想学python了 其他栏目: iOS UI 文章目录LangGraph 流式输出stream_mode 到底该选哪个一、先建立一张对照表二、values 和 updates 的区别看一次输出就懂三、打字机效果messages 模式四、custom 模式想推什么推什么五、astream_events什么都能流式六、异步版本怎么选七、子图的流式八、常见坑汇总九、选择决策树小结LangGraph 流式输出stream_mode 到底该选哪个用户问一句界面上要等 20 秒才啪地弹出整段回答——这个体验在今天已经不能接受了。LangGraph 提供了好几种流式模式名字长得都差不多这篇把它们挨个讲清楚并给出选择依据。一、先建立一张对照表graph.stream(...)/graph.astream(...)的stream_mode参数决定你会收到什么。模式每次产出什么典型用途values完整 state 的快照调试、需要拿到全部字段updates本节点新增/修改的部分最常用展示这一步做了什么messagesLLM 的token 增量 节点元数据打字机效果custom你自己get_stream_writer()写的任意内容自定义进度条、中间提示debug极详细的内部信息排查问题checkpoints每次写入检查点的事件做持久化 UI、断点恢复tasks任务的开始/结束事件并行分支进度模式可以传列表stream_mode[updates, custom]这样每步会产出(mode, chunk)二元组。二、values 和 updates 的区别看一次输出就懂假设图是START → step_a → step_b → ENDstep_a写{foo: 1}step_b写{bar: 2}。stream_modevaluesforchunkingraph.stream({input:x},stream_modevalues):print(chunk)# {input: x}# {input: x, foo: 1}# {input: x, foo: 1, bar: 2}每次都是完整的、累加后的 state。字段越攒越多。stream_modeupdates# {step_a: {foo: 1}}# {step_b: {bar: 2}}每次只给你这一步改了什么并且带上了节点名。实践结论要做进度展示这一步在检索、那一步在生成→ 用updates要拿某个字段的最新值 → 用values更省事不用自己累积生产环境做前端流式我默认选updates因为带节点名方便做步骤高亮。三、打字机效果messages 模式想让用户看到模型一个字一个字往外蹦用messagesfromlangchain_core.messagesimportAIMessageChunkformsg,metadataingraph.stream({messages:[{role:user,content:讲个笑话}]},stream_modemessages,):ifisinstance(msg,AIMessageChunk):print(msg.content,end,flushTrue)metadata里带着当前节点名、langgraph 的tags等信息可以用来判断这段 token 来自哪个节点。但是messages 模式有两个大坑坑一只对流式的 LLM 调用生效。如果节点里用的是llm.invoke(...)不是流式messages 模式下你收不到任何 token只能等到节点结束。LangGraph 在节点内部没法把.invoke()拆成 token 流。坑二用了.with_structured_output()或者输出解析链时token 流会被截断。结构化输出必须等完整 JSON 才能解析中间过程没法流式吐。这时候要么放弃打字机要么改用custom模式自己发进度。如果你的模型调用嵌在更深的层次里比如被包在一个自定义函数里messages 模式也可能收不到。这种情况改用下面第五节说的astream_events。四、custom 模式想推什么推什么节点内部可以拿一个写流器往流里塞任意数据fromlanggraph.configimportget_stream_writerdeflong_task(state):writerget_stream_writer()steps[读文件,切分,向量化,入库]fori,sinenumerate(steps,1):do_something(s)writer({progress:i/len(steps),stage:s})# ← 推给前端return{done:True}消费端formode,chunkingraph.stream({input:...},stream_mode[updates,custom],):ifmodecustom:print(f进度{chunk[progress]:.0%}-{chunk[stage]})else:print(节点完成:,chunk)这是做进度条最干净的方式而且不依赖 LLM 是否流式。一个注意点get_stream_writer()只能在节点函数内部调用在节点外面拿不到。如果你把逻辑抽到了独立的工具函数里要么把 writer 传进去要么让工具函数返回结果后由节点统一推。五、astream_events什么都能流式astream_events是 LangChain 层面的统一事件流注意它是a开头异步专用能捕获所有内部事件链开始、LLM 开始、单个 token、工具开始、工具结束……asyncforeventingraph.astream_events({messages:[{role:user,content:查一下天气}]},versionv2,):kindevent[event]ifkindon_chat_model_stream:print(event[data][chunk].content,end,flushTrue)elifkindon_tool_start:print(f\n[调用工具]{event[name]}参数{event[data].get(input)})elifkindon_tool_end:print(f\n[工具返回]{event[data].get(output)})versionv2是必须的v1 已经不建议用了。常用事件类型事件含义on_chain_start/on_chain_end链或节点开始/结束on_chat_model_stream单个 token 增量on_tool_start/on_tool_end工具调用开始/结束on_retriever_end检索完成可以拿到召回的文档什么情况该用它需要同时展示 token 工具调用 检索结果模型调用被层层封装messages模式收不到做AI 正在思考…这类细粒度提示。代价事件量大处理逻辑比stream_mode复杂。简单场景别上够用就行。六、异步版本怎么选凡是带a前缀的astream、astream_events都必须在async函数里跑。如果你的服务是 FastAPI直接用异步版本如果是脚本或者同步 Web 框架用streammessages模式也够用。importasyncioasyncdefmain():asyncformode,chunkingraph.astream({messages:[{role:user,content:你好}]},stream_mode[messages,custom],):...asyncio.run(main())FastAPI 里配合 SSE 推给前端是标准做法app.get(/chat)asyncdefchat(q:str):asyncdefgen():asyncformsg,metaingraph.astream({messages:[{role:user,content:q}]},stream_modemessages,config{configurable:{thread_id:u-1}},):ifmsg.content:yieldfdata:{json.dumps({t:msg.content})}\n\nreturnStreamingResponse(gen(),media_typetext/event-stream)注意thread_id只要图挂了 checkpointer多轮对话必须传同一个 thread_id否则每轮都是新会话。七、子图的流式前面说过父图默认看不到子图内部。要看到就加subgraphsTrueforns,chunkingraph.stream({input:...},stream_modeupdates,subgraphsTrue,):levellen(ns)print( *levelstr(chunk))ns是命名空间元组长度就是嵌套层级可以直接拿来做缩进。八、常见坑汇总现象原因处理messages 模式收不到任何 token节点里用的是非流式invoke换astream_events或改节点逻辑收不到结构化输出的中间 token结构化输出必须等完整结果改用 custom 发进度每次都从头开始没有记忆没传thread_id或没挂 checkpointer补上config{configurable: {thread_id: ...}}子图内部过程看不到默认不流式子图加subgraphsTrueget_stream_writer()返回 None在节点外部调用移到节点函数内部前端收到乱码/断句错乱token 流含多字节字符被切分服务端按完整字符串发让前端拼最后一条在中文场景尤其常见一个汉字 3 个字节如果按字节切分流会出现乱码。SSE 里传完整的字符串片段就行别按字节切。九、选择决策树照这个选基本不会错只要每一步做了什么 →updates要打字机效果且 LLM 调用是流式的 →messages要自定义进度 →custom可以和上面组合要 token 工具 检索全部细节或 LLM 被深层封装 →astream_events(v2)调试 →debug或values小结values给完整快照updates给增量 —— 做进度展示默认用updatesmessages做打字机但前提是 LLM 调用本身是流式的customget_stream_writer()是最灵活的进度通道深层封装、要全部细节时用astream_events(versionv2)多轮对话别忘了thread_id子图流式别忘了subgraphsTrue。
