第 6 章 · 流式输出与实时反馈

🎯 本章目标

理解为什么流式是 LLM 应用的刚需：用 TTFT（首字延迟）这个数字把"体感"量化
分清 LangGraph 的几种 streamMode：values / updates / messages / custom（以及多模式合并）
用 values 对比 updates，看清"每次吐出来的是全量状态还是增量更新"
用 messages 模式做出 token 级打字机，并用时间戳证明是"逐 token 到达"而非一次性拿到
用 getWriter() 在节点内部上报自定义进度事件（进度条、子任务完成……）
把 streamMode 写成数组同时订阅多个事件流，并认准 [模式名, 数据] 这个二元组
亲手把图的事件翻译成 SSE 帧，知道真实前端该怎么接（EventSource / useStream / Agent Chat UI）

本章代码

📌 代码约定：正文的代码块都保持可直接运行（含 import 与模型初始化）。若某段为聚焦概念的节选，会明确标注「节选」并指向同名的 code/ 完整文件。所有示例的输出都是真实运行结果，不是编造的示例输出。

文件	内容
`code/01-stream-values-vs-updates.ts`	`values` 与 `updates` 对比：全量快照 vs 增量更新
`code/02-stream-tokens.ts`	`messages` 模式做 token 级打字机 + 到达时间证明
`code/03-stream-custom-progress.ts`	`custom` 模式：节点内 `getWriter()` 上报进度条
`code/04-stream-multi-modes.ts`	同时订阅 `updates` + `messages` + `custom`
`code/05-stream-sse-protocol.ts`	把事件翻译成 SSE 帧，实测 TTFT

一、为什么流式是刚需：先看一个数字

前面几章我们一直用 await graph.invoke(...) 拿最终结果，它的问题只有一个字：等。推理型模型答一句话往往要 1～3 秒，这段时间 invoke() 什么都不给你——界面一片空白，用户以为"卡死了"。

流式（streaming）做的事只有一件：把"一大块结果"拆成"一小口一小口"实时送出去。可以用一个指标来量化——TTFT（Time To First Token，首字延迟）：非流式 invoke() 下用户要熬完整个生成过程才看到第一个字，流式则几百毫秒就有反馈。

本机实测（示例 5）：首帧 162ms 就到达，整段要 1899ms 才结束——差了一个数量级。用户在这 1.7 秒里看到的是"文字正在长出来"，而不是"页面死了"。

💡 一句话：流式不改变总耗时，它改变的是**"用户什么时候开始觉得系统活着"**。

二、streamMode 全景图

LangGraph 把"流什么"拆成了若干模式（mode），通过 graph.stream(input, { streamMode }) 选择。本章重点讲前四种：

streamMode	每次给你什么	用在哪
`"values"`	完整状态快照（当前所有字段）	想整体观察状态演化 / 简单镜像
`"updates"`	增量：`{ 节点名: 该节点返回值 }`	想知道"哪一步跑完了、写了什么"
`"messages"`	token 级增量 `[消息增量, 元数据]`	聊天界面的打字机效果
`"custom"`	你自己 `getWriter()` 推出来的任意数据	业务进度条、阶段提示
`"debug"`	框架内部的详细事件（排查用）	调试
`"tasks"` / `"checkpoints"`	任务 / 检查点事件	高级场景（第 4 章持久化相关）

关键区分：values / updates 是框架帮你产的（图每走一步就产一次）；messages 是模型帮你产的（节点里一调模型就有 token 流）；custom 是你自己产的（不主动推就没有）。三者不是互斥的——streamMode 可以写成数组同时订阅多个（示例 4）。

三、values vs updates：全量 vs 增量

示例 1：同一张图，两种模式

typescript

/**
 * 示例 1：streamMode "values" 与 "updates" 对比
 *
 * 同一张图，分别用两种模式 stream，观察每次产出到底给的是什么：
 *   - values  ：每次给【完整状态】（当前所有字段的快照）
 *   - updates ：每次只给【这一步哪个节点写了什么】（增量）
 *
 * 这是理解 LangGraph 流式的第一块基石：先搞清"每次吐出来的是什么"。
 */
import { StateGraph, START, END, Annotation } from "@langchain/langgraph";
import { banner } from "../../lib/model.js";

// ── 状态：三个字段，分别由三个节点写入 ────────────────────────
const JobState = Annotation.Root({
  topic: Annotation<string>, // 输入：一个话题
  outline: Annotation<string>, // 节点 A 写：大纲
  draft: Annotation<string>, // 节点 B 写：草稿
  polished: Annotation<string>, // 节点 C 写：润色稿
});

// ── 节点 A：拆大纲 ───────────────────────────────────────────
async function makeOutline(state: typeof JobState.State) {
  return { outline: `【${state.topic}】的写作大纲：1) 是什么  2) 为什么  3) 怎么用` };
}

// ── 节点 B：写草稿 ───────────────────────────────────────────
async function writeDraft(state: typeof JobState.State) {
  return { draft: `草稿（基于大纲）：${state.outline} —— 已展开成三段。` };
}

// ── 节点 C：润色 ─────────────────────────────────────────────
async function polish(state: typeof JobState.State) {
  return { polished: `润色稿：${state.draft} 让语气更口语化。` };
}

