Agent 实现方法论 From Scratch

Web 前端 v04：流式渲染

Web 前端 v04：流式渲染

到 ch15 为止，我们的 Agent 已经把后端武装到牙齿：core 与传输层分离、自定义事件协议（text_delta / tool_call_start / tool_result / approval_request / done / error）、SSE 下发、多会话。但一个 Agent 产品没有前端，就像一台只有 API 的冰箱：它能干活，却没人看得见它在干什么。

本章做v04：Web 前端。不引框架，原生 TS；核心是一个纯函数的 UI 状态机（state machine）：事件流喂进去，视图模型流出来，渲染层只负责把视图模型画到屏幕上。这是「把状态当数据」在界面层的落地，也是本教程「原理优先于封装」在浏览器里的最后一站。

本章目标

读完本章并做完配套练习后，你应该能够：

概念与动机：没有流式前端会怎样

后端已经流式了，前端如果不跟上，用户体验是什么样的？我们演一遍「最朴素的前端」：每个请求发出去，等整个流结束，再把最终文本一次性放进页面：

// 反面教材：不流式的前端
const res = await fetch('/api/chat', { method: 'POST', body: JSON.stringify({ messages }) });
const stream = await readSse(res);      // 把整个 SSE 流读完，攒成 finalText
renderMessage({ role: 'assistant', text: stream.finalText });

这个版本能跑，但用户会遭遇三件糟心事：

1. 等全部完成才显示：TTFB 焦虑。 模型在想、在调工具、在写长答案，页面上什么都没有。用户盯着空白对话区：它是不是卡了？我的网络是不是断了？哪怕 30 秒后答案「啪」地整块出现，用户也已经走了。（ch03 讲过首字节延迟决定「卡不卡」的第一印象。那是传输层的答案，这里要的是界面层的答案：把流式的能力一直延伸到屏幕。）

2. 工具中间态（tool intermediate state）不可见：Agent 变成黑盒。 模型在调 get_weather、在等审批，用户看到的还是空白。它到底在干什么？是不是死循环了？ch14 我们给 CLI 加了 trace 日志让「模型在想什么」可见；Web 前端如果不显示工具调用中间态，等于把 trace 又关掉了。一个不能让你看到「它在做什么」的 Agent，和卡死没有区别。

3. 无法打断：失控感。 模型开始胡说八道、或者问题问错了想重新说，用户只有两个选择：干等它说完，或者关掉页面。真正的前端要有「停止生成」按钮，以及输入排队（input queue）：在它干活的时候你还能打字，它干完这一轮自动接下一轮。

而这三件事的共同解法，不是三个独立的 DOM 函数，而是同一个状态机。为什么？

因为流式界面的本质是事件驱动的状态：同样一段话，它可能处于「正在生成中」「已完成」「被用户打断」三种状态；一个工具调用有「刚发起」「执行中」「成功」「失败」四种状态。这些状态被几十个乱序到达的事件反复改写——如果你把状态藏进 DOM 的各个角落，就没有任何一份代码能回答「现在界面上到底是什么」。唯一靠得住的答案是：一个显式的视图模型 + 一个纯函数的状态转移，DOM 只是它的投影。

这正是本教程反复出现的那条线（ch14 的 onEvent trace 是雏形）：core 发事件、界面表达；到了前端，界面自己的状态也要「事件化」。OpenAI 的 Responses API 流式设计也是同样的思路：服务器把一次运行拆成细粒度事件（response.output_text.delta 携带文本增量、response.function_call_arguments.delta 携带工具参数增量、response.completed 表示整次运行结束，官方流式指南），客户端把它们喂给前端状态；OpenAI Agents SDK 更进一步，把事件分成「原始 token 级事件」与「条目级生命周期事件」（tool_called / tool_output / message_output_item），后者专门用来驱动 UI 展示「正在调用工具」这类进度（Agents SDK · Streaming）。我们 ch15 的协议走的是同一条路：事件是唯一真相，UI 是事件的函数。

事件流：协议在 UI 视角长什么样

ch15 定了事件协议，前端直接消费。七个 wire 事件 + 两个本地事件（UI 自己产生，喂给同一个 reducer）：

