LangChain - 流式传输（Streaming）

一、先搞懂一件事：流式 vs 普通调用

invoke 是"等模型把整段话说完，一次性返回给你"。问题是大模型生成慢，你要干等好几秒才看到结果，体验差。

**流式传输（stream）**就是改成"模型一边生成、一边把字往你这儿送"，像打字机一样一个字一个字蹦出来，你立刻就能看到。

▎ 比喻：invoke = 等整桌菜做好才端上；stream = 火锅边涮边吃。

用法就是把 invoke 换成 stream，然后在循环里一段段接收：

forchunkinagent.stream({"messages":[...]},...):# chunk 就是模型刚吐出的一小段处理(chunk)

二、核心难点：stream_mode 是什么？

stream_mode 不是"开不开流式"，而是"你想看哪种类型的实时信息"。

一个智能体运行时，同时有好几种信息在产生：

• 进度信息：现在跑到第几步了？调工具了吗？工具返回啥了？
• 文字信息：模型正在一个字一个字地说什么？
• 自定义信息：工具干活时想主动汇报点进度（“已查到 10 条”）

你不可能一次全看，所以让你选要看哪种。这就是 stream_mode。

页面给了三种模式：

三、三种模式逐个用大白话讲

模式 1：“updates” —— 看"步骤进度"

就像看一个任务清单：智能体每完成一步，就给你发一条"这步干完了，结果是xxx"。

比如问"旧金山天气"，智能体调了一次天气工具，你会收到三条进度：

step: model → 模型说：我要调 get_weather 工具
step: tools → 工具说：旧金山天气是晴的
step: model → 模型说：最终回答——旧金山天气晴朗

代码：