function buildGraph() {
  return new StateGraph(JobState)
    .addNode("makeOutline", makeOutline)
    .addNode("writeDraft", writeDraft)
    .addNode("polish", polish)
    .addEdge(START, "makeOutline")
    .addEdge("makeOutline", "writeDraft")
    .addEdge("writeDraft", "polish")
    .addEdge("polish", END)
    .compile();
}

function short(obj: Record<string, unknown>): string {
  // 只打印"有值"的字段，方便对照，避免空字符串刷屏
  const parts: string[] = [];
  for (const [k, v] of Object.entries(obj)) {
    if (v) parts.push(`${k}="${String(v).slice(0, 42)}${String(v).length > 42 ? "…" : ""}"`);
  }
  return `{ ${parts.join(", ")} }`;
}

async function main() {
  const graph = buildGraph();
  const input = { topic: "流式输出" };

  // ── 模式一：values —— 每步给完整状态 ──────────────────────
  banner("示例 1：streamMode = \"values\"（每次给完整状态）");
  console.log(`👤 输入：${JSON.stringify(input)}\n`);
  let i = 0;
  for await (const chunk of await graph.stream(input, { streamMode: "values" })) {
    // values 模式下 chunk 就是"当前完整状态"
    console.log(`   [${i++}] 完整状态 ${short(chunk as unknown as Record<string, unknown>)}`);
  }
  console.log("\n   ↑ 注意：第一次是初始状态（只有 topic），之后每步都是「全量快照」，字段只增不减。");

  // ── 模式二：updates —— 每步只给增量 ───────────────────────
  banner("示例 1：streamMode = \"updates\"（每次只给本步节点写了什么）");
  console.log(`👤 输入：${JSON.stringify(input)}\n`);
  let j = 0;
  for await (const chunk of await graph.stream(input, { streamMode: "updates" })) {
    // updates 模式下 chunk 的形状是 { 节点名: 该节点的返回值 }
    const record = chunk as unknown as Record<string, Record<string, unknown>>;
    for (const [node, update] of Object.entries(record)) {
      console.log(`   [${j++}] 节点 "${node}" 返回 ${JSON.stringify(update)}`);
    }
  }
  console.log("\n   ↑ 注意：每一条都带着【节点名】，且只包含那个节点这次写的字段。");
}

main().catch((e) => {
  console.error("❌ 运行失败：", e);
  process.exit(1);
});

运行：

bash

npx tsx 06-streaming/code/01-stream-values-vs-updates.ts

真实输出：

========================================================================
  示例 1：streamMode = "values"（每次给完整状态）
========================================================================

👤 输入：{"topic":"流式输出"}

   [0] 完整状态 { topic="流式输出" }
   [1] 完整状态 { topic="流式输出", outline="【流式输出】的写作大纲：1) 是什么  2) 为什么  3) 怎么用" }
   [2] 完整状态 { topic="流式输出", outline="【流式输出】的写作大纲：1) 是什么  2) 为什么  3) 怎么用", draft="草稿（基于大纲）：【流式输出】的写作大纲：1) 是什么  2) 为什么  3) 怎么…" }
   [3] 完整状态 { topic="流式输出", outline="【流式输出】的写作大纲：1) 是什么  2) 为什么  3) 怎么用", draft="草稿（基于大纲）：【流式输出】的写作大纲：1) 是什么  2) 为什么  3) 怎么…", polished="润色稿：草稿（基于大纲）：【流式输出】的写作大纲：1) 是什么  2) 为什么  3…" }

   ↑ 注意：第一次是初始状态（只有 topic），之后每步都是「全量快照」，字段只增不减。

========================================================================
  示例 1：streamMode = "updates"（每次只给本步节点写了什么）
========================================================================

👤 输入：{"topic":"流式输出"}

   [0] 节点 "makeOutline" 返回 {"outline":"【流式输出】的写作大纲：1) 是什么  2) 为什么  3) 怎么用"}
   [1] 节点 "writeDraft" 返回 {"draft":"草稿（基于大纲）：【流式输出】的写作大纲：1) 是什么  2) 为什么  3) 怎么用 —— 已展开成三段。"}
   [2] 节点 "polish" 返回 {"polished":"润色稿：草稿（基于大纲）：【流式输出】的写作大纲：1) 是什么  2) 为什么  3) 怎么用 —— 已展开成三段。 让语气更口语化。"}

   ↑ 注意：每一条都带着【节点名】，且只包含那个节点这次写的字段。

对照着看，两种模式的区别一目了然：

	`values`	`updates`
产出条数	4 条（初始 1 + 每步 1）	3 条（只算节点执行）
内容	字段累加变长，是完整快照	每次只有本步新增的字段
带节点名吗	❌ 不带	✅ 带（`{ 节点名: {...} }`）
典型用途	同步整个状态、简单镜像	展示"哪一步完成了"、按节点分流

💡 怎么选？
想"随时拿到最新完整状态"→ values。它像定时给你拍一张全景照。
想"知道每一步是谁干的、改了什么"→ updates。它像日志，每条都标了来源节点。
注意 values 会随状态变大而越来越重（每次都传全量），状态里塞了大文件/长消息时，updates 更省。

四、messages：token 级打字机

values / updates 是"图级"的流——一次给你一整个节点产出的结果。但要做出聊天框里"一个字一个字蹦"的效果，得下沉到 token 级，这就是 streamMode: "messages"。

它的工作原理是：只要节点里调用了大模型，LangGraph 就把那个模型调用产生的 token 增量抠出来给你。

示例 2：打字机 + 到达时间证明

typescript