事件（data.type）载荷前端含义
text_deltadelta打字机增量：追加到当前 assistant 消息
tool_call_startcallId, name, args工具开始执行：插入 running 工具块
tool_resultcallId, ok, result?, error?工具结束：把匹配块置为 done / failed
approval_requestrequestId, toolName, args, risk需要审批：进入 waiting_approval
approval_resultrequestId, approved审批回答：匹配才放行
doneusage?回合结束：收尾、记 token 账单、排空队列
errormessage出错：保留部分文本、标记 interrupted
user_input（本地）text空闲开新回合；忙碌入队
interrupt（本地）—用户停止：冻结当前消息、丢弃旧流

传输层我们在 ch03 已经手写过了（fetch + ReadableStream 逐 chunk 解析 + 半行 buffer），这里不重复协议原理，只消费它的产物：每一帧 data: 行是一个 JSON 对象，解析出来就是一个 AppEvent，丢给 reducer。浏览器原生 EventSource 只支持 GET 且不能带自定义头（MDN: Server-sent events），LLM 场景依然用 ch03 的手写客户端；中断用 AbortController（MDN: AbortController），流的读取用 ReadableStream（MDN: ReadableStream）。

还有一个顺序保证要知道：SSE 是顺序的，服务器按序推、客户端按序收。这给了前端一个重要的底气：只要单个连接的顺序不乱，reducer 就可以放心地把事件按到达顺序处理。乱序风险来自别处（见「常见坑」）：并发多请求、重连、打断后的事件交错。

UI 状态机设计：状态、转移、视图模型

六状态的状态机

把整章前面散落的讨论收敛成一张图。状态机六态，事件是唯一的转移输入：

stateDiagram-v2
    [*] --> idle
    idle --> streaming: user_input（开新回合）
    done --> streaming: user_input / drainQueue（队列非空）
    error --> streaming: user_input
    streaming --> streaming: text_delta / tool_call_start / tool_result
    streaming --> waiting_approval: approval_request
    waiting_approval --> streaming: approval_result（requestId 匹配）
    streaming --> interrupted: interrupt
    waiting_approval --> interrupted: interrupt
    interrupted --> done: done（旧流关闭）
    interrupted --> error: error
    streaming --> done: done
    waiting_approval --> done: done
    streaming --> error: error
    waiting_approval --> error: error
    idle --> error: error
    done --> error: error
    done --> [*]
    error --> [*]

读图要点：

视图模型：渲染层唯一读取的东西

状态机产出**视图模型（view model）**：一份普通的数据，不是 DOM。

ViewModel {
  sessionId: string;                          // ch15 多会话：一个会话一份 ViewModel
  status: 'idle' | 'streaming' | 'waiting_approval' | 'interrupted' | 'done' | 'error';
  messages: MessageView[];                    // 用户与 assistant 消息，按到达顺序
  pendingApproval: ApprovalView | null;       // 审批提示（requestId/toolName/args/risk）
  usage: Usage | null;                        // done 的 token 账单
  error: string | null;
  inputQueue: QueueItem[];                    // 忙碌时到达的用户输入
}
MessageView { id, role, blocks: Block[], status: 'streaming' | 'complete' | 'interrupted' }
Block = TextBlock { kind: 'text', text } | ToolBlock { kind: 'tool', callId, name, args, status: 'running' | 'done' | 'failed', result?, error? }

三个设计决策值得停一下：

决策一：消息内容是「块的保序数组」，文本与工具按到达顺序混排。 真实对话里文本和工具调用是交错的：模型说一句、调一个工具、又说一句。如果把文本和工具分开存，渲染时顺序就丢了。块数组让 text_delta 追加到最后一个文本块、tool_call_start 在消息尾部插一个工具块，到达顺序即渲染顺序。messageText(message) 只拼接文本块，供增量渲染（打字机效果）与「消息摘要」使用。

决策二：thinking 不是新事件，是派生态。 我们的协议里没有 thinking 事件，「正在思考」由状态机派生：status === 'streaming' 且当前 assistant 消息还没有任何块时，UI 显示「(thinking…)」。如果未来协议加入 reasoning_delta，同一个块机制直接接住，状态机形状不变。

