程云来 / 杭州
Back to Blog·LangGraph
·About 14 min

LangGraph 过程监控（一）：过程监控不是把 State 推到页面

从一个只有旋转图标的运行页面出发，区分用户进度、执行诊断和内容流，设计可排序、可重连、可脱敏的事件协议。

一个旋转图标掩盖了三种完全不同的问题

设想页面上的唯一反馈是一个 spinner。十秒过去，用户不知道助手是在检索、整理证据，还是被某个上游服务卡住；开发者也不知道节点是否已经重试。更糟的是，spinner 可能一直转到请求超时，用户却没有任何可以操作的入口。

把完整 State 直接返回浏览器看似省事，实际上会同时带来信息泄露和版本耦合：State 可能包含系统提示词、工具参数、检索原文和访问令牌，字段名也会随着后端重构变化。用户需要的是“正在做什么”和“是否需要我决定”，不是内部对象的转储。

先建立产品可见性模型，再选择传输方式：事件需要稳定、可排序、可去重，并能在断线后补齐。

先把信息分成三层

一次 LangGraph1 运行会产生三类信息。以“检索 8 个来源并生成草稿”为例，同一个运行可能同时产生：“正在检索资料”（用户进度）、collect_sources 重试 2 次（执行诊断）、“模型已生成 120 个 token”（内容流）。它们的消费者不同：

信息层回答的问题适合显示给谁
业务进度流程走到哪一步，下一步是什么最终用户
执行诊断哪个节点慢、重试几次、为什么失败开发 / 运维
内容流模型文本、工具结果、引用如何增量出现用户或受控调试面板

“正在检索官方资料”属于业务进度；collect_sources 超时两次属于执行诊断；模型正在生成草稿属于内容流。三者混成一条文案，用户无法决定下一步，开发者也无法定位失败。

三层信息应在适配层分流，而不是让同一个事件同时承担三种语义：

正在绘制图表…
查看 Mermaid 源码
flowchart LR
    runtime[LangGraph 运行事件] --> adapter[业务事件适配层]
    adapter --> progress[业务进度事件]
    adapter --> diagnostics[执行诊断事件]
    adapter --> content[内容流事件]
    progress --> user[用户界面]
    diagnostics --> ops[开发 / 运维面板]
    content --> chat[消息 / 工具组件]

Mermaid 大图

可滚动查看图表，点击缩放比例可恢复 100%。按 Esc 关闭。

面向用户的事件只需要回答四件事：当前状态、当前步骤、已经完成的事实、是否需要操作。比如页面可以显示“已完成：拆分问题”“当前：检索资料，已找到 8 个来源”“下一步：等待审核”；如果上游超时，则显示“检索失败，可重试”，而不是继续转圈。可以显示节点的产品名称、成功 / 失败状态、来源数量、审核材料和可重试动作；不应显示完整 State、内部函数名、未脱敏的网页正文或没有真实依据的百分比。

LangGraph 的流是运行时输出，不是产品协议

LangGraph 的 graph.stream() 可以按 updates、values、messages、custom 等模式输出执行信息。它们的粒度不同：

模式主要内容过程监控中的位置
updates节点返回的 State 更新推导节点开始 / 完成和变更字段
values每一步之后的完整 State后端调试或快照，不直接下发
messages模型消息块及 metadata需要逐 token 展示时使用
custom节点主动发送的业务数据“已找到 8 个来源”等明确进度
debug更完整的运行诊断受控开发面板

如果使用较新的 typed event streaming，先按目标版本核对 graph.stream_events(input, version="v3") 以及 messages、values、subgraphs、output 等 projection 的真实定义。传统 stream(..., stream_mode=..., version="v2") 与 typed event streaming 不是同一种事件模型，不能把两个版本的示例拼到一个适配器里。