/**
 * 示例 2：streamMode "messages" —— token 级增量 + 打字机效果
 *
 * 前面 values / updates 是"图级"的流：一次给你一整个节点产出的结果。
 * 但做聊天界面时，我们想要的是"逐字蹦出来"。这时用 streamMode:"messages"。
 *
 * messages 模式会从节点内部的大模型调用里，把 token 一级的增量抠出来。
 * 每个 chunk 的形状是： [ 消息增量(AIMessageChunk), 元数据(metadata) ]
 *
 * ⚠️ 本项目用的是推理型模型（deepseek-flash）：
 *    它会先输出一大段"思考过程"，放在 chunk.additional_kwargs.reasoning_content；
 *    真正的正文才在 chunk.content。
 *    所以下面把两者分开处理：思考只打一行提示，正文才做打字机。
 */
import { StateGraph, START, END, MessagesAnnotation } from "@langchain/langgraph";
import { HumanMessage, AIMessageChunk } from "@langchain/core/messages";
import { createModel, banner } from "../../lib/model.js";
import { assertChatKey } from "../../lib/env.js";

async function main() {
  banner("示例 2：streamMode = \"messages\"（token 级打字机）");
  assertChatKey();

  const model = createModel();

  // ── 一个最小图：只有一个节点，把用户问题交给模型 ────────────
  function chat(state: typeof MessagesAnnotation.State) {
    return model
      .invoke(state.messages)
      .then((res) => ({ messages: [res] }));
  }

  const graph = new StateGraph(MessagesAnnotation)
    .addNode("chat", chat)
    .addEdge(START, "chat")
    .addEdge("chat", END)
    .compile();

  const question = "用一句话解释什么是「流式输出」。";
  console.log(`👤 提问：${question}\n`);
  console.log("🤖 回答（打字机效果）：");

  let tokenCount = 0; // content 的 chunk 数量
  let charCount = 0; // content 的字符总数
  let reasoningSeen = false;
  const arrival: number[] = []; // 每个 content chunk 到达的时间戳

  const t0 = Date.now();
  for await (const chunk of await graph.stream(
    { messages: [new HumanMessage(question)] },
    { streamMode: "messages" }
  )) {
    // messages 模式的 chunk 是 [messageChunk, metadata] 二元组
    const [msg, meta] = chunk as unknown as [AIMessageChunk, Record<string, unknown>];

    // ① 先处理推理模型的"思考"部分：不逐字打，只提示一次
    const reasoning = (msg.additional_kwargs as Record<string, unknown>)?.reasoning_content;
    if (typeof reasoning === "string" && reasoning.length > 0) {
      if (!reasoningSeen) {
        process.stdout.write("   (🧠 模型正在思考…)\n");
        reasoningSeen = true;
      }
      continue;
    }

    // ② 正文：逐 token 写出，形成打字机效果
    const text = typeof msg.content === "string" ? msg.content : "";
    if (text.length > 0) {
      const node = String(meta?.langgraph_node ?? "?");
      process.stdout.write(`[${node}]${text}`); // 不换行 → 连着打
      tokenCount++;
      charCount += text.length;
      arrival.push(Date.now());
    }
  }
  const elapsed = Date.now() - t0;

  // ── 打印统计，证明是"逐 token 到达"，而不是一次性拿到 ────────
  console.log("\n");
  console.log("📊 统计：");
  console.log(`   • content 级 chunk 数量：${tokenCount}`);
  console.log(`   • content 字符总数      ：${charCount}`);
  console.log(`   • 首个 chunk 在 t≈${arrival.length ? arrival[0] - t0 : "-"}ms 到达`);
  if (arrival.length > 1) {
    const gaps = arrival.slice(1).map((t, i) => t - arrival[i]);
    const maxGap = Math.max(...gaps);
    const minGap = Math.min(...gaps);
    console.log(`   • 相邻 chunk 间隔：最小 ${minGap}ms / 最大 ${maxGap}ms`);
    console.log(`   • 最后一个 chunk 在 t≈${arrival[arrival.length - 1] - t0}ms 到达`);
    console.log(
      `   • 整轮总耗时：${elapsed}ms —— 远大于单次往返，说明内容是"边生成边到货"。`
    );
  }
}

main().catch((e) => {
  console.error("\n❌ 调用失败：", e);
  process.exit(1);
});

运行：

bash

npx tsx 06-streaming/code/02-stream-tokens.ts

真实输出（[chat] 那一长串就是打字机，每个 [chat] 后面跟的是一批 token）：

========================================================================
  示例 2：streamMode = "messages"（token 级打字机）
========================================================================

👤 提问：用一句话解释什么是「流式输出」。

🤖 回答（打字机效果）：
   (🧠 模型正在思考…)
[chat]流[chat]式[chat]输出[chat]是指[chat]系统[chat]在[chat]生成[chat]内容[chat]时[chat]不等[chat]全部[chat]完成[chat]，[chat]而是[chat]边[chat]生成[chat]边[chat]逐[chat]段[chat]（[chat]或[chat]逐[chat]字[chat]）[chat]实时[chat]返回[chat]给[chat]用户[chat]的技术[chat]。

📊 统计：
   • content 级 chunk 数量：30
   • content 字符总数      ：45
   • 首个 chunk 在 t≈1077ms 到达
   • 相邻 chunk 间隔：最小 0ms / 最大 10ms
   • 最后一个 chunk 在 t≈1099ms 到达
   • 整轮总耗时：1106ms —— 远大于单次往返，说明内容是"边生成边到货"。

