stream与stream_mode

前端要实时看到「哪个节点跑完了、state 变了什么」，而不是等整图 invoke 结束。stream 按 stream_mode 推送事件块，便于 SSE/WebSocket 对接。

1. 定位

维度	内容
角色	增量观测与流式 UX
输入 → 输出	`graph.stream(input, stream_mode=...)` → 迭代事件
核心 API	`stream`、`stream_mode`
依赖 LangChain	messages 模式依赖 BaseMessage

2. 图拓扑

节点表

节点名	职责	读 State	写 State
`step1`	第一步	`count`	`count`, `trace`
`step2`	第二步	`count`	`count`, `trace`

边表

源	目标	类型
START	step1	固定
step1	step2	固定
step2	END	固定

3. invoke / stream 生命周期

stream_mode="updates":
  事件1: {"step1": {"count": 1, "trace": ["s1"]}}
  事件2: {"step2": {"count": 2, "trace": ["s2"]}}

stream_mode="values":
  事件1: 全量 state 快照 after step1
  事件2: 全量 state 快照 after step2

每个事件对应一个 superstep 结束后的输出。

4. 原理

4.1 updates vs values

updates：仅本步节点 partial update，体积小
values：合并后完整 state，适合 UI 全量渲染

4.2 messages 模式

专用于消息流（配合 LLM token 流时需 ChatModel streaming）。

4.3 与 astream

异步服务用 astream（见异步专篇）。

5. 最小可运行示例

import operator
from typing import Annotated, TypedDict

from langgraph.graph import StateGraph, START, END


class State(TypedDict):
    count: int
    trace: Annotated[list[str], operator.add]


def step1(_: State) -> dict:
    return {"count": 1, "trace": ["s1"]}


def step2(state: State) -> dict:
    return {"count": state["count"] + 1, "trace": ["s2"]}


builder = StateGraph(State)
builder.add_node("step1", step1)
builder.add_node("step2", step2)
builder.add_edge(START, "step1")
builder.add_edge("step1", "step2")
builder.add_edge("step2", END)

graph = builder.compile()

print("--- updates ---")
for ev in graph.stream({"count": 0, "trace": []}, stream_mode="updates"):
    print(ev)

print("--- values ---")
for ev in graph.stream({"count": 0, "trace": []}, stream_mode="values"):
    print(ev["count"], ev["trace"])

6. 执行追踪

updates 模式

事件序	内容	累积 trace
1	step1: count=1	[“s1”]
2	step2: count=2	[“s1”,”s2”]

重要配置参数

参数	类型 / 默认	作用与影响	参考起点	配置指导
`stream_mode="updates"`	str	每节点 delta	进度条	SSE 友好
`stream_mode="values"`	str	全量 state	表单 UI	较大 payload
`stream_mode="messages"`	str	消息 token	Chat Agent	需 LLM stream
`subgraphs=True`	bool	含子图事件	嵌套图	08 专篇
`stream(..., config)`	dict	thread 一致	多轮	同 invoke
`astream`	async	FastAPI	13 专篇	IO 密集