LangGraph 11：流式输出

发表于 分类于
讲解 stream/astream 与 stream_mode（values、updates、messages、custom），以及 stream_writer 与 astream_events。

前言

invoke 只在整图结束后返回最终状态，长链路里用户会感觉「卡住」。
LangGraph 提供 stream / astream：图跑到一半就把中间结果推出来。
stream_mode 决定你看见的是完整状态、节点增量、LLM token，还是自定义进度文案。
本文覆盖常用模式与 stream_writer，并简述次要 API astream_events。
涉及模型的示例对接 火山方舟 Coding Plan，模型用 ark-code-latest。
下文需要 Python 3.12+，依赖用 uv 管理。

依赖

建议使用 Python 3.12 及以上。

1
2
3
4
uv init langgraph-streaming
cd langgraph-streaming
uv venv --python 3.12
uv add "langgraph>=1.0,<2.0" "langchain>=1.0,<2.0" langchain-openai python-dotenv rich

配置 Coding Plan 的 .env 后，可用 uv run python demo.py 跑示例。

values 与 updates

values：每次推送当前累积状态（适合刷新整页状态）。
updates：只推本步节点写出的增量（适合日志与 UI 差量）。
可一次传多个 mode：stream_mode=["updates", "values"]。
部分版本可用 version="v2" 统一 chunk 形状，按你安装的 langgraph 文档为准。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
import time
from typing import TypedDict

from langgraph.graph import END, START, StateGraph
from rich import print as rprint

class OverAllState(TypedDict):
    initial_state: str
    node_a_output: str
    node_b_output: str

def node_a(state: OverAllState) -> OverAllState:
    time.sleep(0.2)
    return {"node_a_output": "节点A的输出"}

def node_b(state: OverAllState) -> OverAllState:
    time.sleep(0.2)
    return {"node_b_output": "节点B的输出"}

builder = StateGraph(state_schema=OverAllState)
builder.add_node("node_a", node_a)
builder.add_node("node_b", node_b)
builder.add_edge(START, "node_a")
builder.add_edge("node_a", "node_b")
builder.add_edge("node_b", END)
graph = builder.compile()

for chunk in graph.stream(
    {"initial_state": "初始状态"},
    stream_mode=["updates", "values"],
):
    rprint(chunk)

异步场景把 stream 换成 astream，用 async for 消费即可，mode 含义相同。

messages

当状态含 messages 且节点调用聊天模型时，stream_mode="messages"（或列表中的 "messages"）会推送 LLM 的流式分片。
适合聊天 UI 的「逐字显示」。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
import os

from dotenv import load_dotenv
from langchain.chat_models import init_chat_model
from langchain.messages import HumanMessage
from langgraph.graph import END, START, MessagesState, StateGraph
from rich import print as rprint

load_dotenv()

model = init_chat_model(
    "openai:ark-code-latest",
    temperature=0,
    api_key=os.environ["OPENAI_API_KEY"],
    base_url=os.environ["OPENAI_BASE_URL"],
)

def llm_node(state: MessagesState) -> MessagesState:
    return {"messages": [model.invoke(state["messages"])]}

builder = StateGraph(state_schema=MessagesState)
builder.add_node("llm_node", llm_node)
builder.add_edge(START, "llm_node")
builder.add_edge("llm_node", END)
graph = builder.compile()

for chunk in graph.stream(
    {"messages": [HumanMessage(content="你好，用一句话自我介绍")]},
    stream_mode=["values", "messages"],
):
    rprint(chunk)

若只关心 token，可只开 messages，再在循环里取 chunk 里的文本字段拼到前端。

custom 与 stream_writer

节点签名增加 runtime: Runtime 后，可用 runtime.stream_writer(...) 推自定义字符串或结构化进度。
消费端必须带 stream_mode=["custom"]（可与其它 mode 并列），否则看不到这些推送。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
import time
from typing import TypedDict

from langgraph.graph import END, START, StateGraph
from langgraph.runtime import Runtime
from rich import print as rprint

class OverAllState(TypedDict):
    initial_state: str
    node_a_output: str
    node_b_output: str

def node_a(state: OverAllState, runtime: Runtime) -> OverAllState:
    runtime.stream_writer("节点 A 正在执行...")
    time.sleep(0.2)
    return {"node_a_output": "节点A的输出"}

def node_b(state: OverAllState, runtime: Runtime) -> OverAllState:
    runtime.stream_writer("节点 B 正在执行...")
    time.sleep(0.2)
    return {"node_b_output": "节点B的输出"}

builder = StateGraph(state_schema=OverAllState)
builder.add_node("node_a", node_a)
builder.add_node("node_b", node_b)
builder.add_edge(START, "node_a")
builder.add_edge("node_a", "node_b")
builder.add_edge("node_b", END)
graph = builder.compile()

for chunk in graph.stream(
    {"initial_state": "初始状态"},
    stream_mode=["custom"],
):
    rprint(chunk)

在 ToolRuntime 里同样有 stream_writer，适合工具执行时推送「正在查天气…」一类提示。

其它 mode 简述

带 checkpointer 时还可试 debug / checkpoints / tasks 等模式，观察任务与快照，偏调试用途。
与 HITL 组合时，可对 Command(resume=...) 再 stream，中断前后的状态变化也会以流形式出现。
具体字段名随版本可能微调，以当前文档为准。

astream_events

astream_events 是更细粒度的事件流（节点开始/结束、模型回调等），适合可观测性与复杂编排调试。
日常聊天与状态刷新优先 stream + stream_mode；需要「事件总线」级细节时再用它。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
from typing import TypedDict

from langgraph.graph import END, START, StateGraph
from rich import print as rprint

class OverAllState(TypedDict):
    initial_state: str
    node_a_output: str
    node_b_output: str

def node_a(state: OverAllState) -> OverAllState:
    return {"node_a_output": "节点A的输出"}

def node_b(state: OverAllState) -> OverAllState:
    return {"node_b_output": "节点B的输出"}

builder = StateGraph(state_schema=OverAllState)
builder.add_node("node_a", node_a)
builder.add_node("node_b", node_b)
builder.add_edge(START, "node_a")
builder.add_edge("node_a", "node_b")
builder.add_edge("node_b", END)
graph = builder.compile()

async def main():
    async for chunk in graph.astream_events(
        {"initial_state": "初始状态"},
        version="v2",
    ):
        rprint(chunk)

用 asyncio.run(main()) 或 Jupyter 的顶层 await 运行即可。

总结

  1. 产品面优先：updates（差量）、values（全量）、messages（token）、custom（进度文案）。
  2. 自定义进度必须在节点里调 stream_writer，并在 stream_mode 中包含 custom。
  3. 异步用 astream；事件级调试再用 astream_events。
  4. 与 checkpointer / HITL 组合时，对流式消费保持同一 thread_id。