怎么证明它是"逐 token 到达"，而不是"攒好后一次性吐出来"？

看统计里的三条：chunk 数量 30、首个 chunk 在 1077ms、最后一个 chunk 在 1099ms。如果是"一次性拿到"，那只会是 1 个 chunk；只有真的逐 token 流式，才会分成 30 次、跨时间来——我们正是在代码里给每个 chunk 打时间戳来验证这一点。

⚠️ 推理模型的坑（必须记住）：本项目默认用 deepseek-flash，它是推理型模型。它的"思考过程"不在 content 里，而在 chunk.additional_kwargs.reasoning_content；content 才是正文。
所以做打字机时必须分流：遇到 reasoning_content 就跳过（或单独打一行提示），只对 content 做逐字输出。否则你会把一大段"内心戏"当成回答打给用户。
另一个老坑：推理型模型的 maxTokens 包含思考消耗。设太小（比如 100），思考就把额度花光，content 变成空字符串——而且不报错。排查"回答是空的"时，先怀疑这一条，本章所有示例都不设 maxTokens。

五、custom：把业务进度也推出去

values/updates 是框架产的，messages 是模型产的。可还有很多进度信息只有你的业务逻辑知道——"正在检索第 3/10 篇"、"已完成子任务 A"。这类事件用 streamMode: "custom"。

做法：节点里调用 getWriter() 拿到一个"上报函数"，然后把任意数据喂给它，外层就能以 custom 模式收到。

示例 3：进度条式上报

typescript

/**
 * 示例 3：streamMode "custom" —— 节点内用 getWriter() 上报自定义进度
 *
 * values / updates 是"框架帮你产"的事件；messages 是"模型帮你产"的事件。
 * 而 custom 模式，是"你自己产"的事件：
 *   节点里调用 getWriter() 拿到上报函数，再喂任意数据，就会以
 *   streamMode:"custom" 的形式流出。
 *
 * 典型用途：进度条、阶段提示、"正在检索 3/10"、"已完成子任务 A"……
 * 这些信息框架猜不到，只有你的业务逻辑知道，所以要主动上报。
 *
 * 本示例用一个带 sleep 的模拟检索工作流，做出进度条式输出。
 */
import { StateGraph, START, END, Annotation, getWriter } from "@langchain/langgraph";
import { banner } from "../../lib/model.js";

const WorkState = Annotation.Root({
  task: Annotation<string>, // 输入：任务名
  result: Annotation<string>, // 输出：最终结果
});

const sleep = (ms: number) => new Promise((r) => setTimeout(r, ms));

// 节点：跑一个 10 步的"检索"流程，每步用 getWriter() 上报进度
async function research(state: typeof WorkState.State) {
  // ★ 关键：进入节点后先拿到"上报隧道"。
  //   本版本里 getWriter()（走 AsyncLocalStorage）是可靠的写法。
  const write = getWriter();
  const total = 10;
  for (let i = 1; i <= total; i++) {
    await sleep(120); // 模拟每一步的真实耗时（网络请求 / 计算）
    // 把自定义事件推给外层 stream
    write?.({ type: "progress", step: i, total, label: `检索分片 ${i}/${total}` });

    // 在第 5 步时，额外上报一个"子任务完成"事件
    if (i === 5) {
      write?.({ type: "milestone", label: "✅ 已完成子任务 A：抓取候选文档" });
    }
  }
  write?.({ type: "milestone", label: "✅ 已完成子任务 B：重排与去重" });
  return { result: `对「${state.task}」的检索完成，共处理 ${total} 个分片。` };
}

function bar(step: number, total: number, width = 20): string {
  const filled = Math.round((step / total) * width);
  return "█".repeat(filled) + "░".repeat(width - filled);
}

async function main() {
  banner("示例 3：streamMode = \"custom\"（节点主动上报进度）");

  const graph = new StateGraph(WorkState)
    .addNode("research", research)
    .addEdge(START, "research")
    .addEdge("research", END)
    .compile();

  console.log("👤 任务：检索 10 个分片并汇总\n");
  console.log("📡 实时进度事件流（每个 writer 事件单独一行，真实终端里可原地刷新成进度条）：");

  for await (const event of await graph.stream(
    { task: "LangGraph 流式输出" },
    { streamMode: "custom" }
  )) {
    const e = event as unknown as Record<string, unknown>;

    if (e.type === "progress") {
      const step = e.step as number;
      const total = e.total as number;
      console.log(
        `   ${bar(step, total)} ${String(step).padStart(2)}/${total}  ${e.label}`
      );
    } else if (e.type === "milestone") {
      console.log(`   ${e.label}`);
    }
  }

  // custom 事件里没有最终状态，所以流结束后再 invoke 一次拿结果
  // （writer() 在没有活跃 stream 时是空操作，不会报错）
  const final = await graph.invoke({ task: "LangGraph 流式输出" });
  console.log(`\n📦 最终结果：${final.result}`);
}

main().catch((e) => {
  console.error("❌ 运行失败：", e);
  process.exit(1);
});

运行：

bash

npx tsx 06-streaming/code/03-stream-custom-progress.ts

真实输出：

========================================================================
  示例 3：streamMode = "custom"（节点主动上报进度）
========================================================================

👤 任务：检索 10 个分片并汇总