决策三：纯函数。 reducer 不碰 DOM、不碰网络、不碰任何全局。同样的 (state, event) 序列永远得到同样的结果。这带来三个免费能力：可测（判题测试注入脚本事件流、断言视图模型，无需浏览器）、可回放（线上出 bug，把事件日志重放一遍即可复现）、可恢复（断线重连后，把持久化的事件重放回来，界面就回来了，这是 ch16 断线恢复在前端的接口）。

事件 → 渲染流水线

一个事件从网络到屏幕的完整路径：

flowchart LR
    SSE["SSE 事件流（ch15 服务）<br/>data: 一行一个 JSON"] --> PARSE["SSE 解析（ch03 手写客户端）<br/>JSON.parse → AppEvent"]
    UI["用户操作：输入 / 停止生成"] --> LOCAL["本地事件<br/>user_input / interrupt"]
    PARSE --> REDUCER["reduceViewModel(state, event)<br/>纯函数 · 状态守卫 · 不可变更新"]
    LOCAL --> REDUCER
    REDUCER --> VM["视图模型 ViewModel<br/>（唯一真相）"]
    VM --> RENDER["渲染：renderFrame / DOM patch<br/>打字机文本 · 工具块 · 审批提示 · 光标"]
    RENDER --> DOM["浏览器 UI"]
    DOM -. 操作 .-> UI

注意这条线是单向的：事件只能从左边进来，视图模型只能被 reducer 修改，渲染只读视图模型。任何「组件自己改自己」的旁路都会把状态重新散落回 DOM，那正是我们要消灭的东西。

手写实现：reducer → 视图模型 → 渲染帧

一、reducer 骨架

一个 switch，九个事件，每个分支一个处理函数。done 分支是个组合：先收尾（onDone），再排空队列（drainQueue）：

export function reduceViewModel(state, event) {
  switch (event.type) {
    case 'user_input':     return onUserInput(state, event.text);
    case 'text_delta':     return onTextDelta(state, event.delta);
    case 'tool_call_start': return onToolCallStart(state, event);
    case 'tool_result':    return onToolResult(state, event);
    case 'approval_request': return onApprovalRequest(state, event);
    case 'approval_result': return onApprovalResult(state, event);
    case 'done':           return drainQueue(onDone(state, event));
    case 'error':          return onError(state, event);
    case 'interrupt':      return onInterrupt(state);
    default:               return state;
  }
}

每个处理函数的第一个动作都是**状态守卫（state guard）**：if (state.status !== 'streaming') return state;，不合法的事件原样返回（丢弃）。这个「只认状态、不认来源」的规则，就是打断残留、重复事件、乱序事件的最后防线。

二、开回合与输入排队

user_input 是唯一能创建消息的事件。空闲（idle/done/error）时开新回合：追加用户消息 + 一条空的 assistant 消息，状态机进入 streaming；忙碌时入队：

function onUserInput(state, text) {
  if (isBusy(state.status)) {           // streaming / waiting_approval / interrupted
    const item = { id: `q${state.counter + 1}`, text };
    return { ...state, counter: state.counter + 1, inputQueue: [...state.inputQueue, item] };
  }
  return openTurn(state, text);
}

function openTurn(state, userText) {
  const userMessage = { id: `m${state.counter + 1}`, role: 'user',
    blocks: [{ kind: 'text', text: userText }], status: 'complete' };
  const assistantMessage = { id: `m${state.counter + 2}`, role: 'assistant',
    blocks: [], status: 'streaming' };
  return { ...state, counter: state.counter + 2, status: 'streaming',
    pendingApproval: null, usage: null, error: null,
    messages: [...state.messages, userMessage, assistantMessage] };
}

counter 是消息 id 的铸币机（m1/m2/q3…），保证 id 全局唯一；openTurn 顺带清掉上一回合的 pendingApproval/usage/error，绝不让残留污染新回合。排队的语义是「一件事做完再做下一件」：done 到达时 drainQueue 取队头、开下一回合：

function drainQueue(state) {
  if (state.inputQueue.length === 0) return state;
  const [head, ...rest] = state.inputQueue;
  return openTurn({ ...state, inputQueue: rest }, head.text);
}

三、文本累积与工具中间态

text_delta 追加到当前 assistant 消息的最后一个文本块。如果最后一块是工具块（文本在工具之后到来），就新建文本块：

