第 14 章 · 前端集成
🎯 本章目标
- 想清楚前端到底要什么：实时性、稳定的契约、人机协作、会话与历史——四件事
- 用
node:http起一个真正的 SSE 流式后端，把图的 token 流写进 HTTP 响应，并用fetch + ReadableStream亲手把它解析回来 - 把每种
streamMode事件的真实形状抓下来，生成一张「前后端事件协议清单表」和可直接复制的 TypeScript 类型定义 - 用两个 HTTP 接口（
POST /chat暂停 +POST /resume恢复）把人工介入（第 5 章）接到前端 - 拿到可直接复制的 React 片段（官方
useStream版 + 裸EventSource版，明确标注：本机未安装 React，未实跑） - 用
getStateHistory给前端提供会话与历史回放的数据（时间轴 / 时间旅行 UI） - 认识官方现成的 UI 生态（Agent Chat UI、gen-ui 示例、各框架 SDK），知道该选哪条路
本章代码
📌 代码约定：正文的代码块都保持可直接运行（含 import 与模型初始化）。若某段为聚焦概念的节选，会明确标注「节选」并指向同名的
code/完整文件。所有示例的输出都是真实运行结果，不是编造的示例输出。⚠️ 本章特别说明：本机不能安装 React、也不能在浏览器里跑前端。因此本章的做法是：用 Node 端把「前端需要消费的东西」真实产出来（这部分全部实跑通过），而 React 代码作为「可直接复制的片段」打印出来，并在输出与正文里明确标注未实跑。不伪造任何浏览器里的运行输出。
| 文件 | 内容 | 实跑？ |
|---|---|---|
code/01-sse-backend.ts | SSE 流式后端：node:http + streamMode:"messages"，自起服务、自解析 | ✅ 实跑 |
code/02-event-protocol.ts | 前后端事件协议：抓真实事件形状，生成协议清单 + TS 类型 | ✅ 实跑 |
code/03-hitl-over-http.ts | 通过 HTTP 做人工介入：POST /chat 暂停 + POST /resume 恢复 | ✅ 实跑 |
code/04-react-snippets.ts | 可复制的 React 片段（打印为字符串，非实跑） | ⚠️ 仅打印 |
code/05-session-history-api.ts | 会话与历史回放：getStateHistory → GET /history | ✅ 实跑 |
一、前端到底要什么
走到全书最后一章，后端的能力其实已经齐了：状态、控制流、持久化、人工介入、流式、工程化。唯一还没回答的问题是：坐在浏览器前的那个用户，怎么实时看到这一切？
拆开来看，前端真正需要后端提供的是四样东西：
| 前端要的 | 后端要给的 | 对应章节 |
|---|---|---|
| 实时性：字要一个字一个字蹦出来，别转圈等到天荒地老 | 流式推送（SSE） | 第 6 章 + 本章示例 1 |
| 稳定的契约：后端会推哪些事件、每个长什么样，得写死 | 事件协议（类型定义） | 本章示例 2 |
| 人机协作：关键步骤停下来等用户点「批准 / 拒绝」 | 中断的 HTTP 收发 | 第 5 章 + 本章示例 3 |
| 会话与回放：刷新页面还能看到历史，甚至回看每一步 | getStateHistory | 第 4 章 + 本章示例 5 |
这四件事里，实时性是最容易被低估、又最影响体感的一件。
为什么流式对前端是刚需：一个数字
第 6 章我们用一个指标把「体感」量化了——TTFT（Time To First Token，首字延迟）。那一次本机实测的数字是：
首帧延迟（第一个事件到达）：162 ms
整轮总耗时 ：1899 ms差了一个数量级。 如果前端用非流式的 invoke()，用户要干等 1899ms 才能看到任何东西——这 1.9 秒里界面一片空白，用户会以为「卡死了」，甚至反复点按钮。而流式让用户在 162ms 就看到了「系统在动」：先是一行进度、再是一句「正在思考」，然后文字才开始一个个长出来。
本章示例 1 会再实测一次：它从发出请求到收到第一个正文 token 只花了 1356ms（本次运行），而整段回答到 1392ms 才结束——注意这里 TTFT 已经是「第一个正文 token」，等待的大头是推理型模型先输出的一大段思考过程。
💡 一句话：流式不改变总耗时，它改变的是**「用户什么时候开始觉得系统活着」**。对聊天界面，这不是锦上添花，是生死线。
所以前端集成的第一课，就是先把 SSE 流式后端搭起来——这正是示例 1 要做的。
二、SSE 流式后端
示例 1：SSE 流式后端
什么是 SSE
SSE（Server-Sent Events） 是一条 HTTP 长连接：客户端发起一次请求，服务器保持连接不关闭，持续往里写文本；客户端持续读。它的每一帧格式固定：
event: token
data: {"text":"你"}
<空行>event:是事件类型（前端es.addEventListener("token", ...)监听它）data:是 UTF-8 文本（通常是 JSON）- 一个空行代表一帧结束
第 6 章我们把图的事件翻译成了这样的字符串，但只是打印在终端。本章示例 1 更进一步：真的起一个 HTTP 服务，把这些帧写进响应体。
关键点
- 响应头必须声明
Content-Type: text/event-stream，浏览器/客户端才会把它当流来读。 - 每产生一个事件就
res.write(frame)，中途不要res.end()；全部结束后再res.end()。 streamMode:"messages"负责把模型 token 抠出来（第 6 章），我们只负责「事件 → 帧」的包装。- 推理型模型要分流：
reasoning_content（思考）单发一个reasoning事件，别混进正文。
完整代码（code/01-sse-backend.ts）
typescript
/**
* 示例 1：SSE 流式后端 —— 把图的 token 流变成一条 HTTP 长连接
*
* 第 6 章我们把图的事件翻译成 SSE 帧，但只是"打印在终端里"。本示例真的用
* node:http 起服务，把帧写进 HTTP 响应体，让另一个进程 / 另一台机器上的前端
* 实时收到。关键：响应头声明 Content-Type: text/event-stream，然后每产生一个
* 事件，就往响应流 write 一帧（event: X \n data: {...} \n 空行）。
*
* ⚠️ 本示例被设计成"一次 npx tsx 就能跑完并自动退出"：自己 listen → 自己 fetch
* 请求、手工解析 SSE → 打印每帧 → server.close() 退出，不是常驻服务。
*/
import { createServer, type IncomingMessage, type ServerResponse } from "node:http";
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";
const PORT = 8790; // 挑一个不常冲突的端口
// 把一个事件编码成一帧 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 readJson(req: IncomingMessage): Promise<Record<string, unknown>> {
const chunks: Buffer[] = [];
for await (const chunk of req) chunks.push(chunk as Buffer);
const raw = Buffer.concat(chunks).toString("utf8");
if (!raw) return {};
return JSON.parse(raw) as Record<string, unknown>;
}
async function main() {
banner("示例 1：SSE 流式后端（node:http + graph.stream streamMode:\"messages\"）");
assertChatKey();
const model = createModel({ temperature: 0 });
// 图：一个 chat 节点。streamMode:"messages" 会把它调的 token 抠出来
async function chatNode(state: typeof MessagesAnnotation.State) {
const res = await model.invoke(state.messages);
return { messages: [res] };
}
const graph = new StateGraph(MessagesAnnotation)
.addNode("chat", chatNode)
.addEdge(START, "chat")
.addEdge("chat", END)
.compile();
// ★ 核心：把一个"聊天请求"变成一串 SSE 帧，写进响应流
async function streamChatToClient(res: ServerResponse, message: string, threadId: string) {
// 前端收到的是"流"，必须声明 text/event-stream；另两个头是防中间代理缓冲
res.writeHead(200, {
"Content-Type": "text/event-stream; charset=utf-8",
"Cache-Control": "no-cache, no-transform",
Connection: "keep-alive",
"X-Accel-Buffering": "no",
});
let seq = 0;
let reasoningHinted = false;
try {
const stream = await graph.stream(
{ messages: [new HumanMessage(message)] },
{ streamMode: "messages", configurable: { thread_id: threadId } }
);
for await (const chunk of stream) {
const [msg, meta] = chunk as unknown as [AIMessageChunk, Record<string, unknown>];
const node = String(meta?.langgraph_node ?? "?");
// 推理型模型的"思考"：单独发一个 reasoning 事件（只发一次），不当正文
const reasoning = (msg.additional_kwargs as Record<string, unknown>)?.reasoning_content;
if (typeof reasoning === "string" && reasoning.length > 0) {
if (!reasoningHinted) {
res.write(sseFrame("reasoning", { node, stage: "thinking" }, seq++));
reasoningHinted = true;
}
continue;
}
const text = typeof msg.content === "string" ? msg.content : "";
if (text.length > 0) res.write(sseFrame("token", { node, text }, seq++));
}
res.write(sseFrame("done", { ok: true, thread_id: threadId }, seq++));
} catch (e) {
// 出错也要让前端知道，否则它只会看到连接莫名断开
res.write(sseFrame("error", { message: (e as Error).message }, seq++));
}
res.end();
}
const server = createServer(async (req, res) => {
const url = new URL(req.url ?? "/", `http://localhost:${PORT}`);
const method = req.method ?? "GET";
console.log(` [server] ${method} ${url.pathname}`);
try {
if (method === "GET" && url.pathname === "/health") {
res.writeHead(200, { "Content-Type": "application/json; charset=utf-8" });
return void res.end(JSON.stringify({ status: "ok" }));
}
// POST /chat：从 JSON body 拿参数（推荐，前端 fetch 用）
if (method === "POST" && url.pathname === "/chat") {
const body = await readJson(req);
if (typeof body.message !== "string" || typeof body.thread_id !== "string") {
res.writeHead(400, { "Content-Type": "application/json; charset=utf-8" });
return void res.end(JSON.stringify({ error: "需要 { message: string, thread_id: string }" }));
}
return streamChatToClient(res, body.message, body.thread_id);
}
// GET /chat：给浏览器原生 EventSource 用（它只能发 GET）
if (method === "GET" && url.pathname === "/chat") {
const message = url.searchParams.get("message");
const threadId = url.searchParams.get("thread_id");
if (!message || !threadId) {
res.writeHead(400, { "Content-Type": "application/json; charset=utf-8" });
return void res.end(JSON.stringify({ error: "需要 query 参数 message 和 thread_id" }));
}
return streamChatToClient(res, message, threadId);
}
res.writeHead(404, { "Content-Type": "application/json; charset=utf-8" });
res.end(JSON.stringify({ error: `未知路径：${method} ${url.pathname}` }));
} catch (e) {
console.error(" [server] 处理出错：", (e as Error).message);
if (!res.headersSent) res.writeHead(500, { "Content-Type": "application/json; charset=utf-8" });
res.end(JSON.stringify({ error: (e as Error).message }));
}
});
await new Promise<void>((resolve) => server.listen(PORT, resolve));
const addr = server.address();
const actualPort = typeof addr === "object" && addr ? addr.port : PORT;
const base = `http://127.0.0.1:${actualPort}`;
console.log(`✅ 服务已启动：${base}（跑完会自己关掉）\n`);
// ═══════ 自测：像前端那样去消费这条 SSE 流 ═══════
console.log("🔎 环境探测：");
console.log(` typeof EventSource = ${typeof (globalThis as Record<string, unknown>).EventSource}`);
const hasEventSource = typeof (globalThis as Record<string, unknown>).EventSource === "function";
console.log(
hasEventSource
? " → 本机有内置 EventSource"
: " → 本机没有内置 EventSource，改用 fetch + ReadableStream 手工解析（下面的自测）"
);
console.log(
" ℹ️ 浏览器里的 EventSource 只能发 GET。所以本示例额外保留了 GET /chat：\n" +
" EventSource 打 GET；fetch 打 POST（带 JSON body）。\n"
);
// 手工解析版：不依赖任何库，最能看清协议
async function readSseStream(resp: Response): Promise<void> {
const reader = resp.body!.getReader();
const decoder = new TextDecoder();
let buffer = "";
const t0 = Date.now();
let ttft: number | null = null;
let frameCount = 0;
console.log("📨 前端收到并解析出的帧（每帧一段）：");
console.log(" " + "-".repeat(56));
while (true) {
const { value, done } = await reader.read();
if (done) break;
buffer += decoder.decode(value, { stream: true });
let sep: number;
// SSE 以空行分隔帧；按 "\n\n" 切
while ((sep = buffer.indexOf("\n\n")) !== -1) {
const rawFrame = buffer.slice(0, sep);
buffer = buffer.slice(sep + 2);
let event = "message";
let data = "";
for (const line of rawFrame.split("\n")) {
if (line.startsWith("event:")) event = line.slice(6).trim();
else if (line.startsWith("data:")) data = line.slice(5).trim();
}
frameCount++;
const payload = JSON.parse(data) as Record<string, unknown>;
if (event === "token") {
if (ttft === null) ttft = Date.now() - t0;
console.log(` [event: token ] ${JSON.stringify(payload)}`);
} else {
console.log(` [event: ${event.padEnd(5)}] ${JSON.stringify(payload)}`);
}
if (event === "done" || event === "error") {
console.log(" " + "-".repeat(56));
console.log(` 📊 共收到 ${frameCount} 帧；TTFT（第一个 token）≈ ${ttft} ms；总耗时 ≈ ${Date.now() - t0} ms`);
return;
}
}
}
}
console.log("🔹 自测 A：POST /chat（fetch + ReadableStream，请求体里带 JSON）");
const respA = await fetch(`${base}/chat`, {
method: "POST",
headers: { "Content-Type": "application/json", Accept: "text/event-stream" },
body: JSON.stringify({ message: "用一句话说明什么是 SSE，不要超过 30 个字。", thread_id: "sse-demo-1" }),
});
console.log(` 响应状态：HTTP ${respA.status}，Content-Type = ${respA.headers.get("content-type")}\n`);
await readSseStream(respA);
// 若本机真有全局 EventSource，再演示 GET 路径
if (hasEventSource) {
console.log("\n🔹 自测 B：内置 EventSource 走 GET /chat（浏览器里的原生写法）");
const ES = (globalThis as unknown as { EventSource: new (u: string) => EventSource }).EventSource;
await new Promise<void>((resolve) => {
const url = `${base}/chat?message=${encodeURIComponent("再打个招呼，一句话。")}&thread_id=sse-demo-2`;
const es = new ES(url);
let acc = "";
for (const ev of ["reasoning", "token", "done", "error"]) {
es.addEventListener(ev, (e: MessageEvent) => {
const payload = JSON.parse(e.data) as Record<string, unknown>;
if (ev === "token") acc += String(payload.text);
else console.log(` [EventSource:${ev}] ${JSON.stringify(payload)}`);
if (ev === "done" || ev === "error") {
console.log(` [EventSource] 拼出的正文：${acc}`);
es.close(); // ★ 收到 done 必须 close，否则 EventSource 会自动重连、进程不退出
resolve();
}
});
}
setTimeout(() => { es.close(); resolve(); }, 20000); // 兜底
});
} else {
console.log(
"\n🔹 自测 B：（跳过）本机没有全局 EventSource。\n" +
" 浏览器里等价写法：\n" +
' const es = new EventSource("/chat?message=...&thread_id=...");\n' +
' es.addEventListener("token", (e) => { out.textContent += JSON.parse(e.data).text; });\n' +
' es.addEventListener("done", () => es.close());'
);
}
await new Promise<void>((resolve, reject) =>
server.close((err) => (err ? reject(err) : resolve()))
);
console.log("\n🛑 服务已关闭，示例结束。");
}
main().catch((e) => {
console.error("❌ 运行失败：", e);
process.exit(1);
});运行：
bash
npx tsx 14-frontend-integration/code/01-sse-backend.ts真实输出：
========================================================================
示例 1：SSE 流式后端（node:http + graph.stream streamMode:"messages"）
========================================================================
✅ 服务已启动：http://127.0.0.1:8790（跑完会自己关掉）
🔎 环境探测：
typeof EventSource = undefined
→ 本机没有内置 EventSource，改用 fetch + ReadableStream 手工解析（下面的自测）
ℹ️ 浏览器里的 EventSource 只能发 GET。所以本示例额外保留了 GET /chat：
EventSource 打 GET；fetch 打 POST（带 JSON body）。
🔹 自测 A：POST /chat（fetch + ReadableStream，请求体里带 JSON）
[server] POST /chat
响应状态：HTTP 200，Content-Type = text/event-stream; charset=utf-8
📨 前端收到并解析出的帧（每帧一段）：
--------------------------------------------------------
[event: reasoning] {"node":"chat","stage":"thinking"}
[event: token ] {"node":"chat","text":"SS"}
[event: token ] {"node":"chat","text":"E"}
[event: token ] {"node":"chat","text":"是"}
[event: token ] {"node":"chat","text":"服务器"}
[event: token ] {"node":"chat","text":"通过"}
[event: token ] {"node":"chat","text":"HTTP"}
[event: token ] {"node":"chat","text":"向"}
[event: token ] {"node":"chat","text":"浏览器"}
[event: token ] {"node":"chat","text":"单向"}
[event: token ] {"node":"chat","text":"持续"}
[event: token ] {"node":"chat","text":"推送"}
[event: token ] {"node":"chat","text":"事件"}
[event: token ] {"node":"chat","text":"的技术"}
[event: token ] {"node":"chat","text":"。"}
[event: done ] {"ok":true,"thread_id":"sse-demo-1"}
--------------------------------------------------------
📊 共收到 16 帧；TTFT（第一个 token）≈ 1356 ms；总耗时 ≈ 1392 ms
🔹 自测 B：（跳过）本机没有全局 EventSource。
浏览器里等价写法：
const es = new EventSource("/chat?message=...&thread_id=...");
es.addEventListener("token", (e) => { out.textContent += JSON.parse(e.data).text; });
es.addEventListener("done", () => es.close());
🛑 服务已关闭，示例结束。四个必须记住的细节
① 响应头 text/event-stream 是"通关文牒"
没有它，客户端根本不会把响应当流。Cache-Control: no-cache 和 X-Accel-Buffering: no 是给中间的反向代理（nginx 等）看的——否则它们可能把你的流攒成一大坨再吐给前端，流式就白做了。
② 一帧 = event: 行 +（可选的 data: 行）+ 一个空行
res.write() 是流式写：写进去的内容会立刻发给客户端，不需要等 res.end()。这就是"边生成边到货"在 HTTP 层的实现。
③ EventSource 只能发 GET，所以真实项目常常两个端点都留
浏览器原生的 EventSource 不支持自定义请求体，只能 GET，参数只能塞在 URL 里。而 fetch 可以发 POST + JSON body。所以工程上常见的做法是：POST /chat（fetch 用，带 JSON）+ GET /chat（EventSource 用，参数在 query），两者共用同一段"事件 → SSE 帧"的逻辑。示例 1 两个都留了，你可以看到它们都调用同一个 streamChatToClient()。
💡 本机环境：
typeof EventSource = undefined——本机这套 Node 里没有内置的全局EventSource。所以示例用fetch+ReadableStream手工解析（这也最能看清协议）。浏览器里则直接用EventSource；若哪天本机 Node 提供了全局EventSource，示例的"自测 B"分支会自动改走原生路径。
④ 别忘了 done 和 error 两个收尾事件
流式响应如果没有明确的"结束"信号，前端只能猜什么时候停。所以约定两帧：event: done（正常结束）和 event: error（出错）。前端收到 done 就 es.close()——这一步是必须的：EventSource 默认会在连接断开后自动重连，不手动 close() 的话，聊天界面会在回答结束后不停地重新请求。
三、前后端事件协议
示例 2：前后端事件协议
示例 1 能跑了，但它只推了 token 一种事件。真实前端面对的是好几种事件（token、进度、节点完成、结束、出错、审批……）。如果前后端没有一份写死的契约，前端就得靠猜和试错。
示例 2 的做法：真跑一次多模式 stream，把每种模式的真实形状抓下来，自动推导出一张「协议清单表」，并打印一份可直接复制的 TypeScript 类型定义。前端同学拿着这份类型，就能写出类型安全的消费代码。
完整代码（code/02-event-protocol.ts）
typescript
/**
* 示例 2：前后端事件协议 —— 把每种事件的真实形状抓下来，生成"协议清单"
*
* 前端同学最想要的一件事是：后端到底会推哪些事件、每个长什么样？
* 本示例真跑一次多模式 stream（streamMode 数组），把每种模式的真实 JSON 形状
* 抓下来，生成一张"协议清单表"，并打印一份可直接复制的 TypeScript 类型定义。
* 这样前后端就有一份共同契约，不用靠猜。
*
* ⚠️ 数组模式下每个 chunk 恒为二元组 [模式名, 数据]，看第 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));
// 一个"形状描述器"：从真实数据里推导结构，而不是我手写猜的
function shape(v: unknown): string {
if (v === null) return "null";
if (v === undefined) return "undefined";
if (Array.isArray(v)) return v.length === 0 ? "[]" : `[${shape(v[0])}]`;
if (typeof v !== "object") return typeof v;
// 消息对象（BaseMessage 子类）单独归纳，别把一堆内部字段摊出来
const maybeMsg = v as { _getType?: () => string };
if (typeof maybeMsg._getType === "function") return `Message<${maybeMsg._getType()}>`;
const keys = Object.keys(v as Record<string, unknown>);
if (keys.length === 0) return "{}";
return `{ ${keys.map((k) => `${k}: ${shape((v as Record<string, unknown>)[k])}`).join(", ")} }`;
}
async function main() {
banner("示例 2：把每种事件的真实形状抓下来（协议清单）");
assertChatKey();
const model = createModel({ temperature: 0 });
// 节点 1：plan —— 用 getWriter() 上报自定义进度（对应 custom 模式）
async function plan(state: typeof MessagesAnnotation.State) {
const write = getWriter();
for (const step of ["分析意图", "选择策略", "生成计划"]) {
await sleep(80);
write?.({ type: "step", label: `规划中：${step}` });
}
return { messages: [new AIMessage("（内部）计划：直接作答。")] };
}
// 节点 2：answer —— 调模型，触发 messages 模式
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 = "用一句话介绍事件协议。";
console.log(`👤 提问：${question}\n`);
// 每种模式各留一条"样本"，用于推导形状
const sample: Record<string, unknown> = {};
const counts: Record<string, number> = {};
console.log("🔀 正在跑多模式流（values + updates + messages + custom）…\n");
for await (const chunk of await graph.stream(
{ messages: [new HumanMessage(question)] },
{ streamMode: ["values", "updates", "messages", "custom"] }
)) {
const [mode, data] = chunk as unknown as [string, unknown];
counts[mode] = (counts[mode] ?? 0) + 1;
if (!(mode in sample)) sample[mode] = data; // 只留第一条，避免 token 刷爆内存
}
console.log("📊 本次运行各模式事件条数：");
for (const mode of ["values", "updates", "messages", "custom"]) {
console.log(` • ${mode.padEnd(9)} × ${counts[mode] ?? 0}`);
}
// 把 messages 样本整理成"人看得懂"的形状
function messagesSample(data: unknown): string {
const [msg, meta] = data as [AIMessageChunk, Record<string, unknown>];
const reasoning = (msg.additional_kwargs as Record<string, unknown>)?.reasoning_content;
return JSON.stringify({
content: typeof msg.content === "string" ? "string" : "array",
additional_kwargs: {
reasoning_content: typeof reasoning === "string" ? "string（仅推理型模型，可能没有）" : "undefined",
},
metadata: { langgraph_node: meta.langgraph_node, langgraph_step: meta.langgraph_step },
});
}
console.log("\n📋 协议清单表（从本次运行的真实事件里推导）：\n");
console.log(" ┌─ 模式 ─────┬─ chunk 形状（真实推导）─────────────────────────────────────────┐");
console.log(` │ values │ ${shape(sample.values).slice(0, 70)}`);
console.log(` │ updates │ ${shape(sample.updates).slice(0, 70)}`);
console.log(` │ messages │ 二元组 -> ${messagesSample(sample.messages).slice(0, 58)}`);
console.log(` │ custom │ ${shape(sample.custom).slice(0, 70)}`);
console.log(" └───────────┴──────────────────────────────────────────────────────────────┘");
console.log("\n ↑ 以上是【数据】部分的形状（二元组的第 1 个元素）。");
const types = `// ── 前后端事件协议类型定义（贴进前端项目即可）──────────────────────
/** SSE 帧的通用外壳：event 名 + JSON 数据 */
export interface SseFrame<T = unknown> {
event: "token" | "reasoning" | "node" | "progress" | "done" | "error" | "interrupt";
data: T;
}
/** streamMode:"messages" —— token 级增量 */
export interface TokenEvent {
node: string; // 来自哪个节点（metadata.langgraph_node）
text: string; // 正文片段；累积所有 text 就是完整回答
}
/** streamMode:"custom" —— 节点里 getWriter() 主动上报的业务进度 */
export interface ProgressEvent { [key: string]: unknown }
/** streamMode:"updates" —— 某节点执行完毕，key 是节点名 */
export type UpdatesEvent = Record<string, Record<string, unknown> | undefined>;
/** streamMode:"values" —— 完整状态快照（这里演示 messages 状态） */
export interface ValuesEvent { messages: Array<{ type: string; content: string }> }
/** 结束 / 出错 */
export interface DoneEvent { ok: true; thread_id: string }
export interface ErrorEvent { message: string }
/** 人工介入（见示例 3）：图在 interrupt 处停下，后端把审批卡片推给前端 */
export interface InterruptEvent { value: unknown; next: string[] }`;
console.log("\n🧩 可直接复制的 TypeScript 类型定义：\n");
console.log("─".repeat(72));
console.log(types);
console.log("─".repeat(72));
}
main().catch((e) => {
console.error("❌ 运行失败：", e);
process.exit(1);
});运行：
bash
npx tsx 14-frontend-integration/code/02-event-protocol.ts真实输出：
========================================================================
示例 2：把每种事件的真实形状抓下来（协议清单）
========================================================================
👤 提问：用一句话介绍事件协议。
🔀 正在跑多模式流（values + updates + messages + custom）…
📊 本次运行各模式事件条数：
• values × 3
• updates × 2
• messages × 536
• custom × 3
📋 协议清单表（从本次运行的真实事件里推导）：
┌─ 模式 ─────┬─ chunk 形状（真实推导）─────────────────────────────────────────┐
│ values │ { messages: [Message<human>] }
│ updates │ { plan: { messages: [Message<ai>] } }
│ messages │ 二元组 -> {"content":"string","additional_kwargs":{"reasoning_conten
│ custom │ { type: string, label: string }
└───────────┴──────────────────────────────────────────────────────────────┘
↑ 以上是【数据】部分的形状（二元组的第 1 个元素）。
🧩 可直接复制的 TypeScript 类型定义：
────────────────────────────────────────────────────────────────────────
// ── 前后端事件协议类型定义（贴进前端项目即可）──────────────────────
/** SSE 帧的通用外壳：event 名 + JSON 数据 */
export interface SseFrame<T = unknown> {
event: "token" | "reasoning" | "node" | "progress" | "done" | "error" | "interrupt";
data: T;
}
/** streamMode:"messages" —— token 级增量 */
export interface TokenEvent {
node: string; // 来自哪个节点（metadata.langgraph_node）
text: string; // 正文片段；累积所有 text 就是完整回答
}
/** streamMode:"custom" —— 节点里 getWriter() 主动上报的业务进度 */
export interface ProgressEvent { [key: string]: unknown }
/** streamMode:"updates" —— 某节点执行完毕，key 是节点名 */
export type UpdatesEvent = Record<string, Record<string, unknown> | undefined>;
/** streamMode:"values" —— 完整状态快照（这里演示 messages 状态） */
export interface ValuesEvent { messages: Array<{ type: string; content: string }> }
/** 结束 / 出错 */
export interface DoneEvent { ok: true; thread_id: string }
export interface ErrorEvent { message: string }
/** 人工介入（见示例 3）：图在 interrupt 处停下，后端把审批卡片推给前端 */
export interface InterruptEvent { value: unknown; next: string[] }
────────────────────────────────────────────────────────────────────────这张表怎么用
示例 2 的价值不在"跑一次"，而在方法：它用同一个 shape() 函数从真实的运行时数据推导结构，而不是拍脑袋写。你可以把这个脚本放进自己的项目，让 CI 定期跑一次，一旦框架升级导致事件形状变化，就能第一时间发现。
把上面四种模式的数据部分，加上示例 1 里的 token / reasoning / done / error，合起来就是一份完整的事件协议：
| event | 什么时候发 | data 形状 | 前端怎么用 |
|---|---|---|---|
token | 模型产出正文 token | { node, text } | 追加到气泡（打字机） |
reasoning | 推理模型开始思考 | { node, stage } | 显示"正在思考…" |
node | 某节点执行完毕（updates） | { 节点名: {...} } | 更新进度/步骤条 |
progress | 节点主动上报（custom） | 业务自定义 | 进度条、阶段提示 |
done | 流正常结束 | { ok, thread_id } | 关闭 EventSource |
error | 出错 | { message } | 弹错误提示 |
interrupt | 图暂停等审批（示例 3） | { value, next } | 弹审批卡片 |
💡 为什么建议后端把每种模式"统一包装"成
{ event, data }再发？因为streamMode数组吐出来的messages是[AIMessageChunk, metadata]二元组、values是整个状态对象——形状五花八门。后端在写 SSE 时顺手归一化成一个统一的帧外壳（event名 + 干净的data），前端就只需要认这一种形状。示例 1 里的sseFrame()干的就是这件事。
四、中断与审批怎么传到前端
示例 3：中断与审批怎么传到前端
第 5 章我们学过 interrupt / Command({ resume })，但那是在同一段脚本里完成的。真实产品里，人类坐在浏览器前——图在服务器上跑到 interrupt 停下，服务器必须把"审批卡片"发回前端；用户点完按钮，前端再发一个请求让图继续。
示例 3 用两个 HTTP 接口把这条前后端契约讲清楚：
POST /chat { topic, thread_id } → 跑到 interrupt 停下，响应带审批卡片
POST /resume { thread_id, resume } → 人类答复作为 resume 值，续跑到结束完整代码（code/03-hitl-over-http.ts）
typescript
/**
* 示例 3：通过 HTTP 做人工介入（HITL 的前后端契约）
*
* 第 5 章讲了 interrupt / Command 的机制，但那是"同一台机器、同一段脚本里"。
* 真实产品里人类坐在浏览器前：图在服务器上跑到 interrupt 停下，服务器必须把
* "审批卡片"推回前端；用户点了"批准/拒绝"，前端再发请求让图继续。
*
* 本示例用两个 HTTP 接口表达这套契约：
* POST /chat { topic, thread_id } → 图跑到 interrupt 停下，响应带审批卡片
* POST /resume { thread_id, resume } → 人类答复作为 resume 值，续跑到结束
* 自己起服务、自己发两次真实请求、打印完整收发，最后关服务退出。
*/
import { createServer, type IncomingMessage, type ServerResponse } from "node:http";
import {
StateGraph, START, END, Annotation, MemorySaver,
Command, interrupt, isInterrupted, type Interrupt,
} from "@langchain/langgraph";
import { banner } from "../../lib/model.js";
const PORT = 8791;
// 状态：话题、草稿、人类决定、定稿
const ReviewState = Annotation.Root({
topic: Annotation<string>,
draft: Annotation<string>,
decision: Annotation<string>,
finalText: Annotation<string>,
});
function writeDraft(state: typeof ReviewState.State) {
console.log(" [节点 writeDraft] 正在起草…");
return { draft: `《${state.topic}》初稿：一段自动生成的内容，等待人工确认。` };
}
// 人工审核 —— interrupt 就发生在这里
function review(state: typeof ReviewState.State) {
console.log(" [节点 review] 进入 review（interrupt 首次暂停 / 恢复后返回人类答复）");
// 第一次跑到这里：抛出信号、暂停，把 payload 作为"审批卡片"交给前端
// 恢复后重跑这里：不再暂停，直接把人类的 resume 值返回
const humanAnswer = interrupt({
kind: "approval_request",
question: "这份草稿可以发布吗？请回复 approve 或 reject。",
draft: state.draft,
});
console.log(` [节点 review] 收到人类答复：${JSON.stringify(humanAnswer)}`);
return { decision: String(humanAnswer) };
}
function finalize(state: typeof ReviewState.State) {
const approved = state.decision.toLowerCase().includes("approve");
console.log(` [节点 finalize] 决定=${state.decision} → ${approved ? "发布" : "退回"}`);
return {
finalText: approved
? `✅ 已发布：${state.draft}`
: `❌ 已退回，未发布：${state.draft}`,
};
}
async function readJson(req: IncomingMessage): Promise<Record<string, unknown>> {
const chunks: Buffer[] = [];
for await (const chunk of req) chunks.push(chunk as Buffer);
const raw = Buffer.concat(chunks).toString("utf8");
if (!raw) return {};
return JSON.parse(raw) as Record<string, unknown>;
}
function sendJson(res: ServerResponse, status: number, body: unknown) {
const payload = JSON.stringify(body);
res.writeHead(status, { "Content-Type": "application/json; charset=utf-8", "Content-Length": Buffer.byteLength(payload) });
res.end(payload);
}
async function main() {
banner("示例 3：通过 HTTP 做人工介入（POST /chat → 暂停 → POST /resume）");
const graph = new StateGraph(ReviewState)
.addNode("writeDraft", writeDraft)
.addNode("review", review)
.addNode("finalize", finalize)
.addEdge(START, "writeDraft")
.addEdge("writeDraft", "review")
.addEdge("review", "finalize")
.addEdge("finalize", END)
.compile({ checkpointer: new MemorySaver() });
const server = createServer(async (req, res) => {
const url = new URL(req.url ?? "/", `http://localhost:${PORT}`);
const method = req.method ?? "GET";
console.log(` [server] ${method} ${url.pathname}`);
try {
if (method === "GET" && url.pathname === "/health") return sendJson(res, 200, { status: "ok" });
// POST /chat：跑到 interrupt 停下，把审批卡片作为响应返回
if (method === "POST" && url.pathname === "/chat") {
const body = await readJson(req);
const topic = body.topic;
const threadId = body.thread_id;
if (typeof topic !== "string" || typeof threadId !== "string") {
return sendJson(res, 400, { error: "需要 { topic: string, thread_id: string }" });
}
const config = { configurable: { thread_id: threadId } };
const result = await graph.invoke({ topic }, config);
if (isInterrupted(result)) {
const itr = result.__interrupt__[0] as Interrupt;
const state = await graph.getState(config);
return sendJson(res, 200, {
status: "interrupted",
thread_id: threadId,
interrupt: itr.value, // ★ 前端拿这个去渲染审批卡片
next: state.next, // 暂停在哪些节点
});
}
return sendJson(res, 200, { status: "done", thread_id: threadId, finalText: result.finalText });
}
// POST /resume：人类答复作为 Command({ resume }) 继续
if (method === "POST" && url.pathname === "/resume") {
const body = await readJson(req);
const threadId = body.thread_id;
const resumeValue = body.resume;
if (typeof threadId !== "string" || resumeValue === undefined) {
return sendJson(res, 400, { error: "需要 { thread_id: string, resume: any }" });
}
const config = { configurable: { thread_id: threadId } };
const result = await graph.invoke(new Command({ resume: resumeValue }), config);
return sendJson(res, 200, {
status: "done", thread_id: threadId,
finalText: result.finalText, decision: result.decision,
});
}
return sendJson(res, 404, { error: `未知路径：${method} ${url.pathname}` });
} catch (e) {
console.error(" [server] 出错：", (e as Error).message);
return sendJson(res, 500, { error: (e as Error).message });
}
});
await new Promise<void>((resolve) => server.listen(PORT, resolve));
const addr = server.address();
const actualPort = typeof addr === "object" && addr ? addr.port : PORT;
const base = `http://127.0.0.1:${actualPort}`;
console.log(`✅ 服务已启动：${base}\n`);
// 发一个请求，把【请求 + 响应】完整打印出来
async function call(label: string, path: string, body: unknown): Promise<Record<string, unknown>> {
console.log(`\n🔹 ${label}`);
console.log(` 请求 → POST ${path}`);
console.log(` ${JSON.stringify(body)}`);
const resp = await fetch(`${base}${path}`, {
method: "POST",
headers: { "Content-Type": "application/json" },
body: JSON.stringify(body),
});
const json = (await resp.json()) as Record<string, unknown>;
console.log(` 响应 ← [HTTP ${resp.status}]`);
console.log(` ${JSON.stringify(json, null, 2).split("\n").join("\n ")}`);
return json;
}
// 场景 A：人类批准
console.log("═══════════ 场景 A：人类批准 ═══════════");
await call("① 前端发起对话（后端跑到 interrupt 停下，返回审批卡片）", "/chat", {
topic: "LangGraph 前端集成", thread_id: "http-hitl-A",
});
await call("② 用户点了「批准」→ 前端带 resume=approve 继续", "/resume", {
thread_id: "http-hitl-A", resume: "approve",
});
// 场景 B：人类拒绝（换一个会话）
console.log("\n═══════════ 场景 B：人类拒绝 ═══════════");
await call("① 前端发起对话（另一个 thread_id）", "/chat", {
topic: "LangGraph 前端集成", thread_id: "http-hitl-B",
});
await call("② 用户点了「拒绝」→ 前端带 resume=reject 继续", "/resume", {
thread_id: "http-hitl-B", resume: "reject",
});
await new Promise<void>((resolve, reject) =>
server.close((err) => (err ? reject(err) : resolve()))
);
console.log("\n🛑 服务已关闭，示例结束。");
}
main().catch((e) => {
console.error("❌ 运行失败：", e);
process.exit(1);
});运行：
bash
npx tsx 14-frontend-integration/code/03-hitl-over-http.ts真实输出：
========================================================================
示例 3：通过 HTTP 做人工介入（POST /chat → 暂停 → POST /resume）
========================================================================
✅ 服务已启动：http://127.0.0.1:8791
═══════════ 场景 A：人类批准 ═══════════
🔹 ① 前端发起对话（后端跑到 interrupt 停下，返回审批卡片）
请求 → POST /chat
{"topic":"LangGraph 前端集成","thread_id":"http-hitl-A"}
[server] POST /chat
[节点 writeDraft] 正在起草…
[节点 review] 进入 review（interrupt 首次暂停 / 恢复后返回人类答复）
响应 ← [HTTP 200]
{
"status": "interrupted",
"thread_id": "http-hitl-A",
"interrupt": {
"kind": "approval_request",
"question": "这份草稿可以发布吗？请回复 approve 或 reject。",
"draft": "《LangGraph 前端集成》初稿：一段自动生成的内容，等待人工确认。"
},
"next": [
"review"
]
}
🔹 ② 用户点了「批准」→ 前端带 resume=approve 继续
请求 → POST /resume
{"thread_id":"http-hitl-A","resume":"approve"}
[server] POST /resume
[节点 review] 进入 review（interrupt 首次暂停 / 恢复后返回人类答复）
[节点 review] 收到人类答复："approve"
[节点 finalize] 决定=approve → 发布
响应 ← [HTTP 200]
{
"status": "done",
"thread_id": "http-hitl-A",
"finalText": "✅ 已发布：《LangGraph 前端集成》初稿：一段自动生成的内容，等待人工确认。",
"decision": "approve"
}
═══════════ 场景 B：人类拒绝 ═══════════
🔹 ① 前端发起对话（另一个 thread_id）
请求 → POST /chat
{"topic":"LangGraph 前端集成","thread_id":"http-hitl-B"}
[server] POST /chat
[节点 writeDraft] 正在起草…
[节点 review] 进入 review（interrupt 首次暂停 / 恢复后返回人类答复）
响应 ← [HTTP 200]
{
"status": "interrupted",
"thread_id": "http-hitl-B",
"interrupt": {
"kind": "approval_request",
"question": "这份草稿可以发布吗？请回复 approve 或 reject。",
"draft": "《LangGraph 前端集成》初稿：一段自动生成的内容，等待人工确认。"
},
"next": [
"review"
]
}
🔹 ② 用户点了「拒绝」→ 前端带 resume=reject 继续
请求 → POST /resume
{"thread_id":"http-hitl-B","resume":"reject"}
[server] POST /resume
[节点 review] 进入 review（interrupt 首次暂停 / 恢复后返回人类答复）
[节点 review] 收到人类答复："reject"
[节点 finalize] 决定=reject → 退回
响应 ← [HTTP 200]
{
"status": "done",
"thread_id": "http-hitl-B",
"finalText": "❌ 已退回，未发布：《LangGraph 前端集成》初稿：一段自动生成的内容，等待人工确认。",
"decision": "reject"
}
🛑 服务已关闭，示例结束。三个要点
① POST /chat 的响应就是"审批卡片"
图跑到 interrupt 时会正常返回（不是报错），返回的对象带 __interrupt__，我们用 isInterrupted() 判断后，把 interrupt.value（也就是节点里传给 interrupt() 的那个 payload）原样放进 HTTP 响应的 interrupt 字段。前端拿到的这个 value，就是它要渲染的审批卡片数据——kind、question、draft 全在里面。同时还返回 next（暂停在哪些节点），前端可以据此显示"流程卡在哪一步"。
② POST /resume 就是把人类的按钮点击变成 Command({ resume })
用户在卡片上点了「批准」，前端就发 { thread_id, resume: "approve" }。后端 graph.invoke(new Command({ resume: resumeValue }), config)——resume 的值会原样成为 review 节点里 interrupt() 的返回值，图就从断点续跑。这正是第 5 章的机制，只不过"人类"从同一台机器换成了浏览器。
③ 会话靠 thread_id 粘起来，MemorySaver 负责记住"暂停在哪"
POST /chat 暂停时，状态被存进 MemorySaver（checkpointer）。POST /resume 只要带上同一个 thread_id，就能找回那个暂停点继续跑。所以前端必须在两次请求之间保存 thread_id（通常放在 state 或 URL 里）。
💡 一个真实观察：看输出的日志，
POST /resume时[节点 review]那一行又打印了一次——这不是 bug，而是第 5 章讲过的规则：恢复时中断的那个节点会从头整体重跑一遍（这次interrupt()直接返回人类的答复，不再暂停）。所以interrupt()之前的代码必须无副作用、可重复执行。
五、React 接入（可复制片段，未实跑）
示例 4：React 接入（可复制片段，未实跑）
⚠️ 先说清楚：这一段没有在本项目实跑
┌──────────────────────────────────────────────────────────────────┐
│ 本机未安装 React，此段未在本项目实跑，仅作可复制参考。 │
└──────────────────────────────────────────────────────────────────┘本机不能安装 React，也没法在浏览器里跑前端。所以示例 4 的脚本本身是纯打印——它把"可以直接复制到 React 项目里"的三段代码作为字符串打印出来。脚本能跑通（你下面能看到它的真实输出），但打印出来的 React 代码本身没有运行过。请把这当作"可复制的参考片段"，而不是"已验证的成品"。
完整代码（code/04-react-snippets.ts）
typescript
/**
* 示例 4：可直接复制到 React 项目的前端代码（两版）
*
* ⚠️⚠️⚠️ 重要声明 ⚠️⚠️⚠️
* 本机没有安装 React，也不能在浏览器里运行前端。所以本文件**不实跑任何 React 代码**，
* 它只是把"可以复制到 React 项目里"的代码**作为字符串打印出来**。这样脚本本身能安全
* 跑通（纯打印），你把打印出来的代码复制走就能用。
*
* ┌──────────────────────────────────────────────────────────────────┐
* │ 本机未安装 React，此段未在本项目实跑，仅作可复制参考。 │
* └──────────────────────────────────────────────────────────────────┘
*
* 三版：① 官方 SDK 的 useStream() ② 裸 EventSource 消费示例 1 的 SSE
* ③ 人工审批卡片（消费示例 3 的 POST /chat → /resume 契约）
*/
import { banner } from "../../lib/model.js";
const SNIPPET_SDK = `// ─────────────────────────────────────────────────────────────
// React 片段 ①：用官方 @langchain/langgraph-sdk 的 useStream()
// 安装：npm i @langchain/langgraph-sdk react
// 前提：后端是 LangGraph Platform / 兼容的 SSE 端点（事件协议见示例 2）
// ─────────────────────────────────────────────────────────────
import { useStream } from "@langchain/langgraph-sdk/react";
export function Chat() {
// useStream 内部就是走 SSE，替你把 messages / values / isLoading 都管好了
const { messages, values, isLoading, submit, stop } = useStream({
apiUrl: "http://localhost:2024", // 你的后端地址
assistantId: "agent",
// threadId 不传时 SDK 自动新建会话；持久化会话就传固定 threadId
});
return (
<div>
{messages.map((m) => (
<div key={m.id}>
<b>{m.type}:</b> {typeof m.content === "string" ? m.content : ""}
</div>
))}
{/* isLoading 为 true 时按钮变成"停止" */}
<button onClick={() => (isLoading ? stop() : submit({ messages: [] }))}>
{isLoading ? "停止" : "发送"}
</button>
{/* values 里也能直接读到最终状态，比如自定义的 finalText 字段 */}
<pre>{JSON.stringify(values, null, 2)}</pre>
</div>
);
}`;
const SNIPPET_EVENTSOURCE = `// ─────────────────────────────────────────────────────────────
// React 片段 ②：不依赖任何 SDK，裸用浏览器原生 EventSource
// 消费示例 1 后端推的 SSE 协议：event: token / done / error ...
// 注意：EventSource 只能发 GET，所以后端要保留 GET /chat（示例 1 已保留）
// ─────────────────────────────────────────────────────────────
import { useEffect, useRef, useState } from "react";
export function Chat({ threadId }: { threadId: string }) {
const [text, setText] = useState(""); // 累积中的回答
const [thinking, setThinking] = useState(false);
const esRef = useRef<EventSource | null>(null);
function ask(question: string) {
setText("");
setThinking(false);
// 带上 query 参数（EventSource 不支持自定义 body）
const url = \`/chat?message=\${encodeURIComponent(question)}&thread_id=\${threadId}\`;
const es = new EventSource(url);
esRef.current = es;
es.addEventListener("reasoning", () => setThinking(true)); // 模型开始思考
es.addEventListener("token", (e: MessageEvent) => { // 主体：往界面追加
const { text: piece } = JSON.parse(e.data);
setText((prev) => prev + piece);
});
es.addEventListener("done", () => { // 必须 close，否则会自动重连
setThinking(false);
es.close();
});
es.addEventListener("error", () => es.close());
}
useEffect(() => () => esRef.current?.close(), []); // 卸载时断开
return (
<div>
{thinking && <small>🧠 模型正在思考…</small>}
<p>{text}</p>
<button onClick={() => ask("你好，介绍一下你自己。")}>发送</button>
</div>
);
}`;
const SNIPPET_APPROVAL = `// ─────────────────────────────────────────────────────────────
// React 片段 ③：人工审批卡片（消费示例 3 的 POST /chat → /resume 契约）
// 后端 POST /chat 返回 { status:"interrupted", interrupt:{...}, next:[...] }，
// 前端据此弹出审批卡片；用户点按钮后再 POST /resume。
// ─────────────────────────────────────────────────────────────
import { useState } from "react";
export function ApprovalCard({ threadId }: { threadId: string }) {
const [pending, setPending] = useState<null | { question: string; draft: string }>(null);
const [finalText, setFinalText] = useState("");
async function start() {
const r = await fetch("/chat", {
method: "POST",
headers: { "Content-Type": "application/json" },
body: JSON.stringify({ topic: "LangGraph 前端集成", thread_id: threadId }),
});
const j = await r.json();
if (j.status === "interrupted") setPending(j.interrupt); // 弹出审批卡片
else setFinalText(j.finalText);
}
async function decide(resume: "approve" | "reject") {
setPending(null);
const r = await fetch("/resume", {
method: "POST",
headers: { "Content-Type": "application/json" },
body: JSON.stringify({ thread_id: threadId, resume }),
});
const j = await r.json();
setFinalText(j.finalText);
}
return (
<div>
<button onClick={start}>发起需要审批的任务</button>
{pending && (
<div className="card">
<p>{pending.question}</p>
<pre>{pending.draft}</pre>
<button onClick={() => decide("approve")}>✅ 批准</button>
<button onClick={() => decide("reject")}>❌ 拒绝</button>
</div>
)}
{finalText && <p>{finalText}</p>}
</div>
);
}`;
function main() {
banner("示例 4：可复制的 React 片段（本机未安装 React，未实跑，仅参考）");
console.log("⚠️ 再次强调：下面的代码没有在本项目里运行过——本机既没装 React，也没法跑");
console.log(" 浏览器。它们是「可以复制到你自己的 React 项目里」的参考代码。本脚本本身");
console.log(" 能跑，是因为它只是把这些代码当字符串打印出来。\n");
const blocks: Array<[string, string]> = [
["① 官方 SDK 版：useStream()", SNIPPET_SDK],
["② 裸 EventSource 版：零依赖消费 SSE", SNIPPET_EVENTSOURCE],
["③ 人工审批卡片：消费 HITL 的 HTTP 契约", SNIPPET_APPROVAL],
];
for (const [title, code] of blocks) {
console.log("╔" + "═".repeat(70) + "╗");
console.log(`║ ${title}`);
console.log("╚" + "═".repeat(70) + "╝");
console.log(code);
console.log("\n");
}
console.log("─".repeat(72));
console.log("📦 官方现成的 UI 生态（不用从零写）：");
console.log(" • Agent Chat UI —— 开箱即用的聊天前端，接上后端就能聊");
console.log(" • langgraphjs-gen-ui-examples —— 生成式 UI（generative UI）官方示例集合");
console.log(" • @langchain/langgraph-sdk —— 提供 React / Vue / Svelte / Angular 各框架绑定");
console.log(" • useStream() —— 最省事的钩子，内部走 SSE，事件协议同本书");
console.log("─".repeat(72));
}
main();运行：
bash
npx tsx 14-frontend-integration/code/04-react-snippets.ts真实输出（节选：中间三段 ╔…╗ 之间的内容，就是上面源码里 SNIPPET_SDK / SNIPPET_EVENTSOURCE / SNIPPET_APPROVAL 三个字符串常量原样打印的结果，与源码逐字一致，此处不再重复粘贴）：
========================================================================
示例 4：可复制的 React 片段（本机未安装 React，未实跑，仅参考）
========================================================================
⚠️ 再次强调：下面的代码没有在本项目里运行过——本机既没装 React，也没法跑
浏览器。它们是「可以复制到你自己的 React 项目里」的参考代码。本脚本本身
能跑，是因为它只是把这些代码当字符串打印出来。
╔══════════════════════════════════════════════════════════════════════╗
║ ① 官方 SDK 版：useStream()
╚══════════════════════════════════════════════════════════════════════╝
…（此处为输出节选：该标题下，是 SNIPPET_SDK 常量原样打印的 React 源码）…
╔══════════════════════════════════════════════════════════════════════╗
║ ② 裸 EventSource 版：零依赖消费 SSE
╚══════════════════════════════════════════════════════════════════════╝
…（此处为输出节选：该标题下，是 SNIPPET_EVENTSOURCE 常量原样打印的 React 源码）…
╔══════════════════════════════════════════════════════════════════════╗
║ ③ 人工审批卡片：消费 HITL 的 HTTP 契约
╚══════════════════════════════════════════════════════════════════════╝
…（此处为输出节选：该标题下，是 SNIPPET_APPROVAL 常量原样打印的 React 源码）…
────────────────────────────────────────────────────────────────────────
📦 官方现成的 UI 生态（不用从零写）：
• Agent Chat UI —— 开箱即用的聊天前端，接上后端就能聊
• langgraphjs-gen-ui-examples —— 生成式 UI（generative UI）官方示例集合
• @langchain/langgraph-sdk —— 提供 React / Vue / Svelte / Angular 各框架绑定
• useStream() —— 最省事的钩子，内部走 SSE，事件协议同本书
────────────────────────────────────────────────────────────────────────三段片段分别讲了什么
| 片段 | 依赖 | 用在哪 | 亮点 |
|---|---|---|---|
① useStream() | 官方 @langchain/langgraph-sdk | 生产首选 | messages / values / isLoading / submit / stop 全给你了 |
② 裸 EventSource | 零依赖 | 想自己掌控 / 非 React 环境 | 直接消费示例 1 的 SSE 协议，最能看清原理 |
| ③ 审批卡片 | 零依赖 | HITL 场景 | 消费示例 3 的 POST /chat → POST /resume 契约 |
片段 ①（useStream）：这是生产环境最省事的写法。它内部就是一条 SSE 连接，把"累积消息、维护 loading 状态、处理中断重连"这些脏活全包了，你只需要渲染 messages。它要求后端是一个 LangGraph Platform / 兼容的端点——如果你的后端是自己用 node:http 手写的（像本章示例 1），那就用片段 ②。
片段 ②（裸 EventSource）：不依赖任何 SDK，直接对着本章示例 1 的协议写。两条铁律：参数只能塞 URL（EventSource 不支持 body），以及 收到 done 必须 es.close()（否则浏览器会不停自动重连）。注意 useEffect(() => () => esRef.current?.close(), []) 这行——组件卸载时也要断开，否则会泄漏连接。
片段 ③（审批卡片）：把示例 3 的 HTTP 契约变成 UI——start() 拿到 interrupted 就 setPending(j.interrupt) 弹出卡片，用户点按钮调 decide() 发 resume。interrupt.value 里的字段（question、draft）直接绑定到卡片上。
💡 想更快？直接用官方的 Agent Chat UI。 如果你不需要深度定制界面，官方有一个现成的聊天前端仓库 Agent Chat UI，接上你的 LangGraph 后端就能聊（打字机、工具调用展示、HITL 审批都有）——这比自己从零搭快一个数量级。完整清单见下一节。
六、会话与历史回放
示例 5：会话与历史回放
前端最后两个诉求：会话（刷新页面还能拿到这段对话）和回放（像时间轴 / 时间旅行那样看每一步状态怎么变）。
- 会话：其实早就有了——第 4 章的
MemorySaver+thread_id就是。GET /history?thread_id=xxx一拉，整段消息就回来了。 - 回放：靠第 5 章的
getStateHistory(config)。它返回这条 thread 上的所有存档快照（最新在前），每个快照都是一个StateSnapshot，含values（当时的状态）、next（下一步去哪）、config（含checkpoint_id）、metadata（source/step）、createdAt等。
示例 5 把 getStateHistory 的结果整理成前端好用的时间轴结构，通过 GET /history 暴露出去。
完整代码（code/05-session-history-api.ts）
typescript
/**
* 示例 5：为前端提供「会话与历史回放」的接口
*
* 聊天前端常见两个诉求：
* ① 会话：刷新页面后还能拿回这段对话（按 thread_id 拉历史消息）。
* ② 回放：像"时间轴 / 时间旅行"那样看每一步状态怎么变（第 5 章 getStateHistory）。
*
* 本示例用 getStateHistory() 把"每一步的快照"整理成前端好用的结构，通过
* GET /history 暴露出去，并真跑一遍（先聊两轮，再拉历史）。
*
* GET /history?thread_id=xxx
* → { thread_id, total_checkpoints, checkpoints:[{ checkpoint_id, step,
* source, next, message_count, last_message } ...] } （倒序，最新在前）
*/
import { createServer, type IncomingMessage, type ServerResponse } from "node:http";
import { StateGraph, START, END, MessagesAnnotation, MemorySaver } from "@langchain/langgraph";
import { HumanMessage, type BaseMessage } from "@langchain/core/messages";
import { createModel, banner } from "../../lib/model.js";
import { assertChatKey } from "../../lib/env.js";
const PORT = 8792;
async function readJson(req: IncomingMessage): Promise<Record<string, unknown>> {
const chunks: Buffer[] = [];
for await (const chunk of req) chunks.push(chunk as Buffer);
const raw = Buffer.concat(chunks).toString("utf8");
if (!raw) return {};
return JSON.parse(raw) as Record<string, unknown>;
}
function sendJson(res: ServerResponse, status: number, body: unknown) {
const payload = JSON.stringify(body);
res.writeHead(status, { "Content-Type": "application/json; charset=utf-8", "Content-Length": Buffer.byteLength(payload) });
res.end(payload);
}
// 一条消息在前端展示时需要的最小信息
function msgBrief(m: BaseMessage) {
return { type: m.getType(), content: String(m.content).slice(0, 60) };
}
async function main() {
banner("示例 5：会话与历史回放（getStateHistory → GET /history）");
assertChatKey();
const model = createModel({ temperature: 0 });
async function chatNode(state: typeof MessagesAnnotation.State) {
const res = await model.invoke(state.messages);
return { messages: [res] };
}
const graph = new StateGraph(MessagesAnnotation)
.addNode("chat", chatNode)
.addEdge(START, "chat")
.addEdge("chat", END)
.compile({ checkpointer: new MemorySaver() });
const server = createServer(async (req, res) => {
const url = new URL(req.url ?? "/", `http://localhost:${PORT}`);
const method = req.method ?? "GET";
console.log(` [server] ${method} ${url.pathname}`);
try {
if (method === "GET" && url.pathname === "/health") return sendJson(res, 200, { status: "ok" });
// POST /chat：发一句话，返回回复（会话按 thread_id 隔离）
if (method === "POST" && url.pathname === "/chat") {
const body = await readJson(req);
const message = body.message;
const threadId = body.thread_id;
if (typeof message !== "string" || typeof threadId !== "string") {
return sendJson(res, 400, { error: "需要 { message: string, thread_id: string }" });
}
const result = await graph.invoke(
{ messages: [new HumanMessage(message)] },
{ configurable: { thread_id: threadId } }
);
const last = result.messages[result.messages.length - 1] as BaseMessage;
return sendJson(res, 200, {
reply: String(last.content), thread_id: threadId, message_count: result.messages.length,
});
}
// GET /history：回放这段会话的每一步
if (method === "GET" && url.pathname === "/history") {
const threadId = url.searchParams.get("thread_id");
if (!threadId) return sendJson(res, 400, { error: "需要 query 参数 thread_id" });
const config = { configurable: { thread_id: threadId } };
// ★ getStateHistory 返回异步可迭代对象：从最新快照往最早回滚
const checkpoints: Array<Record<string, unknown>> = [];
for await (const snap of graph.getStateHistory(config)) {
const msgs = (snap.values as { messages: BaseMessage[] }).messages ?? [];
const cfg = snap.config as { configurable?: Record<string, unknown> };
checkpoints.push({
checkpoint_id: cfg.configurable?.checkpoint_id,
created_at: snap.createdAt,
source: snap.metadata?.source, // "input"(新输入) / "loop"(图内部推进)
step: snap.metadata?.step, // 框架内部的步数编号
next: snap.next, // 跑完后图准备去哪（[] 表示已结束）
message_count: msgs.length,
last_message: msgs.length ? msgBrief(msgs[msgs.length - 1]) : null,
});
}
return sendJson(res, 200, { thread_id: threadId, total_checkpoints: checkpoints.length, checkpoints });
}
return sendJson(res, 404, { error: `未知路径：${method} ${url.pathname}` });
} catch (e) {
console.error(" [server] 出错：", (e as Error).message);
return sendJson(res, 500, { error: (e as Error).message });
}
});
await new Promise<void>((resolve) => server.listen(PORT, resolve));
const addr = server.address();
const actualPort = typeof addr === "object" && addr ? addr.port : PORT;
const base = `http://127.0.0.1:${actualPort}`;
console.log(`✅ 服务已启动：${base}\n`);
const THREAD = "history-demo";
async function post(path: string, body: unknown): Promise<Record<string, unknown>> {
const r = await fetch(`${base}${path}`, {
method: "POST",
headers: { "Content-Type": "application/json" },
body: JSON.stringify(body),
});
return (await r.json()) as Record<string, unknown>;
}
// 先聊两轮，制造出多个 checkpoint
console.log("🔹 步骤 1：同一会话聊两轮（每次 invoke 都会写入 checkpoint）");
const r1 = await post("/chat", { message: "我叫蜗牛。只回复一句确认。", thread_id: THREAD });
console.log(` 第 1 轮回复：${r1.reply}`);
const r2 = await post("/chat", { message: "我叫什么？一句话回答。", thread_id: THREAD });
console.log(` 第 2 轮回复：${r2.reply}`);
// 再拉历史，模拟前端"时间轴 / 回放"
console.log("\n🔹 步骤 2：GET /history?thread_id=" + THREAD + "（前端拉时间轴数据）");
const resp = await fetch(`${base}/history?thread_id=${THREAD}`);
const hist = (await resp.json()) as {
thread_id: string;
total_checkpoints: number;
checkpoints: Array<Record<string, unknown>>;
};
console.log(` 响应状态：HTTP ${resp.status}`);
console.log(` thread_id = ${hist.thread_id}`);
console.log(` total_checkpoints = ${hist.total_checkpoints}（含开始/中间每一个快照）\n`);
console.log("📽️ 时间轴（最新在前，这正是前端时间旅行 UI 要的顺序）：");
hist.checkpoints.forEach((cp, i) => {
const last = cp.last_message as { type: string; content: string } | null;
const preview = last ? `[${last.type}] ${last.content.slice(0, 32)}` : "（空）";
console.log(
` #${String(i).padStart(2)} step=${String(cp.step).padStart(3)} ` +
`source=${String(cp.source).padEnd(5)} msgs=${String(cp.message_count).padStart(2)} ` +
`next=${JSON.stringify(cp.next).padEnd(10)} → ${preview}`
);
});
console.log("\n📦 其中最早的一条快照（前端点「回到开头」时会用到）完整结构：");
console.log(JSON.stringify(hist.checkpoints[hist.checkpoints.length - 1], null, 2));
await new Promise<void>((resolve, reject) =>
server.close((err) => (err ? reject(err) : resolve()))
);
console.log("\n🛑 服务已关闭，示例结束。");
}
main().catch((e) => {
console.error("❌ 运行失败：", e);
process.exit(1);
});运行：
bash
npx tsx 14-frontend-integration/code/05-session-history-api.ts真实输出：
========================================================================
示例 5：会话与历史回放（getStateHistory → GET /history）
========================================================================
✅ 服务已启动：http://127.0.0.1:8792
🔹 步骤 1：同一会话聊两轮（每次 invoke 都会写入 checkpoint）
[server] POST /chat
第 1 轮回复：收到，蜗牛。
[server] POST /chat
第 2 轮回复：你叫蜗牛。
🔹 步骤 2：GET /history?thread_id=history-demo（前端拉时间轴数据）
[server] GET /history
响应状态：HTTP 200
thread_id = history-demo
total_checkpoints = 6（含开始/中间每一个快照）
📽️ 时间轴（最新在前，这正是前端时间旅行 UI 要的顺序）：
# 0 step= 4 source=loop msgs= 4 next=[] → [ai] 你叫蜗牛。
# 1 step= 3 source=loop msgs= 3 next=["chat"] → [human] 我叫什么？一句话回答。
# 2 step= 2 source=input msgs= 2 next=["__start__"] → [ai] 收到，蜗牛。
# 3 step= 1 source=loop msgs= 2 next=[] → [ai] 收到，蜗牛。
# 4 step= 0 source=loop msgs= 1 next=["chat"] → [human] 我叫蜗牛。只回复一句确认。
# 5 step= -1 source=input msgs= 0 next=["__start__"] → （空）
📦 其中最早的一条快照（前端点「回到开头」时会用到）完整结构：
{
"checkpoint_id": "1f1c4601-95eb-6e20-ffff-110e5be6f0f2",
"created_at": "2026-10-10T04:07:24.674Z",
"source": "input",
"step": -1,
"next": [
"__start__"
],
"message_count": 0,
"last_message": null
}
🛑 服务已关闭，示例结束。这张时间轴怎么变成 UI
| 前端 UI 元素 | 用响应里的哪个字段 |
|---|---|
| 聊天记录（刷新后恢复） | 最新快照的 message_count 对应的 values.messages（示例里简化成了 last_message 预览） |
| 时间轴上的每个点 | checkpoints[]，每个 checkpoint_id 就是一个可点击的节点 |
| 每个点的时间 | created_at |
| "这一步做完、下一步去哪" | next（[] 表示结束；["chat"] 表示还停在 chat 节点） |
| "回到某一步" | 拿该 checkpoint_id 去 graph.updateState(config, values, checkpointId) 或直接 getState 取该快照的 values 渲染 |
💡 注意顺序：
getStateHistory返回的是倒序（#0是最新，#5是最早）。前端做时间轴通常再反转一次，从最早往后排。另外注意输出里source有两种值：input（用户新输入触发的存档）和loop（图内部推进的存档）——一个"用户发了一句"往往对应多个loop存档，因为图内部每走一步都存一次。所以total_checkpoints = 6而消息只有 4 条。
七、官方 UI 生态与选型
前面五节你把「前端要的东西」全部亲手产出来了。但不必什么车轮都自己造——官方给了一整套现成的 UI 生态：
| 名称 | 是什么 | 什么时候用 |
|---|---|---|
| Agent Chat UI | 官方开箱即用的聊天前端（打字机、工具调用展示、HITL 审批都有），配好后端地址就能聊 | 不想自己写界面，要最快出成品 |
| langgraphjs-gen-ui-examples | 生成式 UI（generative UI）官方示例集合：让模型直接"渲染"出界面组件 | 想做大模型驱动的富交互（表单、图表、卡片） |
@langchain/langgraph-sdk | 官方 SDK，提供 React / Vue / Svelte / Angular 各框架的绑定 | 各主流前端框架都有官方支持 |
useStream()（在 SDK 里） | 各框架里最省事的钩子：内部走 SSE，直接给你 messages / values / isLoading / submit / stop | 生产环境的默认选择 |
选型一句话：
要最快出成品 → Agent Chat UI（现成的）
要自定义但要省事 → useStream()（官方 SDK）
要完全掌控 / 非主流环境 → 自己写 SSE 后端 + EventSource（本章示例 1、2、4②）
要有 HITL 审批 → 后端用示例 3 的契约 + 前端审批卡片（示例 4③）
要有会话与回放 → 示例 5 的 getStateHistory 接口💡 无论选哪条路，底层协议都是本书讲的那套：
graph.stream(streamMode)→ SSE 帧 → 前端事件监听。useStream/ Agent Chat UI 只是把这一层包得更好用而已。懂协议，就不怕任何一个框架换代。
八、本章小结（全书收束）
本章速查表
| 你要做的事 | 代码 |
|---|---|
| 起一个 SSE 服务 | res.writeHead(200, { "Content-Type": "text/event-stream" }) → 逐帧 res.write(...) |
| 编一帧 SSE | event: X\n数据\n\n（一个空行结束一帧） |
| 把 token 流接到 SSE | graph.stream(input, { streamMode: "messages" }) → 每个 token 一帧 |
| 前端 fetch 解析 SSE | resp.body.getReader()，按 \n\n 切帧，每帧解析 event: / data: |
| 前端原生接流 | EventSource（只能 GET）+ addEventListener("token"/"done", ...) + 收到 done 要 close() |
| HITL 的 HTTP 契约 | POST /chat 返回 { status:"interrupted", interrupt }；POST /resume 带 { thread_id, resume } |
| 拿到中断内容 | isInterrupted(result) → result.__interrupt__[0].value |
| 恢复执行 | graph.invoke(new Command({ resume }), config) |
| 会话历史 / 回放 | for await (const snap of graph.getStateHistory(config))（倒序，最新在前） |
| 快照字段 | snap.values / snap.next / snap.config / snap.metadata / snap.createdAt |
| 省事前端 | 官方 SDK 的 useStream() / Agent Chat UI |
三句话总结
- 前端要的四件事——实时性、契约、HITL、会话回放——本章各用一个示例真跑了一遍：SSE 后端（1）、事件协议（2）、HTTP 中断（3）、历史回放（5）；React 代码作为可复制片段（4，未实跑）。
- 前后端靠"一份协议"耦合，别靠"默契"：
streamMode的多模式事件、统一包装成{ event, data }的 SSE 帧、写死的 TypeScript 类型——契约越清楚，前端越好写、系统越不怕改。 - 流式对前端是刚需：
invoke()让用户干等整段生成（本次约 1.4s 起，推理模型更久），SSE 让首字延迟降到几百毫秒-1 秒级之间——差的是"系统是否活着"的体感。
动手练习
- 给示例 1 的
POST /chat加一个streamMode: ["messages", "custom"]的双模式流，让节点在回答前先用getWriter()上报一条event: progress（"正在准备…"），前端据此在打字机开始前先显示一行进度提示。 - 改造示例 3：给
interrupt的 payload 加一个expiresAt（几秒后过期），并在POST /resume里检查——过期了就返回 409，模拟"审批超时"。想一想前端拿到 409 应该显示什么。 - 改造示例 5：把
GET /history的返回再反转一次（从最早到最新），并给每个 checkpoint 加上values.messages的完整列表（不只是last_message预览）。这就是一个时间旅行 UI 的数据源。
下一章预告
—— 到这里，《LangGraph.js 中文入门》正文结束。祝你写出越来越聪明的智能体。
结语
从第 1 章"第一个图"，到本章"前端集成"，我们走完了 LangGraph.js 的完整闭环：
状态 & 节点 & 边 → 控制流 → 持久化 → 人工介入 → 流式 → 多智能体 → 工程化 & 前端
（怎么搭图） （怎么分支） （怎么记住） （怎么插人） （怎么实时） （怎么协作） （怎么上线）前端集成是这条链的最后一环，也是很多项目最容易翻车的一环——因为后端"能跑"和用户体验"好"之间，隔着一条 SSE 连接、一份事件协议、一次 close()。
希望这一章帮你把"最后一公里"走稳。懂协议，就不怕框架换代；懂原理，就不怕需求变化。
更多官方资料（JS 版）：
- 流式：docs.langchain.com/oss/javascript/langgraph/streaming
- 前端 / UI：docs.langchain.com/oss/javascript/langgraph/ui