📡 实时进度事件流（每个 writer 事件单独一行，真实终端里可原地刷新成进度条）：
   ██░░░░░░░░░░░░░░░░░░  1/10  检索分片 1/10
   ████░░░░░░░░░░░░░░░░  2/10  检索分片 2/10
   ██████░░░░░░░░░░░░░░  3/10  检索分片 3/10
   ████████░░░░░░░░░░░░  4/10  检索分片 4/10
   ██████████░░░░░░░░░░  5/10  检索分片 5/10
   ✅ 已完成子任务 A：抓取候选文档
   ████████████░░░░░░░░  6/10  检索分片 6/10
   ██████████████░░░░░░  7/10  检索分片 7/10
   ████████████████░░░░  8/10  检索分片 8/10
   ██████████████████░░  9/10  检索分片 9/10
   ████████████████████ 10/10  检索分片 10/10
   ✅ 已完成子任务 B：重排与去重

📦 最终结果：对「LangGraph 流式输出」的检索完成，共处理 10 个分片。

💡 进度条那一列 ██░░░░ 是我用 bar() 现画的；配合终端里的 \r（回车不换行）就能做成"原地刷新的进度条"。本示例为了输出可复现（方便你复制贴到别处），改成每个事件占一行。
⚠️ 一个真实的坑：writer() 在本版本里是"哑"的。官方文档和不少示例写的是直接 import { writer } 然后 writer({...})。但在 @langchain/langgraph 1.4.21 里，裸 writer() 会静默地什么都不发（它从 AsyncLocalStorage 里读的是旧路径 configurable.writer，而当前版本把上报函数挂在 config.writer 上）。可靠写法是 getWriter()——它会同时兼容两种路径，直接返回可用的上报函数。如果你照文档写 writer() 却发现 custom 事件收不到，改用 getWriter() 即可。

⚠️ custom 事件里不含最终状态。所以示例里流结束后又 invoke() 了一次拿 result。真实应用一般会用多模式订阅（见下一节）在同一个流里同时拿进度和最终状态，避免跑两遍。

六、多模式合并订阅

真实应用往往"既要又要"：既要 token 打字机，又要节点完成通知，还要业务进度条。LangGraph 允许把 streamMode 写成数组：

typescript

graph.stream(input, { streamMode: ["updates", "messages", "custom"] })

⚠️ 数组模式下，每个 chunk 的形状会变——从"裸数据"变成二元组 [模式名, 数据]：

["updates",  { 节点名: 该节点返回值 }]
["messages", [ 消息增量, 元数据 ]]
["custom",   你 getWriter 出来的任意数据 ]

只要看第 0 个元素是哪个模式名，就能分流处理。

示例 4：三模式合并

typescript

/**
 * 示例 4：同时订阅多个模式
 *
 * 真实应用里，我们往往"既要又要"：
 *   既要 token 打字机（messages），
 *   又要知道哪个节点跑完了（updates），
 *   还要业务进度条（custom）。
 *
 * LangGraph 允许把 streamMode 写成数组：
 *   graph.stream(input, { streamMode: ["updates", "messages", "custom"] })
 *
 * ⚠️ 数组模式下，每个 chunk 的形状变成【二元组】：[ 模式名, 数据 ]
 *      ["updates",  { 节点名: 该节点返回值 } ]
 *      ["messages", [ 消息增量, 元数据 ] ]
 *      ["custom",   你 writer 出来的任意数据 ]
 *    只要看第 0 个元素是哪个模式名，就能分流处理。
 */
import { StateGraph, START, END, MessagesAnnotation, getWriter } from "@langchain/langgraph";
import { HumanMessage, AIMessage, AIMessageChunk } from "@langchain/core/messages";
import { createModel, banner } from "../../lib/model.js";
import { assertChatKey } from "../../lib/env.js";

const sleep = (ms: number) => new Promise((r) => setTimeout(r, ms));

async function main() {
  banner("示例 4：同时订阅 updates + messages + custom");
  assertChatKey();

  const model = createModel();

  // 节点 1：模拟"规划"，用 custom 上报进度
  async function plan(state: typeof MessagesAnnotation.State) {
    const write = getWriter(); // 拿到上报隧道
    for (const step of ["分析意图", "选择策略", "生成计划"]) {
      await sleep(150);
      write?.({ type: "step", label: `规划中：${step}` });
    }
    return { messages: [new AIMessage("（内部）计划：直接作答。")] };
  }

  // 节点 2：真正问模型 → 触发 messages 模式的 token 流
  async function answer(state: typeof MessagesAnnotation.State) {
    const res = await model.invoke(state.messages);
    return { messages: [res] };
  }

  const graph = new StateGraph(MessagesAnnotation)
    .addNode("plan", plan)
    .addNode("answer", answer)
    .addEdge(START, "plan")
    .addEdge("plan", "answer")
    .addEdge("answer", END)
    .compile();

  const question = "用一句话介绍 SSE。";
  console.log(`👤 提问：${question}\n`);

  console.log("🔀 多模式合并事件流（前缀标注来源模式）：");
  let reasoningHinted = false;

  for await (const chunk of await graph.stream(
    { messages: [new HumanMessage(question)] },
    { streamMode: ["updates", "messages", "custom"] }
  )) {
    // 数组模式：chunk 恒为 [模式名, 数据]
    const [mode, data] = chunk as unknown as [string, unknown];

    if (mode === "custom") {
      const e = data as Record<string, unknown>;
      console.log(`   [custom ] 🧩 ${e.label}`);
    } else if (mode === "updates") {
      const record = data as Record<string, Record<string, unknown>>;
      for (const node of Object.keys(record)) {
        console.log(`   [updates] ✅ 节点 "${node}" 执行完毕`);
      }
    } else if (mode === "messages") {
      const [msg, meta] = data as [AIMessageChunk, Record<string, unknown>];
      const node = String(meta?.langgraph_node ?? "?");
      // 推理模型的思考：只提示一次，不当正文打
      const reasoning = (msg.additional_kwargs as Record<string, unknown>)?.reasoning_content;
      if (typeof reasoning === "string" && reasoning.length > 0) {
        if (!reasoningHinted) {
          console.log(`   [messages] 🧠 (${node} 正在思考…)`);
          reasoningHinted = true;
        }
        continue;
      }
      const text = typeof msg.content === "string" ? msg.content : "";
      if (text.length > 0) {
        // 用 process.stdout.write 打字机；前缀只打第一次，避免刷屏
        process.stdout.write(`   [messages] ⬇ ${text}`);
      }
    }
  }
  console.log("");
}