function onTextDelta(state, delta) {
  if (state.status !== 'streaming' || delta === '') return state;
  const messages = appendTextBlock(state.messages, delta);
  if (messages === state.messages) return state;
  return { ...state, messages };
}

function appendTextBlock(messages, delta) {
  const last = messages[messages.length - 1];
  if (!last || last.role !== 'assistant') return messages;
  const blocks = last.blocks;
  const tail = blocks[blocks.length - 1];
  const newBlocks =
    tail && tail.kind === 'text'
      ? [...blocks.slice(0, -1), { kind: 'text', text: tail.text + delta }]
      : [...blocks, { kind: 'text', text: delta }];
  return [...messages.slice(0, -1), { ...last, blocks: newBlocks }];
}

工具调用的两个事件把「中间态 → 终态」演完整。tool_call_start 插入 running 块，UI 立刻显示「正在调用 get_weather，参数是…」，用户知道它在干活；tool_result 只解析第一个匹配的 running 块，ok 为真置 done + result，为假置 failed + error（错误也是 UI 数据，ch08 的 error-as-data 一路延续到浏览器）：

function onToolCallStart(state, event) {
  if (state.status !== 'streaming') return state;
  const messages = appendToolBlock(state.messages, {
    kind: 'tool', callId: event.callId, name: event.name,
    args: event.args, status: 'running',
  });
  if (messages === state.messages) return state;
  return { ...state, messages };
}

function onToolResult(state, event) {
  if (state.status !== 'streaming') return state;
  const messages = resolveToolBlock(state.messages, event);   // 匹配 callId 且仍 running 的第一个块
  if (messages === state.messages) return state;
  return { ...state, messages };
}

四、审批门、收尾与错误

approval_request 把状态机钉进 waiting_approval，并把审批提示存进视图模型（UI 渲染成「需要你确认」的卡片）；approval_result 只有 requestId 匹配才放行：

function onApprovalRequest(state, event) {
  if (state.status !== 'streaming') return state;
  return { ...state, status: 'waiting_approval',
    pendingApproval: { requestId: event.requestId, toolName: event.toolName,
      args: event.args, risk: event.risk } };
}

function onApprovalResult(state, event) {
  if (state.status !== 'waiting_approval') return state;
  if (!state.pendingApproval || state.pendingApproval.requestId !== event.requestId) return state;
  return { ...state, status: 'streaming', pendingApproval: null };
}

done 收尾：把当前 assistant 消息（若还在 streaming）标记为 complete，状态机进 done，记下 token 账单。error 是「温柔的错误」：保留已累积的部分文本，把消息标记为 interrupted，让用户看到「它说到一半出错了」，而不是整段消失：

function onDone(state, event) {
  if (state.status !== 'streaming' && state.status !== 'waiting_approval'
      && state.status !== 'interrupted') return state;   // 重复的 done：忽略
  const messages = state.messages.map((m, i) =>
    i === state.messages.length - 1 && m.role === 'assistant' && m.status === 'streaming'
      ? { ...m, status: 'complete' } : m);
  return { ...state, messages, status: 'done',
    usage: event.usage ?? null, pendingApproval: null };
}

五、打断：丢弃旧流

打断是本章最微妙的一格。用户点「停止生成」，前端做两件事：

  1. 传输层：AbortController.abort() 断开当前 SSE 连接（ch03 的「主动中止」），服务器不再推新事件；
  2. 状态层：reducer 收到本地事件 interrupt，冻结当前消息（保留部分文本），进入 interrupted：
function onInterrupt(state) {
  if (state.status !== 'streaming' && state.status !== 'waiting_approval') return state;
  const messages = state.messages.map((m, i) =>
    i === state.messages.length - 1 && m.role === 'assistant'
      ? { ...m, status: 'interrupted' } : m);
  return { ...state, messages, status: 'interrupted', pendingApproval: null };
}

但有一个现实问题：连接断了，事件可能还在路上。已经发出、已进 TCP 缓冲、已经解析出来的 text_delta 还会陆续到达，它们属于「被废弃的那条流」。怎么办？答案是复用前面的状态守卫（state guard）：interrupt 之后状态机是 interrupted，而 onTextDelta/onToolCallStart/onToolResult/onApprovalRequest/onApprovalResult 全都只认 streaming/waiting_approval，残留事件自然全部被丢弃。只有旧流的 done/error 被 onDone/onError 接受（它们认得 interrupted），用来关闭这个回合并触发 drainQueue 开启排队中的下一回合。传输层 abort + 状态层守卫，双保险，一个管「不再来」，一个管「来了也不收」。