后端应该选择需要的投影，再把它翻译成业务事件。例如 updates 中的 {"collect_sources": {"sources": [...]}} 可以只转换成 node.completed + count=8；values 中的完整 State 则只用于服务端调试。LangGraph 的原始 chunk 是实现细节；前端依赖它，意味着下一次升级会同时影响传输、状态管理和组件。

先固定一层与传输无关的事件

一个最小 envelope 可以这样设计：

json
{
  "id": "run-123:42",
  "version": 1,
  "type": "node.completed",
  "run_id": "run-123",
  "thread_id": "research-001",
  "seq": 42,
  "occurred_at": "2026-08-24T10:30:12.412Z",
  "payload": {
    "node": "collect_sources",
    "label": "检索资料",
    "status": "succeeded",
    "count": 8
  }
}

字段不要互相代偿：id 用于去重，seq 用于同一运行内排序和补齐，type 用于分派，version 用于协议演进，run_id / thread_id 用于关联运行，payload 只放通过白名单的业务字段。比如前端收到 node.completed 后更新时间线，收到 run.paused 后展示审核按钮；文案可以变化，事件类型和状态机不能靠“正在检索”这样的字符串解析。

第一版事件不需要很多：

text
run.started
node.started
node.progress
node.completed
node.failed
run.paused
run.completed
run.failed
run.cancelled
heartbeat

不要要求每个节点报告百分比。模型生成、人工等待和外部搜索通常无法提供真实总量；伪造一个“78%”只会制造错误预期。无法量化时，发送阶段、已完成数量或等待原因。

后端适配器要做白名单和生命周期

下面的代码只演示事件边界，假设 graph 已按项目版本编译。它不把 State 值直接发送给浏览器，而是把节点名和变更字段转换为摘要：

python
from collections.abc import Iterator
from datetime import datetime, timezone


def make_event(run_id: str, thread_id: str, seq: int, kind: str, **payload):
    return {
        "id": f"{run_id}:{seq}",
        "version": 1,
        "type": kind,
        "run_id": run_id,
        "thread_id": thread_id,
        "seq": seq,
        "occurred_at": datetime.now(timezone.utc).isoformat(),
        "payload": payload,
    }


def monitor_graph(graph, input_state, config, run_id: str) -> Iterator[dict]:
    thread_id = config["configurable"]["thread_id"]
    seq = 0
    yield make_event(run_id, thread_id, seq, "run.started")

    try:
        for update in graph.stream(
            input_state,
            config=config,
            stream_mode="updates",
        ):
            for node, patch in update.items():
                seq += 1
                yield make_event(
                    run_id,
                    thread_id,
                    seq,
                    "node.completed",
                    node=node,
                    status="succeeded",
                    changed_fields=list(patch.keys()),
                )
        yield make_event(run_id, thread_id, seq + 1, "run.completed")
    except Exception as exc:
        yield make_event(
            run_id,
            thread_id,
            seq + 1,
            "run.failed",
            error_code=type(exc).__name__,
            retryable=False,
        )
        raise

教学实现把节点完成当作事件。生产实现还要处理节点重试、暂停、取消、心跳、typed event streaming 的版本差异，以及事件落库后再发送的顺序。例如节点第一次失败、第二次成功时，前端应看到“重试中 → 已完成”；如果只收到两条 node.completed，用户会误以为流程执行了两遍。每个事件都要经过字段白名单、长度限制和脱敏；changed_fields 可以公开，完整 patch 通常不应该公开。

SSE、Fetch 流和 WebSocket 的分工

SSE2 是服务器到浏览器的单向 HTTP 事件流。它适合“运行已经创建，页面持续接收进度”：

text
POST /api/runs                    创建运行，返回 run_id / thread_id
GET  /api/runs/{run_id}/events   建立 SSE，接收事件
POST /api/runs/{run_id}/cancel   取消运行
POST /api/runs/{run_id}/resume   提交人工决定

SSE 帧可以携带 id 和 event，浏览器重连时服务端根据 Last-Event-ID 重放缺失事件。原生 EventSource 只建立 GET 连接，若订阅需要 POST body 或自定义 Authorization header，应使用 fetch 读取响应流，或者由客户端 transport 封装。