main().catch((e) => {
  console.error("\n❌ 调用失败：", e);
  process.exit(1);
});

运行：

bash

npx tsx 06-streaming/code/04-stream-multi-modes.ts

真实输出：

========================================================================
  示例 4：同时订阅 updates + messages + custom
========================================================================

👤 提问：用一句话介绍 SSE。

🔀 多模式合并事件流（前缀标注来源模式）：
   [custom ] 🧩 规划中：分析意图
   [custom ] 🧩 规划中：选择策略
   [custom ] 🧩 规划中：生成计划
   [messages] ⬇ （内部）计划：直接作答。   [updates] ✅ 节点 "plan" 执行完毕
   [messages] 🧠 (answer 正在思考…)
   [messages] ⬇ SS   [messages] ⬇ E   [messages] ⬇ （   [messages] ⬇ Server   [messages] ⬇ -S   [messages] ⬇ ent   [messages] ⬇  Events   [messages] ⬇ ）   [messages] ⬇ 是一种   [messages] ⬇ 基于   [messages] ⬇  HTTP   [messages] ⬇     [messages] ⬇ 长   [messages] ⬇ 连接   [messages] ⬇ 、   [messages] ⬇ 允许   [messages] ⬇ 服务器   [messages] ⬇ 单向   [messages] ⬇ 、   [messages] ⬇ 实时   [messages] ⬇ 地向   [messages] ⬇ 客户端   [messages] ⬇ 推送   [messages] ⬇ 文本   [messages] ⬇ 事件   [messages] ⬇ 流   [messages] ⬇ 的技术   [messages] ⬇ 。   [updates] ✅ 节点 "answer" 执行完毕

请把注意力放到前缀上，这就是"分流"的真相：

前缀	来自哪个模式	这一行在说什么
`[custom ]`	`custom`	`plan` 节点上报的业务进度
`[messages]`	`messages`	token 增量（打字机）
`[updates]`	`updates`	某个节点执行完毕

有两个细节值得单独拎出来：

💡 细节 1：messages 模式也会吐出"你自己造的消息"。看第 4 行——[messages] ⬇ （内部）计划：直接作答。 并不是模型产的，而是 plan 节点里 new AIMessage(...) 造的。只要一条消息进入图的消息流，messages 模式就会把它当事件推出来（整条推，而非逐 token）。真实项目里如果你想"内部消息不进展示流"，就别把它放进 messages 字段，或在前端按 metadata.langgraph_node 过滤。

💡 细节 2：不同模式的事件顺序是"交错"的。第 4 行里 [messages] 和 [updates] 挤在同一行，是因为 updates 是"节点执行完毕"才发，恰好紧跟在它产生的消息后面。不要假设事件顺序是严格交替的，要用前缀（模式名）来判断类型，而不是靠位置。

七、SSE：前端的最后一公里

到这一步，后端已经能拿到一串事件了。可前端在另一个进程、另一台机器上，怎么实时收到？答案是 SSE（Server-Sent Events）——一条 HTTP 长连接，服务器持续往里写文本，浏览器持续读。

SSE 的每一帧格式固定：

event: token
data: {"text":"你"}
<空行>

event: 是事件类型（前端 addEventListener("token", ...) 监听）；data: 是 UTF-8 文本；一个空行代表一帧结束。

示例 5：把图的事件翻译成 SSE 帧

typescript

/**
 * 示例 5：把图的事件翻译成 SSE（Server-Sent Events）文本帧
 *
 * 前端的"实时打字"最终是靠一条 HTTP 长连接把事件推过去的，
 * 这条连接上跑的最常见协议就是 SSE。SSE 的每一帧长得像这样：
 *
 *     event: token
 *     data: {"text":"你"}
 *     空行
 *
 * 本示例不装任何新依赖，就在终端把"图的事件 → SSE 帧"这一步做出来，
 * 让你看清前端到底收到了什么。真实的服务器只是把这些字符串写进
 * HTTP response 的 body 而已。
 *
 * 为什么流式对 LLM 应用是刚需？关键指标是 TTFT（Time To First Token，
 * 首字延迟）。非流式要等整段回答生成完才返回（数秒空白），
 * 流式则让第一个字在几百毫秒内就出现 —— 体验天差地别。本示例会实测 TTFT。
 */