forchunkinagent.stream({"messages":[{"role":"user","content":"What is the weather in SF?"}]},stream_mode="updates",version="v2",# 推荐用 v2，输出格式统一):ifchunk["type"]=="updates":forstep,datainchunk["data"].items():print(f"step:{step}")

▎ 特点：粒度粗，每步一条，不是逐字。适合"我想知道现在进行到哪了"。

模式 2：“messages” —— 看"逐字吐出"

这就是真正的打字机效果。模型生成的每一个字（token）都立刻送给你，连工具调用的参数都是一块块拼出来的（先 ‘{"’，再 ‘city’，再 ‘“:”’，再 ‘San’…）。

代码：

forchunkinagent.stream({"messages":[{"role":"user","content":"What is the weather in SF?"}]},stream_mode="messages",version="v2",):ifchunk["type"]=="messages":token,metadata=chunk["data"]# token 是那一小段，metadata 是来源信息print(token.content_blocks)

▎ 特点：粒度细，逐字逐块。适合"我想让用户看到字一个个蹦出来"。

模式 3：“custom” —— 看"你自己发的消息"

前两种是系统自动产生的信息，custom 是你自己想发什么就发什么。在工具里用 get_stream_writer() 主动喊话：

fromlanggraph.configimportget_stream_writerdefget_weather(city:str)->str:"""Get weather for a given city."""writer=get_stream_writer()writer(f"开始查{city}的天气...")# 你主动发的进度writer(f"查到了{city}的数据！")returnf"It's always sunny in{city}!"外面接收：forchunkinagent.stream(...,stream_mode="custom",version="v2"):ifchunk["type"]=="custom":print(chunk["data"])# 就会打印你 writer() 发的那些话

▎ 特点：完全自定义。适合"工具干耗时的活，想给用户报进度"。

▎ 注意：用了 get_stream_writer 的工具，不能脱离 LangGraph 执行环境单独调用。

四、能同时开多个模式吗？

能。stream_mode=[“updates”, “custom”] 这种传列表就行。

但记住：不是每种模式各把全部结果吐一遍，而是几种模式的事件按时间顺序"交织"着来，每个事件只出现一次。

就像一条河里混着三种颜色的鱼（updates、custom、messages），它们按游过来的先后顺序一条条到你面前，你用鱼的颜色（chunk[“type”]）区分是哪种：

forchunkinagent.stream(...,stream_mode=["updates","custom"],version="v2"):ifchunk["type"]=="updates":...# 处理进度elifchunk["type"]=="custom":...# 处理自定义

实际输出顺序（页面示例）就是交替的：
updates（模型要调工具）→ custom（开始查）→ custom（查到了）→ updates（工具完成）→ updates（最终回答）

五、每个块长啥样？看 type 区分

不管哪种模式，v2 格式下每个块都是这种结构：

{"type":"updates"或"messages"或"custom",# 这块是哪种模式来的"data":...# 具体内容}

所以循环里永远先看 chunk[“type”]，判断这是哪类信息，再去取 chunk[“data”]。这是看懂流式代码的万能套路。

六、几个进阶用法（知道有这些就行）

1. 看"模型的思考过程"

有些模型（如 Claude）回答前会先"思考"。用 stream_mode=“messages”，然后从内容块里挑出 type==“reasoning” 的就是思考过程，type==“text” 的才是正式回答：

fortoken,metadatainagent.stream(...,stream_mode="messages"):reasoning=[bforbintoken.content_blocksifb["type"]=="reasoning"]text=[bforbintoken.content_blocksifb["type"]=="text"]ifreasoning:print(f"[思考]{reasoning[0]['reasoning']}")iftext:print(text[0]["text"])

▎ 不管 Anthropic 的 thinking 还是 OpenAI 的 reasoning，LangChain 都统一成 “reasoning” 块，换厂商代码不用改。

2. 既看逐字、又看完整的工具调用

messages 给你的是"半个半个的 JSON 片段"，你想看"拼好的完整工具调用"就再配上 updates：

stream_mode=[“messages”, “updates”]
messages 看逐字片段，updates 看拼好的完整结果

3. 多智能体时区分"谁在说话"

给每个智能体起 name，流式时开 subgraphs=True，用 metadata[“lc_agent_name”] 就知道当前是哪个智能体在吐字：

weather_agent=create_agent(...,name="weather_agent")supervisor=create_agent(...,name="supervisor",tools=[call_weather_agent])forchunkinsupervisor.stream(...,stream_mode=["messages","updates"],subgraphs=True):ifagent_name:=metadata.get("lc_agent_name"):

4. 人工审批中断

某些工具调用前先暂停问人（approve/edit），用 HumanInTheLoopMiddleware + checkpointer。流式时收集中断 → 给决策 → 用 Command(resume=decisions) 恢复：

agent=create_agent(...,middleware=[HumanInTheLoopMiddleware(interrupt_on={"get_weather":True})],checkpointer=...)# 第一次流式：会中断，收集 interrupts# 给出决策后，用 Command 恢复：agent.stream(Command(resume=decisions),config=config,...)

七、两个小开关

• version=“v2”：强烈推荐。让所有模式的输出格式统一成 {type, data}，不用再分情况解包。需要 LangGraph ≥ 1.1。
• 禁用流式：某模型不想流式，设 streaming=False（或 disable_streaming=True）。多智能体里可用来控制"只让某个智能体流式"。

八、一句话总结

▎ 流式传输 = 把 invoke 换成 stream，让结果逐段实时送出。stream_mode 决定"看哪种信息"：updates 看步骤进度、messages 看逐字吐出、custom 看自己发的消息。
▎可多选，事件交织着来，每个块靠 chunk[“type”] 区分。

九、看懂流式代码的万能三步

不管页面里哪段代码，都套这个套路：

1. 看 stream_mode：他开了哪些模式？
2. 循环里看 chunk[“type”]：这块是哪个模式来的？
3. 取 chunk[“data”]：这个模式的实际内容是什么？

记住这三步，所有例子都能看懂。

十、和前面学的串起来

• 流式拿到的"字"就是 AIMessageChunk（消息页讲过），可以累加拼成完整 AIMessage。
• content_blocks 把各厂商的思考/推理统一成 “reasoning” 块（消息页讲过）。
• get_stream_writer() 就是工具页讲的 runtime.stream_writer 的全局版本。
• 人工审批用的 checkpointer + thread_id 是短期记忆页讲的那套。