WebSocket3 适合在同一连接中高频发送 subscribe、cancel、resume 和人工输入。它不会自动提供事件顺序、重放、权限或幂等；断线后仍需根据 run_id 和 seq 补齐。

场景默认选择必须额外实现
只展示运行进度SSE事件持久化、重连、鉴权
启动请求是 POST，需自定义请求头Fetch 流 / SSE分帧、超时、重试
高频双向控制和人工输入WebSocket心跳、重连、顺序、权限
断线后继续显示三者均可快照、seq、事件重放

传输协议只决定字节如何到达，不能替代运行状态。页面刷新后仍应有一个 GET /api/runs/{id} 快照接口：先拿到“当前节点=检索、已完成=拆题、last_seq=12”，再从序号 12 之后订阅事件。这样即使漏掉一条中间消息，页面也不会从头猜测运行状态。

前端用 reducer 消费事件

网络层、解析层、状态层和组件层应保持边界：

text
transport → parse → validate → reduce → UI components

一个最小的运行视图可以这样建模：

typescript
type RunEvent = {
  id: string;
  type: string;
  seq: number;
  payload: Record<string, unknown>;
};

type RunView = {
  status: "idle" | "running" | "paused" | "succeeded" | "failed";
  currentNode?: string;
  completedNodes: string[];
  lastSeq: number;
  error?: string;
};

function reduceRun(view: RunView, event: RunEvent): RunView {
  if (event.seq <= view.lastSeq) return view;

  switch (event.type) {
    case "run.started":
      return { ...view, status: "running", lastSeq: event.seq };
    case "node.completed":
      return {
        ...view,
        status: "running",
        currentNode: String(event.payload.node),
        completedNodes: [...view.completedNodes, String(event.payload.node)],
        lastSeq: event.seq,
      };
    case "run.paused":
      return { ...view, status: "paused", lastSeq: event.seq };
    case "run.completed":
      return { ...view, status: "succeeded", lastSeq: event.seq };
    case "run.failed":
      return {
        ...view,
        status: "failed",
        error: String(event.payload.error_code ?? "unknown"),
        lastSeq: event.seq,
      };
    default:
      return { ...view, lastSeq: event.seq };
  }
}

示例省略了 RunEvent 的 schema 校验，生产实现要按 id 去重，并处理节点重试产生的重复事件。组件只接收 RunView，不直接读取 LangGraph chunk；这样后端改成 WebSocket 或更换图版本时，页面不必一起重写。

两条过程监控失败路径

连接断了，页面显示停在“检索中”

先看服务端是否继续产生事件、最后持久化的 seq 和客户端的 Last-Event-ID。如果事件没有落库，SSE 自动重连无法补齐；处理方式是让页面先拉运行快照，再从快照序号订阅，或者明确显示“无法补齐历史进度”。验证时在第 2 个节点主动断开浏览器连接，恢复后页面只能出现一份节点记录。

重连后时间线出现两遍

先比较重复事件的 id 和 seq。如果同一事件被重复追加，说明 reducer 只做了数组 append；处理方式是维护 lastSeq 和事件 ID 集合，并把事件落库 / 发送顺序固定下来。验证时重复建立两次订阅，最终 completedNodes 不应出现重复节点。

监控和 UI 协议的边界

过程监控首先服务于“用户知道运行处于什么状态”，事件因此要稳定、少量、可解释。文本生命周期、工具卡片、引用和审批等细节，也应沿用同一条边界：页面事件只表达组件真正需要的语义，不把内部事件全部暴露出去。

参考资料

Footnotes

  1. LangGraph 官方文档：提供带状态、节点和流式输出的图执行运行时。 ↩

  2. MDN：Server-sent events：服务器通过 HTTP 向浏览器单向推送事件的机制。 ↩

  3. MDN：WebSocket：浏览器与服务器之间的双向长连接协议。 ↩