import { StateGraph, START, END, MessagesAnnotation, getWriter } from "@langchain/langgraph";
import { HumanMessage, AIMessageChunk } from "@langchain/core/messages";
import { createModel, banner } from "../../lib/model.js";
import { assertChatKey } from "../../lib/env.js";

const sleep = (ms: number) => new Promise((r) => setTimeout(r, ms));

// 把一个事件编码成一帧 SSE 文本
function sseFrame(event: string, data: unknown, id?: number): string {
  const lines: string[] = [];
  if (id !== undefined) lines.push(`id: ${id}`);
  lines.push(`event: ${event}`);
  lines.push(`data: ${JSON.stringify(data)}`);
  return lines.join("\n") + "\n\n"; // 帧与帧之间用空行分隔
}

async function main() {
  banner("示例 5：翻译成 SSE 协议帧");
  assertChatKey();

  const model = createModel();

  // 节点：先上报 3 步进度（custom），再让模型流式产出正文（messages）
  async function job(state: typeof MessagesAnnotation.State) {
    const write = getWriter();
    for (let i = 1; i <= 3; i++) {
      await sleep(150);
      write?.({ phase: `预处理 ${i}/3` });
    }
    const res = await model.invoke(state.messages);
    return { messages: [res] };
  }

  const graph = new StateGraph(MessagesAnnotation)
    .addNode("job", job)
    .addEdge(START, "job")
    .addEdge("job", END)
    .compile();

  const question = "用一句话说明流式输出的好处。";
  console.log(`👤 提问：${question}\n`);
  console.log("📨 下面是后端真正写进 HTTP 响应体的 SSE 文本帧：\n");
  console.log("--------------------------------------------------------");

  const t0 = Date.now();
  let firstFrame: number | null = null; // 首帧延迟（任意事件）
  let ttft: number | null = null; // 首字延迟（第一个正文 token）
  let seq = 0;
  let reasoningHinted = false;

  for await (const chunk of await graph.stream(
    { messages: [new HumanMessage(question)] },
    { streamMode: ["custom", "messages"] }
  )) {
    if (firstFrame === null) firstFrame = Date.now() - t0;
    const [mode, data] = chunk as unknown as [string, unknown];

    if (mode === "custom") {
      const e = data as Record<string, unknown>;
      process.stdout.write(sseFrame("progress", e, seq++));
    } else if (mode === "messages") {
      const [msg] = data as [AIMessageChunk, Record<string, unknown>];
      const reasoning = (msg.additional_kwargs as Record<string, unknown>)?.reasoning_content;
      if (typeof reasoning === "string" && reasoning.length > 0) {
        if (!reasoningHinted) {
          process.stdout.write(sseFrame("reasoning", { stage: "thinking" }, seq++));
          reasoningHinted = true;
        }
        continue;
      }
      const text = typeof msg.content === "string" ? msg.content : "";
      if (text.length > 0) {
        if (ttft === null) ttft = Date.now() - t0; // 记下第一个正文 token 的时刻
        process.stdout.write(sseFrame("token", { text }, seq++));
      }
    }
  }

  process.stdout.write(sseFrame("done", { ok: true }, seq++));
  const total = Date.now() - t0;
  console.log("--------------------------------------------------------");
  console.log("\n📊 指标：");
  console.log(`   • 首帧延迟（第一个事件到达）：${firstFrame ?? "-"} ms`);
  console.log(`   • TTFT（第一个正文 token）：${ttft ?? "-"} ms`);
  console.log(`   • 整轮总耗时              ：${total} ms`);
  console.log(
    `   • 结论：用户在 ${firstFrame ?? "-"} ms 就看到了实时反馈，` +
      `而整段回答要到 ${total} ms 才结束。若用非流式 invoke()，` +
      `用户要干等 ${total} ms 才看到任何东西 —— 这就是流式的体感差异。`
  );
}

main().catch((e) => {
  console.error("\n❌ 调用失败：", e);
  process.exit(1);
});

运行：

bash

npx tsx 06-streaming/code/05-stream-sse-protocol.ts

真实输出（中间十来个 token 帧格式完全一致，为节省篇幅仅在此处做输出节选，其余均为逐字复制的终端结果）：

========================================================================
  示例 5：翻译成 SSE 协议帧
========================================================================

👤 提问：用一句话说明流式输出的好处。

📨 下面是后端真正写进 HTTP 响应体的 SSE 文本帧：

--------------------------------------------------------
id: 0
event: progress
data: {"phase":"预处理 1/3"}

id: 1
event: progress
data: {"phase":"预处理 2/3"}

id: 2
event: progress
data: {"phase":"预处理 3/3"}

id: 3
event: reasoning
data: {"stage":"thinking"}

…（此处为输出节选：中间 4~30 号均为格式与上面完全一致的 `token` 帧，逐字省略）…

id: 31
event: done
data: {"ok":true}

--------------------------------------------------------

📊 指标：
   • 首帧延迟（第一个事件到达）：162 ms
   • TTFT（第一个正文 token）：1788 ms
   • 整轮总耗时              ：1899 ms
   • 结论：用户在 162 ms 就看到了实时反馈，而整段回答要到 1899 ms 才结束。若用非流式 invoke()，用户要干等 1899 ms 才看到任何东西 —— 这就是流式的体感差异。

💡 首帧 162ms vs 总耗时 1899ms：前者是用户"看到系统在动"的时刻（先 progress 后 reasoning），后者才是回答真正完成。用 invoke() 会让用户白白多等这一整段。

