LangGraph LLM 输出协议（一）：后端已经在流式输出，前端为什么仍然无法渲染

“支持流式”不等于页面能正确展示

设想一次调研运行：模型先输出两段文字，随后调用检索工具，最后暂停等待审核。后端确实在持续产生 token，但浏览器还需要知道哪段文字属于哪条消息、哪个工具正在执行、引用何时可展开，以及什么时候应该显示“继续审核”。一个 LLM 调用可能返回文本、工具调用、使用量和停止原因；LangGraph 又会产生节点更新、条件分支、循环、暂停和恢复。

把模型返回值直接序列化给前端，最早能看到 token，后面却会遇到三个问题：事件形状随 SDK 版本变化，内部 State 泄露到页面，断线重连后无法判断哪些片段已经消费。流式输出必须先经过一个稳定的业务协议。

先把四层边界拆开

同一个“流式响应”其实包含四层：

flowchart LR
    A[LLM / Tool 返回] --> B[LangGraph 运行时流]
    B --> C[业务事件适配层]
    C --> D[SSE / Fetch / WebSocket]
    D --> E[前端 reducer]
    E --> F[消息、进度、工具、审批组件]

事件从 LLM / Tool 进入 LangGraph，再经业务适配层、传输和 reducer，最后成为组件可以渲染的状态。

每层的责任不同。比如 LangGraph 产生“collect_sources 节点更新了 sources”，适配层把它翻译成“已找到 8 个来源”，SSE 负责把事件送到浏览器，reducer 负责把它合并进时间线，组件最终显示“检索资料 ✓”。任何一层越界，后续演进都会变得昂贵：如果组件直接读取 collect_sources，后端改名就会变成前端改造。

先比较“处在不同层”的方案

下面的方案不能只按“是否支持流式”排一张表：

选型的第一问应该是“前端要消费什么事件”。如果页面需要文本增量、节点进度和审批，就先定义这三类事件；第二问才是“这些事件怎么传输”。流式只表示数据分批到达，不保证顺序、幂等、恢复或可渲染。

LangGraph 原生流先做版本核对

LangGraph¹ 的 graph.stream() 可以按模式输出：

较新的版本还提供 typed event streaming。写文章和写适配器时，必须以目标版本为准，核对 graph.stream_events(input, version="v3") 以及 messages、values、subgraphs、output projection 的实际接口；传统 graph.stream(..., stream_mode=..., version="v2") 使用的是另一套事件模型。不要把 astream_events（Runnable 层 API）直接当作 LangGraph 图的同名协议，也不要把两个版本的 chunk 形状混用。

后端可以同时订阅 messages 和 custom，再从 updates 推导节点状态。例如一个模型 token 应转换为 text.delta，工具完成应转换为 tool.result，人工暂停应转换为 approval.required。前端永远不应该依赖原始 chunk；适配器需要为每个版本和 mode 写小型 translator，并用协议契约测试保证输出稳定。

先固定事件 envelope，再决定协议

我们为一次资料调研运行定义一层与传输无关的 envelope：

{
  "id": "run-123:42",
  "version": 1,
  "type": "text.delta",
  "run_id": "run-123",
  "thread_id": "research-001",
  "seq": 42,
  "occurred_at": "2026-08-24T12:00:02.100Z",
  "payload": {
    "message_id": "answer-1",
    "text": "状态"
  }
}

id 负责去重，seq 负责同一运行内排序和补齐，type 负责分派，version 负责演进，run_id / thread_id 负责关联，payload 只承载事件特有数据。文案放在前端或业务映射表中，不要让组件通过“正在检索资料”这种字符串猜状态。

第一版事件可以控制在：

run.started / run.completed / run.failed / run.cancelled
node.started / node.completed / node.failed
text.start / text.delta / text.end
tool.start / tool.result
approval.required / error

每种事件要说明状态机关系。比如收到 text.delta 时，reducer 只能把文本追加到同一个 message_id；收到 tool.result 前，如果没有对应的 tool.start，组件应显示“工具结果缺少开始事件”，而不是直接渲染为成功；收到 approval.required 后，运行视图才进入暂停状态。未知事件需要安全忽略或降级显示，不能让整个页面崩溃。

后端适配器只暴露业务事件

下面的代码是一个最小边界示例。它假设 graph 已按目标 LangGraph 版本编译，生产实现应把多 mode / typed event 的解析拆成独立 translator：

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


def make_event(run_id: str, thread_id: str, seq: int, kind: str, payload=None):
    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 or {},
    }


