1. 从“等待”到“流淌”为什么我们需要流式输出如果你最近在捣鼓大语言模型LLM的应用比如自己部署一个类似ChatGPT的聊天界面一定会遇到一个核心体验问题等待感。传统的API调用比如一个普通的HTTP请求是“一问一答”的模式。你发送问题服务器端的大模型吭哧吭哧开始推理生成全部文本直到最后一个字写完才一次性打包成完整的响应体通过HTTP返回给你。这个过程用户面对的是一个空白的输入框或一个旋转的加载图标除了等待别无他法。对于生成长文本的场景这种等待可能是几十秒甚至几分钟体验极其糟糕。流式输出Streaming Output就是为了解决这个“等待”问题而生的。它的核心思想是“边生成边发送边渲染”。服务器不再等大模型生成完整的回答而是模型每生成一个词元token服务器就立刻把这个词元发送给客户端。客户端在收到第一个数据块时就可以立即开始渲染让用户看到文字是逐字“流淌”出来的。这种即时反馈极大地提升了交互的流畅度和沉浸感是当前AI应用前端体验的基石。那么这个“流淌”的过程在技术上是如何实现的呢这背后是一套从前端到后端的完整技术链。在前端我们主要和三个核心概念打交道SSEServer-Sent Events、Fetch API 与 ReadableStream以及用于处理二进制数据的Uint8Array。很多人可能用过类似EventSource的库来对接SSE但在现代前端框架如 Vue 3 的生态下尤其是在追求更精细控制和更好性能时直接使用 Fetch API 来处理流式响应正成为更主流和灵活的选择。本文将从一个实践者的角度拆解从建立连接到最终渲染的完整流程并分享在 Vue 3 Vite 项目中调试这些流式数据时我踩过的坑和总结的技巧。2. 协议层基石理解 SSE 的本质与局限当我们谈论大模型的流式输出时SSE 是无法绕开的一个协议。很多人对它有个误解认为它是一种全新的、复杂的黑科技。其实不然SSE 本质上是一个轻量级的、基于 HTTP 的协议标准。它的设计非常简洁在客户端和服务器之间建立一个单向的、长久的 HTTP 连接服务器可以通过这个连接持续地向客户端推送数据。SSE 的工作原理可以这样理解客户端发起请求浏览器或前端应用向一个特定的服务器端点发送一个普通的 HTTP GET 请求。服务器保持连接服务器收到请求后并不立即关闭连接而是将响应的Content-Type设置为text/event-stream并保持 TCP 连接打开。服务器推送事件每当有新的数据需要发送时服务器就向这个连接写入遵循特定格式的文本数据块。每个数据块称为一个“事件”Event格式大致如下event: message data: {content: 这是, finish_reason: null} data: {content: 一段, finish_reason: null} event: message data: {content: 流式, finish_reason: null} data: {content: 文本。, finish_reason: stop}注意每个事件以两个换行符\n\n结束。data:行可以有多行最终会拼接成一个完整的字符串。客户端监听处理客户端通过EventSourceAPI 或类似库监听这个连接。每当收到一个完整的事件块就会触发一个message事件开发者可以在事件回调中获取并处理data字段的内容。为什么大模型场景常用 SSE简单易用对于标准的文本流推送浏览器原生EventSource使用起来非常简单。自动重连EventSource内置了连接断开后的自动重连机制。文本友好SSE 设计上就是传输文本的与大模型返回的 JSON 或纯文本数据天然契合。然而SSE 的局限性也很明显这促使我们寻找更底层的方案仅支持 GET 请求EventSource只能发起 GET 请求。而大模型的 API 调用往往需要携带复杂的提示词prompt和参数这些数据放在 GET 请求的 URL 查询字符串中非常笨重且有长度限制更适合通过 POST 请求的 Body 发送。协议与实现耦合EventSource对响应格式有严格要求text/event-stream并且隐藏了底层连接细节。当我们需要更精细的控制如自定义请求头、处理非标准格式、手动管理连接生命周期时它就力不从心了。单向通信SSE 是服务器向客户端的单向通信。虽然对于单纯的输出流这没问题但在一些需要双向交互验证的复杂场景下显得不足。正是这些限制让我们把目光投向了更基础、更强大的 Web API——Fetch。3. 核心武器库解剖 Fetch API、ReadableStream 与 Uint8Array要突破 SSE 的限制实现一个功能完备、可控性强的流式请求我们需要深入下一层使用 Fetch API 直接处理流式响应体。这是现代前端处理流数据的核心手段。3.1 Fetch API不仅仅是“获取”Fetch API 提供了一个名为fetch()的全局方法用于发起网络请求。它返回一个 Promise该 Promise 在收到 HTTP 响应头后就会解析resolve解析的值是一个Response对象。关键在于这个Response对象的body属性本身就是一个ReadableStream。这意味着我们不必等待整个响应体下载完毕就可以开始读取数据。这对于大模型动辄几十KB甚至几MB的流式响应来说是至关重要的性能优化。一个发起流式 POST 请求的示例const response await fetch(/api/chat/stream, { method: POST, headers: { Content-Type: application/json, }, body: JSON.stringify({ model: gpt-3.5-turbo, messages: [{ role: user, content: 你好请介绍一下你自己。 }], stream: true // 关键告诉后端需要流式响应 }) }); // 此时连接已建立响应头已收到但响应体body还在传输中 const readableStream response.body; // 接下来就可以处理这个流了3.2 ReadableStream数据流的抽象ReadableStream是 Web Streams API 的一部分它代表了一个可读的二进制数据流。你可以把它想象成一根水管数据像水一样从服务器端流过来。我们的任务是安装一个“水龙头”reader来接水。处理ReadableStream的基本模式是获取一个读取器reader。循环调用reader.read()方法。这个方法返回一个 Promise当流中有数据可读时Promise 解析并返回一个对象包含两个属性value(本次读取到的数据块是一个Uint8Array) 和done(流是否已结束)。处理value然后继续读取直到done为true。const reader readableStream.getReader(); const decoder new TextDecoder(utf-8); // 用于将二进制数据解码成字符串 let accumulatedText ; try { while (true) { const { done, value } await reader.read(); if (done) { console.log(流式传输结束); break; } // value 是一个 Uint8Array需要解码 const chunk decoder.decode(value, { stream: true }); accumulatedText chunk; // 这里可以更新UI例如将 accumulatedText 设置到 Vue 的响应式变量中 console.log(收到数据块:, chunk); } } catch (error) { console.error(读取流失败:, error); } finally { reader.releaseLock(); // 重要释放读取器锁 }注意decoder.decode(value, { stream: true })中的{ stream: true }选项。这告诉解码器当前解码的数据可能是一个多字节字符如中文的一部分不要抛出错误等待后续数据到来再一起解码。这是处理流式文本时的一个关键细节否则你可能会在控制台看到乱码或解码错误。3.3 Uint8Array二进制数据的容器为什么reader.read()返回的value是Uint8Array而不是直接的字符串因为网络传输的本质是二进制字节流。Uint8Array是一个表示 8 位无符号整数数组的类型化数组它是 JavaScript 中处理原始二进制数据的高效方式。服务器发送的每一个数据块在 TCP 层面都是一串字节。Fetch API 将这些字节原封不动地包装成Uint8Array交给我们。这样做的好处是保真保留了数据的原始格式无论是纯文本、JSON 片段还是其他二进制数据如图片碎片前端都可以按需处理。高效避免了不必要的字符串编码/解码开销直到我们需要使用文本内容时才用TextDecoder进行解码。一个常见的坑数据块的分割与拼接服务器端在写入流时并没有义务保证每次写入的数据块恰好是一个完整的逻辑单元比如一个完整的 JSON 对象或一句话。它可能因为网络缓冲区、服务器端框架的实现等原因将一次逻辑上完整的数据拆分成多个Uint8Array发送也可能将多次逻辑上小的数据合并成一个大的Uint8Array发送。例如后端可能发送了这样的数据数据块1: {content: Hello 数据块2: , world!, finish: false}\n\n如果你对每个数据块都尝试JSON.parse()那么在数据块1处就会抛出错误因为它不是一个合法的 JSON。解决方案是使用“缓冲区”和“分隔符”。大模型流式接口通常使用\n\n或\n作为数据块之间的分隔符即所谓的 “data: ...\n\n” 格式或简化的 “{...}\n” 格式。我们的前端逻辑需要将收到的二进制数据解码成字符串后缓存在一个变量里然后根据分隔符来分割出完整的逻辑数据块。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 lines buffer.split(\n); // 最后一行可能是不完整的保留在缓冲区中 buffer lines.pop() || ; for (const line of lines) { const trimmedLine line.trim(); if (!trimmedLine || trimmedLine data: [DONE]) continue; // 忽略空行和结束标记 if (trimmedLine.startsWith(data: )) { const jsonStr trimmedLine.substring(6); // 去掉 data: 前缀 try { const data JSON.parse(jsonStr); // 处理真正的数据如 data.choices[0].delta.content console.log(解析后的数据:, data); } catch (e) { console.warn(解析JSON失败可能是不完整数据:, jsonStr); } } } } // 循环结束后记得处理缓冲区中可能残留的最后一小段数据 if (buffer.trim()) { console.warn(流结束缓冲区仍有未处理数据:, buffer); }这段代码展示了一个健壮的流式数据解析器核心逻辑解码 - 缓冲 - 按分隔符分割 - 逐条解析。这是手动处理大模型流式响应时最需要理解和实现的部分。4. 在 Vue 3 项目中构建流式对话组件理解了底层原理我们就可以在 Vue 3 项目中构建一个真实的、可复用的流式对话组件。我们将使用script setup语法和 Composition API并搭配 Vite 构建工具。4.1 组件结构与状态设计首先我们设计组件的状态。一个典型的聊天界面需要消息列表、当前用户输入、加载状态。!-- ChatStream.vue -- script setup import { ref, reactive } from vue; // 消息列表每个消息包含角色和内容 const messages ref([ { role: assistant, content: 你好我是AI助手有什么可以帮您 } ]); // 用户输入 const userInput ref(); // 是否正在流式接收中 const isLoading ref(false); // 当前助手消息的索引用于追加流式内容 const currentAssistantMessageIndex ref(-1); /script template div classchat-container div classmessages div v-for(msg, index) in messages :keyindex :class[message, msg.role] strong{{ msg.role user ? 你 : 助手 }}:/strong {{ msg.content }} /div div v-ifisLoading classloading-indicator思考中.../div /div div classinput-area textarea v-modeluserInput keydown.enter.preventsendMessage :disabledisLoading/textarea button clicksendMessage :disabledisLoading || !userInput.trim()发送/button /div /div /template4.2 核心流式请求函数接下来是核心的sendMessage函数。我们将使用 Fetch API 来处理流式响应。script setup // ... 其他状态定义 const sendMessage async () { if (!userInput.value.trim() || isLoading.value) return; const userMessage userInput.value.trim(); userInput.value ; // 清空输入框 // 将用户消息添加到列表 messages.value.push({ role: user, content: userMessage }); // 为助手创建一个初始为空的消息占位 messages.value.push({ role: assistant, content: }); currentAssistantMessageIndex.value messages.value.length - 1; isLoading.value true; try { const response await fetch(http://your-backend-api/chat/stream, { method: POST, headers: { Content-Type: application/json, // 如果需要认证可以在这里添加 Authorization 头 // Authorization: Bearer ${yourToken} }, body: JSON.stringify({ model: gpt-3.5-turbo, messages: [ ...messages.value.slice(0, -1).map(m ({ role: m.role, content: m.content })), { role: user, content: userMessage } ], stream: true, temperature: 0.7, }) }); if (!response.ok || !response.body) { throw new Error(网络请求失败: ${response.status}); } await handleStreamResponse(response); } catch (error) { console.error(请求出错:, error); // 更新最后一条助手消息为错误信息 messages.value[currentAssistantMessageIndex.value].content 抱歉出错了: ${error.message}; } finally { isLoading.value false; currentAssistantMessageIndex.value -1; } }; const handleStreamResponse async (response) { const reader response.body.getReader(); const decoder new TextDecoder(utf-8); let buffer ; try { while (true) { const { done, value } await reader.read(); if (done) { // 流结束处理缓冲区可能残留的数据如果有 processBuffer(buffer); break; } // 解码并追加到缓冲区 buffer decoder.decode(value, { stream: true }); // 处理缓冲区中的完整行 buffer processBuffer(buffer); } } catch (error) { console.error(读取流时出错:, error); throw error; } finally { reader.releaseLock(); } }; const processBuffer (buffer) { const lines buffer.split(\n); // 保留最后一行可能不完整作为新的缓冲区 const newBuffer lines.pop() || ; for (const line of lines) { const trimmedLine line.trim(); // 忽略空行和特殊事件行 if (!trimmedLine || trimmedLine.startsWith(event:)) continue; if (trimmedLine data: [DONE]) { // 流式传输结束信号 return newBuffer; } if (trimmedLine.startsWith(data: )) { const jsonStr trimmedLine.substring(6); // 移除 data: 前缀 try { const parsed JSON.parse(jsonStr); // 假设后端返回格式类似 OpenAI API: { choices: [{ delta: { content: ... } }] } const contentDelta parsed.choices?.[0]?.delta?.content; if (contentDelta) { // 关键更新 Vue 响应式数据触发视图更新 messages.value[currentAssistantMessageIndex.value].content contentDelta; } // 也可以处理 finish_reason 等字段 if (parsed.choices?.[0]?.finish_reason) { console.log(生成结束原因:, parsed.choices[0].finish_reason); } } catch (e) { // 忽略解析错误可能是不完整的 JSON 片段 console.warn(解析数据行失败可能数据不完整:, trimmedLine); } } } return newBuffer; }; /script这个实现包含了几个关键点错误处理对网络请求失败和流读取失败都进行了捕获。缓冲区管理processBuffer函数负责按\n分割并解析数据正确处理了数据块可能被拆分或合并的情况。响应式更新通过直接修改messages.value[currentAssistantMessageIndex.value].content来追加内容Vue 的响应式系统会自动检测到变化并更新 DOM实现文字的逐字打印效果。资源清理在finally块中调用reader.releaseLock()确保读取器被正确释放。4.3 用户体验优化添加打字机效果与中断控制基本的流式展示已经完成但我们可以做得更好。打字机效果直接追加内容可能太快可以添加一个简单的动画模拟逐字打印。我们可以使用setTimeout或requestAnimationFrame来缓冲。// 在 processBuffer 函数中替换直接的内容追加 if (contentDelta) { // 改为调用一个打字机函数 typewriterEffect(contentDelta); } const typewriterEffect (text) { let i 0; const speed 20; // 每个字符的间隔毫秒 const targetIndex currentAssistantMessageIndex.value; const type () { if (i text.length targetIndex currentAssistantMessageIndex.value) { messages.value[targetIndex].content text.charAt(i); i; setTimeout(type, speed); } }; type(); };注意这个简单实现有个问题如果流式数据到达很快会创建很多定时器。更优的方案是将所有到达的contentDelta先缓存在一个队列里由一个统一的定时器消费。中断控制用户可能在生成过程中希望停止。我们需要提供一个“停止”按钮并能够中止网络请求。script setup // 新增一个 AbortController 用于中断请求 let abortController null; const sendMessage async () { if (!userInput.value.trim() || isLoading.value) return; // 创建新的 AbortController abortController new AbortController(); isLoading.value true; // ... 其他初始化代码 try { const response await fetch(http://your-backend-api/chat/stream, { // ... 其他配置 signal: abortController.signal // 传入中断信号 }); // ... 处理响应 } catch (error) { // 如果是主动中断错误名称为 AbortError if (error.name AbortError) { console.log(请求被用户中断); messages.value[currentAssistantMessageIndex.value].content \n\n[已中断]; } else { // ... 其他错误处理 } } finally { // ... 清理工作 abortController null; } }; // 停止生成函数 const stopGenerating () { if (abortController) { abortController.abort(); // 触发 AbortError } }; /script template !-- 在按钮区域添加一个停止按钮 -- div classinput-area button clickstopGenerating v-ifisLoading停止/button !-- ... 其他按钮 -- /div /template通过AbortController我们实现了对正在进行的 Fetch 请求的控制提升了应用的交互性。5. Vue 3 Vite 项目中的流式调试实战与避坑指南在开发过程中流式相关的 bug 往往比较隐蔽因为涉及异步数据的分块到达和解析。下面分享我在 Vue 3 Vite 项目中调试流式应用时总结的一套方法和遇到的典型问题。5.1 调试工具与技巧浏览器开发者工具 - 网络面板查看请求确认你的 POST 请求是否成功发出Content-Type和body是否正确。查看响应这是最重要的调试步骤。找到你的流式请求点击它在“响应”Response标签页里你看到的不是最终结果而是服务器持续推送过来的原始数据流。你可以在这里直观地看到数据格式是否是data: {...}\n\n是否有不该出现的字符数据块是否完整。复制响应体你可以将看到的原始响应数据复制出来用于离线分析和单元测试。浏览器开发者工具 - 控制台在你的handleStreamResponse和processBuffer函数中大量使用console.log。打印出原始的value(Uint8Array)打印出解码后的buffer打印出分割后的每一行line打印出尝试解析的jsonStr和解析结果parsed。通过日志流水线你可以精准定位问题发生在哪个环节是数据没收到解码出错分割不对还是 JSON 格式不符构建一个模拟后端 在开发初期或者后端接口不稳定时在本地模拟一个流式接口极其有用。你可以用 Node.js 的http模块或者 Express 快速搭建一个。// mock-server.js (使用 Express) import express from express; const app express(); app.use(express.json()); app.post(/api/mock-stream, (req, res) { res.setHeader(Content-Type, text/event-stream); res.setHeader(Cache-Control, no-cache); res.setHeader(Connection, keep-alive); const message 这是一段模拟的流式响应。; let index 0; const intervalId setInterval(() { if (index message.length) { const chunk message.substring(index, index 1); // 每次发送一个字 const data JSON.stringify({ choices: [{ delta: { content: chunk } }] }); res.write(data: ${data}\n\n); index; } else { res.write(data: [DONE]\n\n); clearInterval(intervalId); res.end(); } }, 50); // 每50毫秒发送一个字 }); app.listen(3001, () console.log(Mock server running on port 3001));这样你就可以完全控制返回的数据格式和速度前端代码可以独立于真实后端进行开发和调试。5.2 常见问题与解决方案问题一控制台报错 “Unexpected token in JSON at position 0”现象在JSON.parse(jsonStr)时抛出错误提示 JSON 不合法。排查检查网络面板的响应体。很可能服务器返回的不是你期望的 JSON 流而是一个 HTML 页面比如 404 或 500 错误页。响应体的开头是html或!DOCTYPE。检查请求 URL 和代理配置Vite 的server.proxy是否正确确保请求真正到达了后端流式接口而不是静态文件服务器。解决修正 API 地址或代理配置。确保后端接口正确设置了Content-Type: text/event-stream并返回了正确的流式数据。问题二中文显示乱码或出现“”字符现象接收到的文本中中文变成了乱码或问号方块。排查检查TextDecoder的编码是否设置为utf-8。这是目前 Web 标准的通用编码。关键确认decoder.decode(value, { stream: true })中传入了{ stream: true }选项。没有这个选项当Uint8Array恰好在一个多字节字符如一个中文字符占3个字节的中间被截断时解码会失败并输出替换字符。检查服务器端编码是否也是 UTF-8。解决确保正确使用TextDecoder和stream: true选项。问题三数据接收不完整或拼接错误现象UI 上显示的文字缺失、重复或者 JSON 解析频繁失败。排查在processBuffer函数中打印每一行trimmedLine观察数据格式是否严格符合data: {...}。有时服务器可能多发了空格、少了换行或者使用了不同的前缀如data: 和data:。检查你的缓冲区逻辑。确保lines.pop()正确地保留了不完整的最后一行。在流结束后的finally块或done为true时检查buffer变量是否还有未处理的数据。服务器可能使用了\r\n作为换行符。使用buffer.split(/\r?\n/)可以同时兼容\n和\r\n。解决根据服务器实际返回的格式调整数据分割和前缀剥离的逻辑。增强解析器的容错性比如尝试匹配trimmedLine.match(/^data:\s*(.)/)来提取数据部分。问题四Vue 响应式更新不及时或性能问题现象文字不是逐字出现而是卡顿一下然后大段出现或者界面在流式接收时变得很卡。排查更新频率过高如果每个 token可能是一个字或一个词都触发一次 Vue 的响应式更新和 DOM 渲染对于长文本来说频率太高了。可以尝试“节流”累积一定数量的字符比如10个或固定时间间隔比如50毫秒再更新一次message.content。使用了深度监视或计算属性如果组件中对messages数组使用了watch且设置了deep: true或者有依赖它的复杂计算属性每次内容追加都会触发昂贵的计算。解决实现一个简单的节流更新let updateBuffer ; let updateTimer null; const scheduleUpdate (delta) { updateBuffer delta; if (!updateTimer) { updateTimer setTimeout(() { messages.value[currentAssistantMessageIndex.value].content updateBuffer; updateBuffer ; updateTimer null; }, 50); // 每50毫秒批量更新一次 } }; // 在 processBuffer 中用 scheduleUpdate(contentDelta) 替换直接的内容追加审查组件中的watch和computed避免不必要的深度监听。问题五在开发热重载HMR时流连接异常现象使用 Vite 开发时每次保存文件触发热更新后之前的流式连接可能不会自动关闭导致内存泄漏或错误。排查Vite 的 HMR 会重新执行模块代码但不会自动清理之前组件实例中创建的AbortController或持有的ReadableStreamreader。解决在组件的onUnmounted生命周期钩子中执行清理操作。script setup import { onUnmounted } from vue; // ... 其他代码 onUnmounted(() { if (abortController) { abortController.abort(); } // 如果有 reader也需要考虑释放锁但通常 reader 会在请求结束时释放。 // 更稳妥的做法是在发送新请求前检查并中断旧请求。 }); /script流式输出的实现是一个融合了网络协议、数据流处理和前端框架响应的综合课题。从理解 SSE 的局限到掌握 Fetch ReadableStream 的底层操作再到在 Vue 3 中构建出稳定、流畅的用户界面每一步都需要对细节的精准把控。调试的过程就是与不稳定的网络、非常规的数据分块和框架的响应式机制不断博弈的过程。希望本文拆解的原理、提供的代码范例和总结的避坑经验能帮助你顺利地将“流淌”的 AI 智慧优雅地呈现在你的下一个应用之中。