文章目录
- LangGraph 流式输出：stream_mode 到底该选哪个
- 一、先建立一张对照表
- 二、values 和 updates 的区别，看一次输出就懂
- 三、打字机效果：messages 模式
- 四、custom 模式：想推什么推什么
- 五、astream_events：什么都能流式
- 六、异步版本怎么选
- 七、子图的流式
- 八、常见坑汇总
- 九、选择决策树
- 小结
LangGraph 流式输出：stream_mode 到底该选哪个
用户问一句，界面上要等 20 秒才"啪"地弹出整段回答——
这个体验在今天已经不能接受了。LangGraph 提供了好几种流式模式，名字长得都差不多，
这篇把它们挨个讲清楚，并给出选择依据。
一、先建立一张对照表
graph.stream(...)/graph.astream(...)的stream_mode参数决定"你会收到什么"。
| 模式 | 每次产出什么 | 典型用途 |
|---|---|---|
"values" | 完整 state 的快照 | 调试、需要拿到全部字段 |
"updates" | 本节点新增/修改的部分 | 最常用，展示"这一步做了什么" |
"messages" | LLM 的token 增量 + 节点元数据 | 打字机效果 |
"custom" | 你自己get_stream_writer()写的任意内容 | 自定义进度条、中间提示 |
"debug" | 极详细的内部信息 | 排查问题 |
"checkpoints" | 每次写入检查点的事件 | 做持久化 UI、断点恢复 |
"tasks" | 任务的开始/结束事件 | 并行分支进度 |
模式可以传列表：stream_mode=["updates", "custom"]，
这样每步会产出(mode, chunk)二元组。
二、values 和 updates 的区别，看一次输出就懂
假设图是START → step_a → step_b → END，step_a写{"foo": 1}，step_b写{"bar": 2}。
stream_mode="values"：
forchunkingraph.stream({"input":"x"},stream_mode="values"):print(chunk)# {'input': 'x'}# {'input': 'x', 'foo': 1}# {'input': 'x', 'foo': 1, 'bar': 2}每次都是完整的、累加后的 state。字段越攒越多。
stream_mode="updates"：
# {'step_a': {'foo': 1}}# {'step_b': {'bar': 2}}每次只给你这一步改了什么，并且带上了节点名。
实践结论：
- 要做"进度展示"（这一步在检索、那一步在生成）→ 用
updates； - 要拿某个字段的最新值 → 用
values更省事，不用自己累积； - 生产环境做前端流式，我默认选
updates，因为带节点名，方便做步骤高亮。
三、打字机效果：messages 模式
想让用户看到模型一个字一个字往外蹦，用messages：
fromlangchain_core.messagesimportAIMessageChunkformsg,metadataingraph.stream({"messages":[{"role":"user","content":"讲个笑话"}]},stream_mode="messages",):ifisinstance(msg,AIMessageChunk):print(msg.content,end="",flush=True)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):writer=get_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"],):ifmode=="custom":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":"查一下天气"}]},version="v2",):kind=event["event"]ifkind=="on_chat_model_stream":print(event["data"]["chunk"].content,end="",flush=True)elifkind=="on_tool_start":print(f"\n[调用工具]{event['name']}参数={event['data'].get('input')}")elifkind=="on_tool_end":print(f"\n[工具返回]{event['data'].get('output')}")version="v2"是必须的，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 框架，用stream+messages模式也够用。
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_mode="messages",config={"configurable":{"thread_id":"u-1"}},):ifmsg.content:yieldf"data:{json.dumps({'t':msg.content})}\n\n"returnStreamingResponse(gen(),media_type="text/event-stream")注意thread_id：只要图挂了 checkpointer，多轮对话必须传同一个 thread_id，
否则每轮都是新会话。
七、子图的流式
前面说过，父图默认看不到子图内部。要看到就加subgraphs=True：
forns,chunkingraph.stream({"input":"..."},stream_mode="updates",subgraphs=True,):level=len(ns)print(" "*level+str(chunk))ns是命名空间元组，长度就是嵌套层级，可以直接拿来做缩进。
八、常见坑汇总
| 现象 | 原因 | 处理 |
|---|---|---|
| messages 模式收不到任何 token | 节点里用的是非流式invoke | 换astream_events，或改节点逻辑 |
| 收不到结构化输出的中间 token | 结构化输出必须等完整结果 | 改用 custom 发进度 |
| 每次都从头开始，没有记忆 | 没传thread_id或没挂 checkpointer | 补上config={"configurable": {"thread_id": ...}} |
| 子图内部过程看不到 | 默认不流式子图 | 加subgraphs=True |
get_stream_writer()返回 None | 在节点外部调用 | 移到节点函数内部 |
| 前端收到乱码/断句错乱 | token 流含多字节字符被切分 | 服务端按完整字符串发，让前端拼 |
最后一条在中文场景尤其常见：一个汉字 3 个字节，
如果按字节切分流，会出现乱码。SSE 里传完整的字符串片段就行，别按字节切。
九、选择决策树
照这个选基本不会错：
- 只要"每一步做了什么" →
updates - 要打字机效果，且 LLM 调用是流式的 →
messages - 要自定义进度 →
custom（可以和上面组合） - 要 token + 工具 + 检索全部细节，或 LLM 被深层封装 →
astream_events(v2) - 调试 →
debug或values
小结
values给完整快照，updates给增量 —— 做进度展示默认用updates；messages做打字机，但前提是 LLM 调用本身是流式的；custom+get_stream_writer()是最灵活的进度通道；- 深层封装、要全部细节时用
astream_events(version="v2")； - 多轮对话别忘了
thread_id，子图流式别忘了subgraphs=True。