真实前端怎么消费这些帧？

后端把上面的字符串写进 HTTP 响应的 body（Content-Type: text/event-stream），前端有三种主流接法：

① 直接用浏览器原生 EventSource（零依赖，最能看清协议）：

typescript

const es = new EventSource("/api/chat");       // 你的后端 SSE 端点
es.addEventListener("token", (e) => {          // 对应 event: token
  const { text } = JSON.parse(e.data);
  out.textContent += text;                     // 往界面上追加一个字
});
es.addEventListener("done", () => es.close()); // 收到 done 就断开

② 用 LangGraph 官方的 @langchain/langgraph-sdk 的 useStream()（React Hook，把上面这些活儿全包了）：它内部就是走 SSE，直接给你 messages、values、isLoading 等响应式状态，配合官方 Agent Chat UI 可以几分钟搭出一个聊天前端。这是生产环境最省事的路子。

③ 自己写后端，接任何前端框架（Vue / Svelte / 小程序都行）：本质上就是"把示例 5 里 sseFrame() 的输出，res.write() 到响应流里"。核心逻辑你已经在本节亲手写过了。

💡 一句话总结这一章：stream() 是后端把"结果"拆成"过程"；SSE 是把"过程"跨进程送到前端；useStream / EventSource 是把"过程"画到屏幕上。三段拼起来，就是所有"实时 AI 界面"的共同骨架。

八、本章小结

本章速查表

你要做的事	代码
拿到完整状态快照	`graph.stream(input, { streamMode: "values" })`
拿到"哪个节点写了什么"	`graph.stream(input, { streamMode: "updates" })` → `{ 节点名: {...} }`
token 级打字机	`graph.stream(input, { streamMode: "messages" })` → `[消息增量, 元数据]`
节点内上报自定义进度	`const write = getWriter(); write?.({...})`
订阅自定义进度	`graph.stream(input, { streamMode: "custom" })`
同时订阅多个模式	`graph.stream(input, { streamMode: ["updates","messages","custom"] })`
区分多模式事件	chunk 是 `[模式名, 数据]`，看第 0 个元素
取本节点名	`metadata.langgraph_node`
拿推理模型的思考	`chunk.additional_kwargs.reasoning_content`
翻译成 SSE	`event: X\ndata: {...}\n\n`
前端接 SSE	`EventSource` / `useStream()`

三句话总结

流式不改变总耗时，只改变"用户什么时候开始看到东西" —— 用 TTFT / 首帧延迟来量化这份体感。
四种模式各有分工：values 全量、updates 增量、messages token、custom 业务进度；写成数组就能同时订阅，靠 [模式名, 数据] 二元组分流。
custom 事件要靠 getWriter() 主动上报（本版本裸 writer() 是哑的），它让你能把"框架不知道、只有业务知道"的进度实时推给用户。

动手练习

给示例 2 的图加一个"翻译节点"（把模型的英文回答翻译成中文），用 streamMode: "updates" 观察两个节点分别产出了什么，再用 ["updates","messages"] 同时订阅，确认 metadata.langgraph_node 能区分出 token 来自哪个节点。
改造示例 4：让 plan 节点上报的事件里带一个百分比（0~100），并在打印时把百分比拼进 [custom ] 那一行。试着只用 ["custom","values"] 两个模式，在同一个流里既拿到进度、又拿到最终状态（提示：values 每次给完整状态，取最后一条即可）。
给示例 5 加一个 error 事件：让节点在"预处理"到一半时 throw，观察 SSE 流里会多出什么（提示：真实的 SSE 服务器要在 try/catch 里补发一帧 event: error，否则前端只会看到连接断开）。

下一章预告

第 7 章我们进入多智能体（Multi-Agent）：当一个任务复杂到"一个智能体忙不过来"，怎么让多个专家分工协作？主管（Supervisor）怎么派活、子智能体之间怎么交接、什么时候该用 @langchain/langgraph-supervisor、什么时候该用 swarm 式的自由转交？我们会把第 6 章的流式输出接到多智能体上，让你实时看到"哪个专家正在干活"。

← 上一章：人工介入（HITL）｜返回总目录｜下一章：多智能体 →

第 6 章 · 流式输出与实时反馈 ​

🎯 本章目标 ​

一、为什么流式是刚需：先看一个数字 ​

二、streamMode 全景图 ​

三、values vs updates：全量 vs 增量 ​

示例 1：同一张图，两种模式 ​

四、messages：token 级打字机 ​

示例 2：打字机 + 到达时间证明 ​

五、custom：把业务进度也推出去 ​

示例 3：进度条式上报 ​

六、多模式合并订阅 ​

示例 4：三模式合并 ​

七、SSE：前端的最后一公里 ​

示例 5：把图的事件翻译成 SSE 帧 ​

真实前端怎么消费这些帧？ ​

八、本章小结 ​

第 6 章 · 流式输出与实时反馈

🎯 本章目标

一、为什么流式是刚需：先看一个数字

二、streamMode 全景图

三、values vs updates：全量 vs 增量

示例 1：同一张图，两种模式

四、messages：token 级打字机

示例 2：打字机 + 到达时间证明

五、custom：把业务进度也推出去

示例 3：进度条式上报

六、多模式合并订阅

示例 4：三模式合并

七、SSE：前端的最后一公里

示例 5：把图的事件翻译成 SSE 帧

真实前端怎么消费这些帧？

八、本章小结