六、渲染：视图模型 → 帧

reducer 讲完了，渲染其实是最「无聊」的部分，这正是我们想要的效果：渲染不承担任何状态逻辑，只把视图模型画出来。浏览器版是 DOM patch（打字机文本节点、工具块、审批卡片、光标动画）；演示与判题版把它渲染成文本帧，零依赖可打印：

export function renderFrame(viewModel) {
  const lines = [`[${viewModel.status}] session=${viewModel.sessionId}`];
  lines.push('-'.repeat(56));
  for (const message of viewModel.messages) {
    lines.push(...renderMessage(message));   // [you] / [assistant·streaming] …
  }
  if (viewModel.pendingApproval) {
    lines.push(`approval: ${viewModel.pendingApproval.toolName} [risk=${viewModel.pendingApproval.risk}] — waiting for you`);
  }
  if (viewModel.error) lines.push(`error: ${viewModel.error}`);
  if (viewModel.inputQueue.length > 0) {
    lines.push(`queue: ${viewModel.inputQueue.map((item) => item.text).join(' | ')}`);
  }
  if (viewModel.usage) lines.push(`usage: in=${viewModel.usage.inputTokens} out=${viewModel.usage.outputTokens} tokens`);
  return lines.join('\n');
}

Markdown 是极简实现：围栏代码块（以行首三个反引号开头与结尾的代码段）渲染成一个带语言标签的盒子，行内代码（`x`）转成 [x]，其余原样输出。两个细节值得说：

一轮对话在文本帧里长什么样

把脚本化事件流喂进状态机，逐事件打印帧，一次「查天气 + 查时间」的对话（文本与两个工具调用交错）渲染出来是：

--- frame 1 — event: user_input ---
[streaming] session=demo-17
--------------------------------------------------------
[you] 帮我看一下上海的天气，顺便告诉我当前的 UTC 时间
[assistant·streaming] (thinking…)
--------------------------------------------------------

--- frame 4 — event: tool_call_start ---
[streaming] session=demo-17
--------------------------------------------------------
[you] 帮我看一下上海的天气，顺便告诉我当前的 UTC 时间
[assistant·streaming] 我先查一下上海的天气。请稍等。
[assistant·streaming] tool get_weather {"location":"Shanghai"} (running)
--------------------------------------------------------

--- frame 5 — event: tool_result ---
[streaming] session=demo-17
--------------------------------------------------------
[you] 帮我看一下上海的天气，顺便告诉我当前的 UTC 时间
[assistant·streaming] 我先查一下上海的天气。请稍等。
[assistant·streaming] tool get_weather {"location":"Shanghai"} -> 上海：晴，27°C，微风，紫外线中等
--------------------------------------------------------

--- frame 14 — event: done ---
[done] session=demo-17
--------------------------------------------------------
[you] 帮我看一下上海的天气，顺便告诉我当前的 UTC 时间
[assistant] 我先查一下上海的天气。请稍等。
[assistant] tool get_weather {"location":"Shanghai"} -> 上海：晴，27°C，微风，紫外线中等
[assistant] 上海今天晴，27°C，微风。接下来查一下当前 UTC 时间。
[assistant] tool get_time {} -> 2026-08-11T12:39:34.454Z
[assistant] 当前时间是 2026-08-11T12:39:34Z。
--------------------------------------------------------
usage: in=218 out=152 tokens

观察三个细节：frame 1 的 (thinking…) 是派生状态（streaming + 空消息）；frame 4 的 (running) 是工具中间态，frame 5 立刻被结果填上；frame 14 的 done 把消息全部标记为 complete（标签从 assistant·streaming 变 assistant）、状态机进 done、打印 token 账单。这就是打字机效果 + 工具可见性的全部秘密——没有任何魔法，只是一张状态表被逐事件驱动。

审批场景（删除文件前需要确认）与打断场景（输入排队、旧流残留被丢弃）的帧同样直观：

