Skip to content
基于 LLM 的 Web 前端流式交互工程化
概述
在传统的请求-响应模式中，前端向 LLM 服务发起请求后，必须等待模型把整段回答生成完毕，HTTP 响应才会返回。模型的生成速度通常与输出长度成正比，一段数百字的回答可能需要数秒甚至更久。在这段时间里，用户只能面对一个 loading 状态，无法判断请求是否正常、模型是否已经开始工作。
这种模式还存在一个实际约束：HTTP 请求在等待期间可能被网关、代理或浏览器中断，一旦连接断开，用户必须重新提交整个请求。对于需要长文本生成的场景，这个问题尤为突出。
流式接口改变了上述交互方式。模型每生成一小段内容，就立即推送到客户端。用户看到的不是等待一整段回答，而是文字逐字出现。首字到达的时间被大幅提前，整个交互过程的“空白期”被压缩到模型开始生成之前的延迟内。
基本概念
Token 与 Chunk：流式输出的基本单位
LLM 的生成过程以 token 为基本单位。token 是模型内部使用的文本片段，一个 token 不一定等于一个汉字或一个英文单词，可能是半个词、一个标点或一个常用组合。模型每生成一个 token，就需要一次前向计算。API 服务不会为每个 token 单独发送一次网络请求，而是把若干 token 打包成一个 chunk 传输。
从浏览器视角看，chunk 是网络层的数据片段。一次 fetch 响应体可能被 TCP 分包成多个 chunk，也可能一个请求返回多个业务事件。前端需要区分三层单位：
- 传输层：字节流，由
ReadableStream按块读取。 - 协议层：SSE 事件，以
data:行为单位。 - 业务层：模型输出片段，即 OpenAI 接口中的
delta。
UI 的渲染单位通常是业务层的 delta。一个 delta 可能只包含一个 token，也可能包含一段完整句子。因此，前端不能用“网络包数量”来驱动 UI，而应该解析出业务内容后再决定如何渲染。
SSE、fetch ReadableStream 与 WebSocket 的差异
三种常见接入方式的差异如下。
| 方式 | 传输方向 | 连接方式 | 协议 | 适用场景 |
|---|---|---|---|---|
| EventSource（SSE） | 服务器单向推送 | HTTP | text/event-stream | 以 GET 请求订阅文本事件流 |
| fetch + ReadableStream | 双向请求/响应 | HTTP | 任意 Content-Type | 以 POST 提交消息，同时读取流式响应 |
| WebSocket | 全双工 | 独立连接 | WebSocket 帧 | 需要双向持续通信，或由服务端主动推送事件 |
EventSource 是最早为服务器推送设计的 API。它内置断线重连和事件 ID 机制，但只能使用 GET 请求，不适合把用户消息放在请求体中传给 LLM 服务。因此，目前大多数 LLM 前端采用第二种方式：用 fetch 发起 POST 请求，再通过 response.body 读取流式响应。这种方式不需要额外协议，能够携带数据，也能在流未结束时被 AbortController 中断。
WebSocket 适用于更复杂的交互场景，例如需要服务端多次主动推送、需要客户端和服务端双向发送消息。代价是连接管理、心跳维持和消息帧解析都需要自行处理。在典型 LLM 聊天应用中，fetch 流式读取已经足够。
基本用法
使用 fetch 读取 ReadableStream
fetch 返回的 Response 对象，其 body 属性是一个 ReadableStream。浏览器在收到响应头后就会开始传递数据，response.body 不需要等到完整响应结束。下面是最基础的读取方式：
js
const response = await fetch('/api/chat', {
method: 'POST',
headers: { 'Content-Type': 'application/json' },
body: JSON.stringify({
messages: [{ role: 'user', content: '你好' }],
stream: true
})
});
const reader = response.body.getReader();
const decoder = new TextDecoder();
while (true) {
const { done, value } = await reader.read();
if (done) break;
const chunkText = decoder.decode(value, { stream: true });
console.log(chunkText);
}reader.read() 返回一个 Promise，解析后得到一个包含 done 和 value 两个属性的对象。value 是一个 Uint8Array。因为字节流可能把 UTF-8 字符截断，所以必须使用 TextDecoder 并启用 stream: true，让解码器保留跨 chunk 的多字节序列。
响应完成后，done 变为 true，循环退出；如果中途需要停止，可以调用 reader.cancel()。
EventSource 与 WebSocket 的接入方式
EventSource 适合已经被后端封装为 SSE 的 GET 接口：
js
const source = new EventSource('/api/chat-stream');
source.onopen = () => {
console.log('连接已建立');
};
source.onmessage = (event) => {
console.log(event.data);
};
source.onerror = () => {
console.log('连接异常或断开');
};EventSource 会把每个 SSE 事件解析为 message 事件，event.data 是字符串。它自动处理了连接断开后的重连，但重连行为由浏览器内部决定，前端无法自定义请求体。若 LLM 服务只提供 POST 接口，EventSource 就不适用。
WebSocket 的接入方式与之不同：
js
const socket = new WebSocket('wss://example.com/chat');
socket.onopen = () => {
socket.send(JSON.stringify({ messages: [{ role: 'user', content: '你好' }] }));
};
socket.onmessage = (event) => {
const data = JSON.parse(event.data);
console.log(data);
};
socket.onclose = () => {
console.log('连接已关闭');
};WebSocket 没有 SSE 那样的事件格式约束，消息可以按任意自定义结构发送。前端需要注意分帧、重连、心跳和指数退避，这些都超出了简单流式读取的范围。多数 AI 应用的 REST 流式接口用 fetch 就能覆盖，因此下面的章节默认采用 fetch 方式。
for await...of 与异步迭代
ReadableStream 实现了异步可迭代协议，浏览器允许直接使用 for await...of 消费响应体中的到达块[1]：
js
let total = 0;
for await (const chunk of response.body) {
total += chunk.length;
}循环会持续消费，直到数据耗尽或流终止。这个写法比手动调用 reader.read() 更简洁，但有两个行为需要了解[1]：
- 迭代期间流被锁定，其他消费者不能获得
reader。 - 循环退出后锁释放，但默认会取消流。若想退出后继续使用流，可以调用
stream.values({ preventCancel: true })。
在普通场景中，for await...of 足够使用。需要精细控制取消或提前退出时机时，使用 getReader() 更灵活。
工作原理
chunk、token 与事件流的边界
ReadableStream 读取到的 chunk 是网络字节块，它可能与业务事件没有边界对应关系。一个 SSE 事件可能拆在两个网络 chunk 中，多个 SSE 事件也可能合并在一个网络 chunk 里。因此，前端需要自己实现事件帧解析。
SSE 事件的格式足够简单：每行以字段名开头，data: 后是数据，事件之间以空行分隔。一个最小解析器如下：
js
function parseSSE(chunkText, buffer = '') {
const events = [];
let current = buffer + chunkText;
while (current.includes('\n\n')) {
const index = current.indexOf('\n\n');
const rawEvent = current.slice(0, index);
current = current.slice(index + 2);
const dataLines = rawEvent
.split('\n')
.filter((line) => line.startsWith('data:'))
.map((line) => line.slice(5).trim());
if (dataLines.length > 0) {
events.push(dataLines.join('\n'));
}
}
buffer = current;
return { events, buffer };
}这个解析器把未遇到空行的数据保留在 buffer 中，等待下一个 chunk 到达时继续拼接。对于 LLM 场景，data: 行内的内容通常是一个 JSON 字符串，需要二次转换。
Partial JSON 与增量数据解析
LLM 接口中，文本内容 delta.content 是普通字符串，直接拼接即可。但函数调用（Function Calling）和工具调用（Tool Call）的参数是一个 JSON 对象，模型会把 JSON 拆成多个 token 逐步返回。例如：
data: {"choices":[{"delta":{"tool_calls":[{"index":0,"function":{"arguments":"{\"city\":\"杭"}}]}}]}
data: {"choices":[{"delta":{"tool_calls":[{"index":0,"function":{"arguments":"州\"}"}}]}}]}这两个事件单独看都不是合法的 JSON。前端在解析增量参数时，应当把片段累积到缓冲区，再尝试整体解析：
js
let argumentBuffer = '';
function appendToolCallDelta(delta) {
const toolCall = delta.tool_calls?.[0];
if (!toolCall) return null;
argumentBuffer += toolCall.function.arguments;
try {
const args = JSON.parse(argumentBuffer);
return args;
} catch {
return null; // 数据不完整，等待下一个 chunk
}
}当 JSON.parse 成功时，说明参数已经完整。若后续仍有内容，通常意味着模型流式输出的是数组或对象中的多个部分组成，需要根据具体接口规范处理。
流式状态机：loading、streaming、error、done
前端需要用一个状态机管理整个流式请求的生命周期。基础状态包括：
js
const StreamStatus = {
IDLE: 'idle', // 尚未发起请求
LOADING: 'loading', // 请求已发出，等待第一个 chunk
STREAMING: 'streaming', // 正在接收增量内容
ERROR: 'error', // 请求失败
DONE: 'done' // 正常结束
};状态转换规则：
IDLE → LOADING：用户触发生成。LOADING → STREAMING：收到第一个业务数据块。STREAMING → DONE：收到结束标记。LOADING/STREAMING → ERROR：网络中断、HTTP 错误或协议错误。STREAMING → IDLE：用户手动停止生成。
代码中可以用 reducer 管理这些状态：
js
function streamReducer(state, action) {
switch (action.type) {
case 'START':
return { status: StreamStatus.LOADING, text: '', error: null };
case 'DELTA':
return {
status: StreamStatus.STREAMING,
text: state.text + action.content,
error: null
};
case 'ERROR':
return { status: StreamStatus.ERROR, text: state.text, error: action.error };
case 'DONE':
return { status: StreamStatus.DONE, text: state.text, error: null };
case 'STOP':
return { status: StreamStatus.IDLE, text: state.text, error: null };
default:
return state;
}
}DONE 与 STOP 语义不同：DONE 是服务端正常结束，STOP 是前端主动取消。两者都应当保留已生成的内容，但后续 UI 行为不同。
取消、超时、重连与竞态处理
AbortController 是取消 fetch 的标准方式：
js
const controller = new AbortController();
async function startChat() {
try {
const response = await fetch('/api/chat', {
method: 'POST',
body: JSON.stringify({ messages }),
signal: controller.signal
});
} catch (error) {
if (error.name === 'AbortError') {
console.log('请求已取消');
} else {
console.error('请求失败：', error);
}
}
}
function stopChat() {
controller.abort();
}controller.abort() 会立刻取消请求，fetch 抛出的 AbortError 可以用于区分“用户主动停止”和“网络异常”。
超时与取消可以用同一个 AbortController 实现：
js
const timeoutMs = 30000;
const timeoutId = setTimeout(() => controller.abort(), timeoutMs);
try {
const response = await fetch(url, { signal: controller.signal });
// 成功后清除定时器
} finally {
clearTimeout(timeoutId);
}竞态条件在连续发送请求时容易出现。例如用户点击“重新生成”后，前一个请求仍然在传输。解决思路是为每次请求生成一个递增 ID，在状态更新前校验 ID 是否为最新：
js
let latestRequestId = 0;
async function startStream() {
const requestId = ++latestRequestId;
const response = await fetch('/api/chat', {
method: 'POST',
body: JSON.stringify({ messages })
});
const reader = response.body.getReader();
while (true) {
const { done, value } = await reader.read();
if (done) break;
if (requestId !== latestRequestId) {
reader.cancel();
return;
}
// 更新 UI
}
}
function stopStream() {
latestRequestId++;
}这种做法的优点是即使没有调用 abort()，旧的请求回调也不会污染最新状态。
Streaming UI 渲染模式
逐字输出与游标反馈
流式 UI 最直接的呈现方式是逐字输出。每收到一个 delta，把新内容追加到已有文本。为了给用户一个明确的“正在输出”信号，可以在文本末尾添加一个游标符号：
jsx
function TypewriterText({ text }) {
return (
<span>
{text}
<span className="stream-cursor" aria-hidden="true">▍</span>
</span>
);
}游标使用 aria-hidden="true" 标记，避免被屏幕阅读器当成内容朗读。CSS 中可以通过 @keyframes 设置闪烁动画：
css
.stream-cursor {
display: inline-block;
width: 0.6em;
height: 1em;
background: currentColor;
animation: blink 1s steps(1) infinite;
}
@keyframes blink {
50% { opacity: 0; }
}对于开启 prefers-reduced-motion 的用户，应当关闭闪烁动画：
css
@media (prefers-reduced-motion: reduce) {
.stream-cursor {
animation: none;
}
}骨架屏与首屏反馈
模型生成前存在一段等待时间，通常是从用户提交请求到收到第一个 token。这个阶段 UI 如果只有一个静态 loading 文案，用户的等待感会很明显。骨架屏可以给用户一个内容即将出现的预期。
一个简单的做法是渲染一个与回答区域结构一致的灰色占位块：
jsx
function ChatMessage({ text, status }) {
return (
<div className="message">
{status === 'loading' && <Skeleton />}
{status === 'streaming' && <TypewriterText text={text} />}
{status === 'done' && <div>{text}</div>}
</div>
);
}
function Skeleton() {
return (
<div className="skeleton" aria-hidden="true">
<div className="skeleton-line" />
<div className="skeleton-line" />
<div className="skeleton-line" />
</div>
);
}骨架屏与流式状态机配合：LOADING 状态显示骨架，STREAMING 状态切换到逐字输出。骨架屏只覆盖等待阶段，不应在流式输出过程中持续显示。
渲染调度：防抖、批处理与速度控制
LLM 接口返回 chunk 的频率通常远高于浏览器 UI 渲染频率。如果一个流式请求每秒返回 60 个 chunk，而 React 需要同步更新状态，就可能造成不必要的重渲染。常见的做法是用缓冲区聚合 chunk，定时批量更新 UI。
js
let buffer = '';
let flushTimer = null;
function onChunkReceived(text) {
buffer += text;
if (!flushTimer) {
flushTimer = setTimeout(() => {
setResponseText((prev) => prev + buffer);
buffer = '';
flushTimer = null;
}, 50);
}
}这个例子里，50ms 是一个示例值。实际上，可以根据内容的“打字速度”预期选择刷新间隔。间隔越短，渲染越平滑；间隔越长，性能越好。关键是让 UI 更新频率与人类阅读速度匹配，而不是与网络事件频率匹配。
如果需要在 60fps 的动画循环中实时更新文本，可以用 requestAnimationFrame 调度：
js
let rafId = null;
let fullText = '';
function scheduleRender() {
if (rafId !== null) return;
rafId = requestAnimationFrame(() => {
setDisplayText(fullText);
rafId = null;
});
}
function appendDelta(deltaText) {
fullText += deltaText;
scheduleRender();
}requestAnimationFrame 会等待浏览器进入下一次绘制前执行回调，这样可以把累积的多个 delta 合并成一次状态更新，避免在每帧之间重复渲染。
Markdown 与代码块增量呈现
LLM 经常会输出 Markdown，包括标题、列表、代码块等结构。如果前端把每个 delta 直接追加到文本节点，那么用户的屏幕会先看到 #、# 标、# 标题 这样的中间过程，Markdown 语法符号会闪烁出现。把未完成的 Markdown 文本交给渲染器解析，还会产生两个问题：
- 性能：每次新增 token 后重新解析整篇 Markdown，开销会随文本长度线性增长。
- 体验：未匹配的代码围栏会瞬间渲染成一整段代码块，造成布局抖动。
一种常用的策略是“边显示边缓冲”。在流式输出期间，把内容作为纯文本展示，同时对代码块进行简单检测；待流结束后，再统一渲染 Markdown。这样实现简单，但牺牲了 Markdown 实时预览。
另一种策略是延迟渲染。把累积的内容切成短片段，只渲染已经确定完整的部分。例如，只有当文本中出现两个连续的代码围栏 ``` 时，才尝试渲染代码块；对未闭合的围栏，保留为纯文本。
js
function isCodeBlockComplete(text) {
const fenceMatches = text.match(/```/g);
return fenceMatches !== null && fenceMatches.length % 2 === 0;
}这种策略可以避免未闭合代码块的闪烁，但实现复杂度较高。实际项目中，可以在流式阶段使用纯文本加简单格式（例如行内代码、链接），流结束后再完整渲染 Markdown。
自动滚动与用户滚动冲突
流式输出时，内容逐行增加，页面需要自动滚动到底部，让用户看到最新文本。但当用户主动向上滚动查看历史消息时，自动滚动会打断阅读。常见做法是：在滚动容器上监听 scroll 事件，判断用户是否接近底部；只在用户处于底部附近时执行自动滚动。
jsx
const containerRef = useRef(null);
function isNearBottom() {
const el = containerRef.current;
return el.scrollHeight - el.scrollTop - el.clientHeight < 80;
}
function scrollToBottom() {
const el = containerRef.current;
el.scrollTop = el.scrollHeight;
}
function handleScroll() {
setIsSticky(isNearBottom());
}
useEffect(() => {
const el = containerRef.current;
el.addEventListener('scroll', handleScroll);
return () => el.removeEventListener('scroll', handleScroll);
}, []);
useEffect(() => {
if (isSticky) {
scrollToBottom();
}
}, [text, isSticky]);80 是示例阈值，具体值取决于设计。关键逻辑是：当用户未触底时，自动滚动暂停，避免把用户拉回底部。
可访问性设计
流式更新对屏幕阅读器用户并不友好。屏幕阅读器每次检测到内容变化都会重新朗读整个区域，如果每个 chunk 都触发一次朗读，用户会听到大量重复内容。
将流式输出区域放入 aria-live="polite" 容器中，可以让辅助技术平滑地处理变化：
jsx
<div
aria-live="polite"
aria-busy={status === 'loading' || status === 'streaming'}
>
{text}
</div>aria-busy 在流式输出期间标记区域忙碌，辅助技术会等待更新完成后再朗读。aria-live="polite" 表示不打断当前的任务，适合聊天内容；如果内容属于重要提示，可以使用 assertive。
对于逐字输出，不应让每字都触发一次 aria-live 通知。更好的做法是通知“生成中”，并在生成结束后完整朗读结果。这个可以通过把流式文本放入 aria-hidden 区域，结束后再暴露给辅助技术实现：
jsx
<div>
<p aria-hidden={status === 'streaming'}>{text}</p>
{status === 'streaming' && (
<p className="sr-only">正在生成回答，请稍候。</p>
)}
</div>其中 .sr-only 是只对屏幕阅读器可见的内部样式类。
AI 场景错误分类与处理
错误分类
LLM 接口的错误类型比较多，前端不能只把错误消息原样展示给用户。常见错误可以归为以下几类。
- 限流：请求过多，服务端返回 HTTP 429。
- 上下文超长：请求中的消息总长度超过模型窗口，通常会以 400 类状态码返回，错误信息中含 context length 等关键词。
- 鉴权失败：API key 无效或缺失，返回 401 或 403。
- 网络中断：请求没有收到响应，浏览器抛出
TypeError: Failed to fetch。 - 内容拦截：生成内容触发审核策略，返回 400、422 等状态码，错误信息中含 moderation、policy 等关键词。
前端可以根据状态码和错误文本进行统一映射：
js
function normalizeAIError(error) {
if (error.name === 'AbortError') {
return { type: 'cancelled', message: '生成已停止' };
}
if (error instanceof TypeError && error.message.includes('fetch')) {
return { type: 'network', message: '网络连接中断，请检查网络后重试' };
}
const status = error.status;
if (status === 429) {
return { type: 'rate_limit', message: '请求过于频繁，请稍后重试' };
}
if (status === 401 || status === 403) {
return { type: 'unauthorized', message: '身份验证失败，请检查 API Key 配置' };
}
const rawMessage = error.message || '';
if (rawMessage.includes('context length')) {
return { type: 'context_window_exceeded', message: '内容超出模型支持的长度，请精简对话' };
}
if (rawMessage.includes('moderation') || rawMessage.includes('policy')) {
return { type: 'content_policy', message: '内容被安全策略拦截，请调整输入后重试' };
}
return { type: 'unknown', message: '生成失败，请稍后重试' };
}这里的状态码和关键词覆盖的是常见模式。不同服务商的错误体结构不完全一致，前端需要根据实际接入的接口补充细节。
重试与退避策略
网络中断和限流可以重试，但重试不应该立即执行。限流错误需要等待一段时间，网络错误也不一定在重试瞬间恢复。简单的指数退避如下：
js
async function fetchWithRetry(requestFn, { maxRetries = 3 } = {}) {
let lastError;
for (let attempt = 0; attempt < maxRetries; attempt++) {
try {
return await requestFn();
} catch (error) {
lastError = error;
if (error.status === 429) {
const retryAfter = Number(error.headers?.get('retry-after')) || 2 ** attempt;
await sleep(retryAfter * 1000);
continue;
}
if (error.status >= 400 && error.status < 500 && error.status !== 429) {
// 鉴权失败、上下文超长等错误，重试也没有意义
break;
}
await sleep(2 ** attempt * 1000);
}
}
throw lastError;
}注意，maxRetries、retry-after 的解析都是可配置策略。对于 LLM 流式请求，重试必须发生在“尚未开始消费流”之前；如果流已经开始输出，中途失败不能直接重发同样请求，否则用户会看到重复内容。
停止生成、草稿保留与修改请求
用户主动停止生成后，已经输出的内容应当保留。用户可以选择复制这部分内容、修改提示词后重新生成，或者继续提问。
前端在调用 abort() 后，应当把当前状态置为 IDLE，但不清空 text。这样 UI 可以展示“已停止”的提示，同时保留已有结果。
jsx
function handleStop() {
stop();
setStatus('stopped');
}如果用户修改了提示词并再次点击发送，旧内容与新一轮输出之间需要明确分隔。可以考虑把每条消息作为独立的卡牌，新请求生成新的消息对象，避免相互覆盖。
降级方案与错误提示设计
当流式接口不可用时，可以考虑降级为非流式请求。降级策略不是自动隐式完成的，因为非流式请求的等待时间更长，需要用户知情。可以在错误处理 UI 中提供一个“以普通模式重试”的按钮，明确告知用户此模式不显示逐字输出。
错误提示本身需要遵循几条原则：
- 不暴露原始错误堆栈。
- 给出可执行操作，而不只是描述失败。
- 对可恢复错误（网络、限流）提供重试入口；对不可恢复错误（鉴权失败、上下文超长）提供修改建议。
jsx
function ErrorBanner({ error, onRetry, onEditQuery }) {
return (
<div role="alert" className="error-banner">
<p>{error.message}</p>
<div>
{error.type === 'network' || error.type === 'rate_limit' ? (
<button onClick={onRetry}>重新尝试</button>
) : null}
{error.type === 'context_window_exceeded' ? (
<button onClick={onEditQuery}>编辑消息</button>
) : null}
</div>
</div>
);
}role="alert" 用于把错误区域标记为重要提醒，辅助技术会优先朗读。
前端架构设计与协议隔离
协议隔离：StreamClient 适配层设计
不同模型服务商的流式协议存在细节差异。OpenAI 兼容接口使用 chat.completion.chunk 事件，Anthropic 使用 content_block_delta 事件[5][7]。如果业务组件直接解析这些协议，切换模型服务商时需要修改大量 UI 代码。因此，前端需要定义一个统一的 StreamClient 适配层。
一个最小接口定义如下：
ts
interface StreamClient {
request(body: ChatRequest, signal?: AbortSignal): AsyncIterable<StreamEvent>;
abort(): void;
}StreamEvent 是统一消息模型，下游 UI 只依赖这个模型，不关心协议细节：
ts
type StreamEvent =
| { type: 'start' }
| { type: 'delta'; content: string }
| { type: 'tool_call'; name: string; arguments: unknown }
| { type: 'done'; finishReason: string }
| { type: 'error'; error: AIError };OpenAI 兼容接口的实现可以把原始 chunk 转换为 StreamEvent：
js
class OpenAIStreamClient {
constructor({ apiKey, baseUrl }) {
this.apiKey = apiKey;
this.baseUrl = baseUrl;
}
async *request(body, signal) {
const response = await fetch(`${this.baseUrl}/chat/completions`, {
method: 'POST',
headers: {
'Content-Type': 'application/json',
Authorization: `Bearer ${this.apiKey}`
},
body: JSON.stringify({ ...body, stream: true }),
signal
});
if (!response.ok) {
throw new AIError(response.status, await response.text());
}
const reader = response.body.getReader();
const decoder = new TextDecoder();
let buffer = '';
while (true) {
const { done, value } = await reader.read();
if (done) break;
buffer += decoder.decode(value, { stream: true });
const { events, buffer: rest } = parseSSE(buffer);
buffer = rest;
for (const raw of events) {
yield this.parseEvent(raw);
}
}
}
parseEvent(raw) {
if (raw === '[DONE]') {
return { type: 'done', finishReason: 'stop' };
}
const json = JSON.parse(raw);
const delta = json.choices?.[0]?.delta;
if (delta?.content) {
return { type: 'delta', content: delta.content };
}
if (delta?.tool_calls) {
// 处理工具调用增量
}
return { type: 'start' };
}
}使用方只需要拿到 AsyncIterable<StreamEvent>，遍历并更新 UI。
统一消息模型与流式状态存储
流式状态存储有两个层次：协议层的 StreamEvent 与应用层的 UI 状态。协议层负责解析；应用层负责把 delta 追加到消息文本，把 tool_call 放入工具调用列表。
在 React 中，useReducer 适合这种场景。在 Vue 中，使用 ref 加函数同样可以完成。关键是让每个事件对应一个明确的 action，避免在组件内部散落异步更新逻辑。
组件划分与错误边界
组件划分应遵循一个原则：流的解析与 UI 展示解耦。
StreamProvider：管理 StreamClient 实例和状态机。MessageList：渲染消息列表，不关心数据来源。MessageItem：渲染单条消息，包括 Markdown、代码块、错误提示。ControlBar：渲染发送、停止、重试等操作按钮。
React 错误边界可以捕获渲染异常：
jsx
class StreamErrorBoundary extends React.Component {
state = { hasError: false };
static getDerivedStateFromError() {
return { hasError: true };
}
componentDidCatch(error, info) {
console.error('流式组件渲染错误：', error, info);
}
render() {
if (this.state.hasError) {
return <div>界面渲染失败，请刷新页面重试。</div>;
}
return this.props.children;
}
}错误边界只能捕获渲染阶段的错误，不能捕获事件处理器或异步代码中的错误。流的解析错误应当在 streamReducer 中处理，而不是依赖错误边界。
可观测性与前端埋点
流式 UI 的埋点要点是时间戳。以下指标可以采集：
- 请求发起时间：用户点击发送。
- 首字节时间：
response.body可读后。 - 首个 token 时间：解析出第一个业务 delta。
- 流结束时间：收到
done事件。 - 错误类型与发生阶段：请求前、传输中、解析中。
js
const metrics = {
requestStart: performance.now(),
ttft: null,
firstTokenAt: null,
endAt: null
};
// 收到第一个 delta 时
metrics.firstTokenAt = performance.now();
metrics.ttft = metrics.firstTokenAt - metrics.requestStart;
// 收到 done 时
metrics.endAt = performance.now();在发送埋点报告时，注意不要记录用户消息原文，避免隐私风险。
React 与 Vue 中的流式封装
React Hook：useStream 的基本实现
下面是一个基于 React 的 useStream Hook。它把 fetch、状态机、取消操作封装在一起：
js
import { useCallback, useEffect, useRef, useState } from 'react';
function useStream({ url, headers = {} } = {}) {
const [status, setStatus] = useState('idle');
const [text, setText] = useState('');
const [error, setError] = useState(null);
const abortRef = useRef(null);
const latestRequestRef = useRef(0);
const start = useCallback(
async (messages) => {
const requestId = ++latestRequestRef.current;
const controller = new AbortController();
abortRef.current = controller;
setStatus('loading');
setText('');
setError(null);
try {
const response = await fetch(url, {
method: 'POST',
headers: {
'Content-Type': 'application/json',
...headers
},
body: JSON.stringify({ messages, stream: true }),
signal: controller.signal
});
if (!response.ok) {
throw new Error(`HTTP ${response.status}`);
}
const reader = response.body.getReader();
const decoder = new TextDecoder();
let buffer = '';
let accumulatedText = '';
while (true) {
const { done, value } = await reader.read();
if (done) break;
if (requestId !== latestRequestRef.current) {
reader.cancel();
return;
}
buffer += decoder.decode(value, { stream: true });
const { events, buffer: rest } = parseSSE(buffer);
buffer = rest;
for (const raw of events) {
const delta = parseDelta(raw);
if (delta) {
accumulatedText += delta;
setText(accumulatedText);
setStatus('streaming');
}
}
}
if (requestId === latestRequestRef.current) {
setStatus('done');
}
} catch (err) {
if (err.name === 'AbortError' || requestId !== latestRequestRef.current) {
return;
}
setError(err);
setStatus('error');
}
},
[url, headers]
);
const stop = useCallback(() => {
latestRequestRef.current++;
abortRef.current?.abort();
}, []);
useEffect(() => {
return () => {
abortRef.current?.abort();
};
}, []);
return { status, text, error, start, stop };
}这个 Hook 的使用方式：
jsx
function Chat() {
const { status, text, error, start, stop } = useStream({
url: '/api/chat'
});
return (
<div>
<button onClick={() => start([{ role: 'user', content: '你好' }])}>
发送
</button>
<button onClick={stop} disabled={status !== 'streaming'}>
停止
</button>
<div>{text}</div>
{error && <div>{error.message}</div>}
</div>
);
}start 内部的 setText 在每次 delta 后都会触发 React 渲染。高频 chunk 时可配合前面提到的调度策略。
Vue Composable：useStream 的基本实现
Vue 3 的组合式 API 写法类似：
js
import { ref } from 'vue';
export function useStream(url) {
const status = ref('idle');
const text = ref('');
const error = ref(null);
const controller = ref(null);
let latestRequestId = 0;
async function start(messages) {
const requestId = ++latestRequestId;
const abortController = new AbortController();
controller.value = abortController;
status.value = 'loading';
text.value = '';
error.value = null;
try {
const response = await fetch(url, {
method: 'POST',
headers: { 'Content-Type': 'application/json' },
body: JSON.stringify({ messages, stream: true }),
signal: abortController.signal
});
if (!response.ok) {
throw new Error(`HTTP ${response.status}`);
}
const reader = response.body.getReader();
const decoder = new TextDecoder();
let buffer = '';
while (true) {
const { done, value } = await reader.read();
if (done) break;
if (requestId !== latestRequestId) {
reader.cancel();
return;
}
buffer += decoder.decode(value, { stream: true });
const { events, buffer: rest } = parseSSE(buffer);
buffer = rest;
for (const raw of events) {
const delta = parseDelta(raw);
if (delta) {
text.value += delta;
status.value = 'streaming';
}
}
}
if (requestId === latestRequestId) {
status.value = 'done';
}
} catch (err) {
if (err.name === 'AbortError' || requestId !== latestRequestId) {
return;
}
error.value = err;
status.value = 'error';
}
}
function stop() {
latestRequestId++;
controller.value?.abort();
}
return { status, text, error, start, stop };
}增量 Markdown 渲染与 XSS 防御
这一节只考虑常见的 Markdown 渲染风险。LLM 输出可能包含 HTML 标签。如果 Markdown 渲染器允许原始 HTML 通过，攻击者可以通过提示词注入的方式，让模型输出 <img src=x onerror=alert(1)>。一旦前端将其插入页面，这段代码就会被浏览器执行。
markdown-it 默认并不渲染原始 HTML，而是把标签转义为实体。这是安全的默认行为。如果你出于产品需求开启 html: true 选项，例如允许 LLM 输出表格或 iframe，那么必须在上屏前兜底清洗。
js
import MarkdownIt from 'markdown-it';
import DOMPurify from 'dompurify';
const md = new MarkdownIt({
html: true,
linkify: true,
breaks: true
});
function renderMarkdown(text) {
const rawHtml = md.render(text);
return DOMPurify.sanitize(rawHtml, { USE_PROFILES: { html: true } });
}DOMPurify.sanitize() 会移除 onerror 等事件属性、javascript: 协议链接以及其他危险内容。对于带 html: false 的 markdown-it，仍然建议做一次清洗，因为 Markdown 链接地址可能包含非法协议：
md
[点击这里](javascript:alert(1))markdown-it 默认会验证链接协议，只允许安全协议。但这不是所有渲染器的默认行为，因此不依赖单一防御机制更可靠。
在流式输出阶段，不建议频繁调用 marked 或 markdown-it 重新渲染整篇内容。可以采用以下策略：
- 流式期间显示纯文本，并用换行符保留段落结构。
- 在收到
done事件后，执行一次完整 Markdown 渲染。 - 如果必须流式渲染 Markdown，可以在渲染器外部缓存
text，每 100ms 左右重新渲染一次，并让生成结果进入 DOMPurify。
渲染后的 HTML 必须通过 dangerouslySetInnerHTML（React）或 v-html（Vue）挂载。这两个 API 会直接设置元素的 innerHTML，不做任何转义。保证 v-html 安全的前提是数据确实已经被清洗。
断线重连与停止生成的 UI 实现
断线重连需要考虑“已经生成的内容是否保留”。通常保留，因为重新生成会得到不同结果。UI 可以提供一个“继续生成”的入口，在中断状态下重新发起请求，并携带之前的上下文：
jsx
function StreamMessages({ messages, status, onRetry }) {
return (
<div>
{messages.map((message) => (
<MessageItem key={message.id} message={message} />
))}
{status === 'error' && (
<div className="reconnect-bar">
<span>连接已断开，已生成的内容已保留。</span>
<button onClick={onRetry}>继续生成</button>
</div>
)}
</div>
);
}停止生成与断线重连的状态需要在 UI 上区分：
- 用户主动停止：显示“已停止”按钮，可让用户“重新生成”。
- 网络中断：显示“连接断开”，可让用户“尝试恢复”或“重新生成”。
测试与性能度量
构造 Mock Stream
测试流式 UI 的关键是能在浏览器或测试环境中产生一个可读的流。ReadableStream 是标准 Web API，可以直接构造：
js
function createMockSSEStream() {
const encoder = new TextEncoder();
const events = [
'data: {"choices":[{"delta":{"role":"assistant"}}]}\n\n',
'data: {"choices":[{"delta":{"content":"你好"}}]}\n\n',
'data: {"choices":[{"delta":{"content":"，"}}]}\n\n',
'data: {"choices":[{"delta":{"content":"世界"}}]}\n\n',
'data: {"choices":[{"delta":{},"finish_reason":"stop"}]}\n\n',
'data: [DONE]\n\n'
];
return new ReadableStream({
start(controller) {
events.forEach((event, index) => {
controller.enqueue(encoder.encode(event));
if (index === events.length - 1) {
controller.close();
}
});
}
});
}在测试中，把它包装成 Response：
js
const mockResponse = new Response(createMockSSEStream(), {
status: 200,
headers: { 'Content-Type': 'text/event-stream' }
});Response 是浏览器环境内置对象，在 jsdom（Vitest/Jest 环境）中需要确认全局存在。Node.js 18 起提供了全局 Response，测试环境可以直接使用。
错误注入测试
错误注入需要覆盖两类场景：HTTP 层错误和流传输中错误。
HTTP 错误注入：
js
const mockErrorResponse = new Response('rate limit', {
status: 429,
headers: { 'Retry-After': '1' }
});流传输中错误注入可以通过在 ReadableStream 中突然 error：
js
function createBrokenStream() {
const encoder = new TextEncoder();
return new ReadableStream({
start(controller) {
controller.enqueue(encoder.encode('data: {"choices":[{"delta":{"content":"部分内容"}}]}\n\n'));
controller.error(new Error('network reset'));
}
});
}这样 reader.read() 会在第一个 chunk 之后抛出错误，前端应把状态切换到 error，并保留已有文本。
自动化测试用例设计
以 React + Vitest 为例。测试目标是验证从请求发出到内容渲染的完整流程。
jsx
import { render, screen, waitFor } from '@testing-library/react';
import userEvent from '@testing-library/user-event';
import { Chat } from './Chat';
function mockStreamResponse() {
global.fetch = vi.fn().mockResolvedValue(
new Response(createMockSSEStream(), {
headers: { 'Content-Type': 'text/event-stream' }
})
);
}
test('点击发送后展示流式内容', async () => {
mockStreamResponse();
render(<Chat />);
userEvent.click(screen.getByRole('button', { name: '发送' }));
await waitFor(() => {
expect(screen.getByText(/你好/)).toBeInTheDocument();
});
await waitFor(() => {
expect(screen.getByText(/世界/)).toBeInTheDocument();
});
expect(global.fetch).toHaveBeenCalledWith(
expect.stringContaining('/chat'),
expect.objectContaining({
method: 'POST'
})
);
});测试停止生成：
jsx
test('点击停止后不渲染新内容', async () => {
global.fetch = vi.fn().mockResolvedValue(
new Response(createSlowStream(), {
headers: { 'Content-Type': 'text/event-stream' }
})
);
render(<Chat />);
userEvent.click(screen.getByRole('button', { name: '发送' }));
userEvent.click(screen.getByRole('button', { name: '停止' }));
await waitFor(() => {
expect(screen.getByText(/已停止/)).toBeInTheDocument();
});
});其中 createSlowStream() 可以设计为多个 chunk 之间有延迟的流。
Response 对象的 body 是单次读取的。多个测试间需要各自创建新的 Response 实例，不能复用同一个流。
性能度量：TTFT 与 token/s
性能度量的两个核心指标：
- TTFT（Time To First Token）：从请求发起到收到第一个内容 delta 的时间。
- token/s：每秒渲染的 token 数或字符数，反映输出速度。
TTFT 的记录方式：
js
const startTime = performance.now();
let ttft = null;
let tokenCount = 0;
for await (const event of streamClient.request(body)) {
if (event.type === 'delta') {
if (ttft === null) {
ttft = performance.now() - startTime;
}
tokenCount += countTokens(event.content);
}
if (event.type === 'done') {
const totalTime = (performance.now() - startTime) / 1000;
const tokenPerSecond = tokenCount / totalTime;
reportMetrics({
ttft,
tokenPerSecond,
totalTime
});
}
}countTokens 在浏览器中很难精确实现。作为前端度量，可以用字符数代替 token 数，或使用服务端返回的 usage 字段。OpenAI 兼容接口中，stream_options: { include_usage: true } 可以让最后一个 chunk 携带用法统计[4]。
Playwright 捕获流式响应
Playwright 中测试流式 UI 的常见做法是拦截网络请求，模拟一个 SSE 响应。page.route() 可以拦截指定 URL，并用 route.fulfill() 返回响应。要模拟真正的“流式”到达，需要响应体在发送后再逐步刷新。
一个可行做法是让 route.fulfill() 返回预先构建好的完整 SSE 字符串。浏览器读取响应体时，会逐步解析其中的事件，一般足以触发流式渲染：
js
import { test, expect } from '@playwright/test';
const SSE_BODY = [
'data: {"choices":[{"delta":{"role":"assistant"}}]}\n',
'data: {"choices":[{"delta":{"content":"你"}}]}\n',
'data: {"choices":[{"delta":{"content":"好"}}]}\n',
'data: {"choices":[{"delta":{},"finish_reason":"stop"}]}\n',
'data: [DONE]\n'
].join('\n');
test('页面显示流式回复', async ({ page }) => {
await page.route('**/api/chat', (route) => {
route.fulfill({
status: 200,
contentType: 'text/event-stream',
body: SSE_BODY
});
});
await page.goto('/');
await page.getByRole('button', { name: '发送' }).click();
await expect(page.getByText('你好')).toBeVisible();
});如果希望模拟有间隔的真实网络传输，可以启动一个本地 Node HTTP 服务器，按时间间隔向 res.write() 写入 SSE 事件：
js
import http from 'node:http';
function createSSEServer() {
return http.createServer((req, res) => {
res.writeHead(200, { 'content-type': 'text/event-stream' });
res.write('data: {"choices":[{"delta":{"content":"你"}}]}\n\n');
setTimeout(() => {
res.write('data: {"choices":[{"delta":{"content":"好"}}]}\n\n');
res.end('data: [DONE]\n\n');
}, 50);
});
}然后在 Playwright 的 webServer 配置中启动应用和这个 mock 服务，让应用请求指向 mock 服务。这种方式更接近真实网络行为，也方便模拟中断、超时等异常。
生态与标准趋势
Vercel AI SDK 的流式抽象
Vercel AI SDK 提供了面向 AI 应用的 React/Vue/Svelte 封装。其核心思路与上文所述一致：在服务端把模型响应转为标准流式格式，在客户端通过 useChat 等 hook 消费。
AI SDK 把流式协议抽象为统一的 AIStream 接口，React 的 useChat 内部管理消息数组、isLoading 状态、错误对象以及 stop() 方法。这套抽象让应用层不直接接触 SSE 格式，同时可以在不同的模型服务商之间切换。理解其设计有助于对比自己实现时需要的边界。
OpenAI 与 Anthropic 的流式事件规范
OpenAI Chat Completions 的流式响应由多个 chat.completion.chunk 事件组成[4]。每个 chunk 的 choices[0].delta 包含增量内容，finish_reason 仅在最后一块有意义。事件以 data: [DONE] 标记结束[3]。
Anthropic Messages API 的流式事件类型不同，例如 message_start、content_block_delta、message_delta、message_stop。文本增量在 content_block_delta 中。当协议转换层需要把 Anthropic 流映射到 OpenAI 兼容端时，必须过滤掉非业务事件（例如 ping），否则混合协议会让客户端解析失败[5]。
Agent 与 Function Calling 事件流
多轮 Agent 交互中，流式响应不仅包含文本，还会包含工具调用。OpenAI 兼容接口中，delta 可能包含 tool_calls 字段，且 role 可能是 tool，不一定是 assistant[7]。前端状态机需要区分“文本增量”和“工具调用增量”，后者需要累积参数 JSON，并等待工具执行完成后继续下一轮模型调用。
js
function isToolCallDelta(delta) {
return Array.isArray(delta.tool_calls) && delta.tool_calls.length > 0;
}UI 上，工具调用通常以独立的卡片呈现，而不是混入文本流。这样用户能分辨“模型在调用工具”与“模型在生成答案”。
Web Streams 标准与 AI 交互范式演进
ReadableStream、WritableStream、TransformStream 组成的 Streams 标准正在被更多浏览器 API 采纳。Chrome 105 起，ReadableStream 可以作为 fetch 请求体，使流式上传成为可能[2]。这一能力对 AI 交互的意义在于：前端可以把用户语音输入、屏幕录制或长文档分段流式上传给模型，而不需要一次性发送完整内容。
未来 LLM 前端可能会看到更多基于流的交互模式：
- 客户端流式上传用户上下文，服务端边接收边处理。
- 服务端通过同一连接发送文本、工具状态和结构化数据。
- 前端用
TransformStream在浏览器内做实时内容后处理。
这些都建立在上文所述的基础概念之上：流式协议解析、增量状态管理、渲染调度与错误恢复。
参考链接
- [1] https://developer.mozilla.org/en-US/docs/Web/API/ReadableStream
- [2] https://developer.chrome.com/docs/capabilities/web-apis/fetch-streaming-requests
- [3] https://github.com/link-assistant/formal-ai/issues/604
- [4] https://developers.openai.com/api/reference/resources/chat/subresources/completions/streaming-events
- [5] https://github.com/Wei-Shaw/sub2api/issues/5203
- [7] https://community.openai.com/t/openai-chat-completion-stream-response-delta-contains-roles-other-than-assistant/1262781