def stream_run(graph, 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(
            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, "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",
            {"code": type(exc).__name__},
        )
        raise


def to_sse(item: dict) -> str:
    return (
        f"id: {item['id']}\n"
        f"event: {item['type']}\n"
        f"data: {json.dumps(item, ensure_ascii=False)}\n\n"
    )

示例只发送变更字段名称，不发送 patch 值。真实 translator 可以把来源数量、节点产品名称和错误分类加入白名单，但不能把完整 State 序列化。事件生成、持久化和发送的顺序也要固定，否则 seq 只是看起来有序。

SSE、Fetch 流和 WebSocket 如何选择

SSE：只推送事件时的默认方案

SSE² 使用普通 HTTP 长响应，服务端发送 event、data 和可选的 id：

id: run-123:42
event: text.delta
data: {"message_id":"answer-1","text":"状态"}

它适合“POST 创建运行，GET 持续接收结果”。取消、审批和 resume 使用独立的 POST，可以分别鉴权，也不用为了少量双向操作引入 WebSocket。原生 EventSource 只支持 GET；如果订阅需要请求体或 Authorization header，使用 fetch 读取 SSE 响应。

SSE 自动重连只解决重新建立 TCP 连接，不解决业务恢复。服务端必须根据 Last-Event-ID 或 after_seq 重放事件，否则重连后只能获取一个快照并承认中间事件无法补齐。

Fetch + ReadableStream：启动和订阅必须是一个 POST 时

如果启动运行本身返回流，或必须携带自定义请求头，前端可以读取 fetch() 的 ReadableStream：

const response = await fetch("/api/runs", {
  method: "POST",
  headers: { Authorization: `Bearer ${token}` },
  body: JSON.stringify({ topic }),
});

if (!response.body) throw new Error("响应不支持流式读取");
const reader = response.body.getReader();

Fetch 流不是新的业务协议，字节之上仍要约定 SSE 或 NDJSON 帧。浏览器不会替你实现重连、Last-Event-ID 和解析失败后的恢复，这些都由 transport 层负责。

WebSocket：双向交互足够频繁时

WebSocket³ 适合同一连接中持续发送 subscribe、cancel、resume 和人工输入：

浏览器 ── subscribe / cancel / resume ──> 服务端
浏览器 <── text.delta / tool.result / approval.required ── 服务端

它不规定 JSON 结构，也不自动提供权限、事件顺序和重放。若业务只是看进度，SSE 加独立控制接口通常更容易扩容和排查；WebSocket 的价值来自双向频繁交互，而不是“更实时”。

前端不要让组件直接读网络

推荐的消费链路是：

transport → parse → schema validate → reducer → UI components

前端视图状态可以围绕事件建模：

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

type RunView = {
  status: "idle" | "running" | "paused" | "succeeded" | "failed";
  text: string;
  nodes: Record<string, "running" | "completed" | "failed">;
  tools: Record<string, { name: string; status: string }>;
  lastSeq: number;
  error?: string;
};

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

  switch (event.type) {
    case "run.started":
      return { ...view, status: "running", lastSeq: event.seq };
    case "text.delta":
      return {
        ...view,
        text: view.text + String(payload.text ?? ""),
        lastSeq: event.seq,
      };
    case "approval.required":
      return { ...view, status: "paused", lastSeq: event.seq };
    case "run.completed":
      return { ...view, status: "succeeded", lastSeq: event.seq };
    case "run.failed":
    case "error":
      return {
        ...view,
        status: "failed",
        error: String(payload.code ?? "unknown"),
        lastSeq: event.seq,
      };
    default:
      return { ...view, lastSeq: event.seq };
  }
}

真实代码还要按事件 id 去重，并分别处理多个 message_id、工具结果晚于文本增量到达，以及节点重试产生的重复事件。组件只接收 RunView 和经过脱敏的 payload；它不应该知道 StateGraph 里的 Python 节点名。

事件如何映射到自己的 UI 组件

内部节点名要映射成产品名称，例如 collect_sources 显示为“检索资料”，但开发者详情面板可以保留原名。引用、工具参数和中间 State 默认只显示摘要，用户主动展开后再显示经过脱敏的细节。

断线恢复是协议的一部分

一个可用的最小恢复协议包括：

1. 每个事件有单调递增的 seq 和稳定 id；
2. 服务端短期保存事件，支持 after_seq 或 Last-Event-ID 重放；
3. 页面刷新先获取运行快照，再从快照序号之后订阅；
4. cancel、resume 和审批提交使用独立接口，并校验 run_id 权限；
5. 以 run.completed、run.failed 或 run.cancelled 判断结束，不以 TCP 断开判断业务结束。

如果事件没有持久化，自动重连无法补齐丢失内容；如果运行状态没有单独保存，前端也无法知道服务端是否仍在执行。传输协议只是通道，运行状态和事件存储才决定恢复能力。页面刷新时应先拿到 last_seq=12 和当前文本，再订阅序号 12 之后的事件，而不是从空白状态重新猜测。

选择一套可以先落地的组合

对于“资料调研 + LangGraph + Phoenix + 自定义页面”的业务，我会先采用：

• 后端执行：按目标版本使用 updates、messages、custom 或 typed event streaming；
• 业务协议：稳定 envelope、有限事件类型、schema 校验和版本字段；
• 传输：SSE 推送，POST 创建、取消、恢复和审批；
• 前端：transport hook、事件 reducer、消息 / 进度 / 工具 / 审批组件；
• 监控：事件的 run_id 与 Phoenix Trace 关联，但 UI 事件和 Trace 数据分开存储；
• 演进：协议版本、应用版本和组件版本都可追踪。

将来改成 WebSocket，替换的是 transport；接入 AI SDK，替换的是 message translator；页面仍然消费自己的 RunView。这条边界让 LangGraph、模型供应商和 UI 可以各自迭代。

两条协议失败路径

SDK 升级后页面突然不再显示节点

先比较目标版本的 stream / stream_events API 和原始 chunk，确认是否把 v2 与 v3 事件模型混用了。处理方式是把版本核对和 translator 契约测试放在适配层，而不是让前端兼容每一种 chunk。验证时使用固定输入，节点事件仍能映射为同一组业务 type。

页面刷新后文本重复或状态倒退

先看重复事件的 id、seq 和快照序号。若 reducer 只追加文本，重连重放就会产生两遍内容；处理方式是按 message_id + seq 去重，刷新先加载快照，再从快照之后订阅。验证时在 text.delta 中途刷新页面，最终文本只出现一次且状态到达 run.completed。

参考资料

• LangGraph Streaming（官方文档，访问日期：2026-08-24）
• LangGraph Graph API（官方文档，访问日期：2026-08-24）
• LangSmith Agent Server Protocol v2：SSE 事件流（官方文档，访问日期：2026-08-24）
• Vercel AI SDK：Stream Protocols（官方文档，访问日期：2026-08-24）
• MDN：Using server-sent events（访问日期：2026-08-24）
• MDN：WebSocket（访问日期：2026-08-24）