--- frame 5 — event: approval_request ---
[waiting_approval] session=demo-17
--------------------------------------------------------
[you] 把 /tmp/stale.log 删掉
[assistant·streaming] 删除文件是高风险操作，需要你确认。
[assistant·streaming] tool rm_file {"path":"/tmp/stale.log"} (running)
--------------------------------------------------------
approval: rm_file {"path":"/tmp/stale.log"} [risk=approve] (request req_1) — waiting for you

--- frame 7 — event: done ---
[streaming] session=demo-17
--------------------------------------------------------
[you] 1 + 1 等于多少？
[assistant·interrupted] 好的，我来算一下。
[you] 那 2 + 2 呢？
[assistant·streaming] (thinking…)
--------------------------------------------------------

常见坑与失败模式

坑一：没有视图模型，事件直接改 DOM。 每个 text_delta 回调里 el.textContent += delta、tool_call_start 回调里 el.appendChild(toolCard)，状态散落在各处，没有任何代码能回答「现在界面上是什么」。断线重连（ch16）时无从恢复、测试无从下手、interrupt 时无法冻结。判断标准：把所有 DOM 操作删掉，界面状态还能被完整描述吗？ 能，说明有视图模型；不能，就是坑一。

坑二：reducer 不纯。 在 reducer 里 console.log、读 Date.now()、直接改传入的 state，状态机从此不可重放：同样的事件日志两次跑出不同结果，回放调试与判题全废。纯函数的纪律和 ch14 的 core 一模一样：输入即全部，输出即唯一，不碰全局。

坑三：乱序。 单条 SSE 连接是顺序的，但多个来源会乱：并发多个请求（并行工具各自回流）、断线重连后新旧流交错、打断后残留与新一轮事件混在一起。解法是状态守卫 + 连接隔离：传输层保证「一条连接只属于一个回合」，reducer 用状态守卫拒绝不属于当前状态的事件。协议里 approval_result 带 requestId 匹配、tool_result 带 callId 匹配，都是同一招：凡是可能乱的事件，都要带身份。

坑四：重复事件。 重连会重放，网络层可能重试，服务器可能重复推送。同一个 done 到达两次，消息被收两次尾；同一个 tool_result 到达两次，running 块被解析两次。reducer 的幂等规则：done 只在 streaming/waiting_approval/interrupted 生效（第二次 done 状态已是 done，被守卫拒绝）；tool_result 只解析第一个匹配且仍为 running 的块（第二次匹配到的块已是 done，条件不满足，原样返回）。错误处理同理：已经是 error 的状态再收到 error，忽略。

坑五：打断残留。 只做了传输层 abort，没做状态守卫。abort 后已经在缓冲里的 text_delta 到达时，reducer 状态还是 streaming，会把废弃流的文本拼进新回合，造成「答案里混着上一轮的残渣」。传输层 abort + 状态层 interrupted 守卫，缺一不可；本章的 onInterrupt 就是后者的落点。

坑六：把 Markdown 渲染放进增量路径。 在 onTextDelta 里对每个增量单独做 renderMarkdown。代码块的开头围栏和结尾围栏落在不同增量里，增量渲染永远只见一半；text_delta 还可能把多字节字符或换行切在中间。正确做法：增量只负责「拼」，渲染只负责「读」。渲染永远发生在完整文本上（这正是视图模型存在的意义之一）。

坑七：状态机与后端实际状态脱节。 前端猜「它在等审批」「它应该完了」，而不是从事件里读。后端加了新中间态（比如 ch18 会有更复杂的审批流程），前端还在用旧的猜测渲染，界面与真实状态漂移。事件是唯一真相：视图模型里没有任何一个字段是前端猜出来的，全部由事件写入。

小结

下一章（ch18）我们进入 v05：权限与 Human-in-the-Loop：审批从「状态机的一格」变成完整的产品能力：工具风险分级、auto/suggest/approve 三模式、审批双向流（前端确认回传）、操作审计落库。本章的 waiting_approval 状态和 pendingApproval 视图模型，就是它在前端的接缝。先把本章练习做完，亲手把状态机的三个 stage（reducer 基础 → 中间态/完成态 → 队列与打断）写出来。

延伸阅读

完成阅读，去做练习 →