📚 AI 工程与智能体教程站

简体中文

English

简体中文

English

第 7 章 · 流式输出与前端事件协议 ​

把运行中的状态增量逐步交给界面，而不是等整张图完成。先用不调用模型的例子区分 values、updates 和 custom，再考虑 token 流。

工作机制与设计选择 ​

本例显式使用 version="v2"，每个事件都包含 type、ns、data，要求 LangGraph 1.1+。updates 仅包含节点更新；values 是状态快照；custom 是节点主动发出的业务进度。messages 提供模型消息片段与元数据，不是任意节点的字符串输出。当前官方推荐新应用考虑 1.2 的 event streaming；不要把 stream 的 v2 格式与 stream_events 的版本混为一谈。

示例 ​

python
from typing import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.config import get_stream_writer

class State(TypedDict):
    topic: str
    report: str

def report(state: State):
    writer = get_stream_writer()
    writer({"stage": "collect", "completed": 1, "total": 2})
    writer({"stage": "format", "completed": 2, "total": 2})
    return {"report": f"Report: {state['topic']}"}

builder = StateGraph(State)
builder.add_node("report", report)
builder.add_edge(START, "report")
builder.add_edge("report", END)
graph = builder.compile()
parts = list(graph.stream(
    {"topic": "local fixtures", "report": ""},
    stream_mode=["updates", "custom"], version="v2",
))
assert len([p for p in parts if p["type"] == "custom"]) == 2
for part in parts:
    print(part["type"], part["ns"], part["data"])

运行与验收 ​

保存并运行，应该看到两条 custom 进度和 report 节点的 updates。测试只约束本章承诺的事件，不依赖内部调度日志数量。界面传输时再把这些事件包装为 SSE 或 WebSocket 消息。

容易踩的坑 ​

状态中可能有用户输入、检索原文和内部字段，不能把所有 values 原样发送给客户端。断开浏览器连接不等于服务端任务必然取消；需要独立定义取消和恢复协议。

🧪 随堂测验

点击你认为正确的选项。答错时会展示正确答案与原因解析。

1. updates 包含什么？

2. v2 用什么区分事件？

3. 普通节点返回字符串就是 messages 吗？

4. 可直接公开全部状态吗？

动手实践 ​

  1. 只显示阶段与完成比例，过滤其余字段。
  2. 加入子图，检查 ns 如何区分事件来源。
  3. 制定断线重连策略，避免界面重复显示同一业务事件。

官方参考 ​

核验日期：2026-09-17。