1. 项目概述为什么流式响应是 Coding Agent 的“呼吸”如果你正在构建一个 Coding Agent或者任何需要与大型语言模型LLM进行长时间、多轮次交互的 AI 应用那么“流式响应”绝对不是你锦上添花的功能而是决定用户体验成败的“呼吸系统”。想象一下你向一个助手提问它需要思考一分钟然后才把完整的答案一股脑地“吐”给你。在这漫长的等待中你无法判断它是卡住了、出错了还是在认真工作。这种体验无疑是糟糕的尤其是在代码生成、调试、解释等场景下用户需要实时看到模型的“思考”过程以便及时引导或纠正。这就是“流式响应”要解决的核心痛点。它允许服务器将 LLM 生成的内容像水流一样以数据块chunk的形式持续推送给客户端而不是等待整个内容生成完毕再一次性返回。对于 Coding Agent 而言这意味着实时反馈用户能立即看到模型生成的第一个单词、第一行代码感知到 Agent 正在工作降低等待焦虑。动态交互在代码生成过程中如果用户发现方向不对可以随时中断而不是等一个可能完全错误的冗长结果。性能感知流式传输能更早地开始渲染内容从用户感知上大幅缩短响应时间提升应用响应速度。本篇文章我们将深入探讨如何为你的 Coding Agent 实现稳定、高效的流式响应。我们将超越简单的“Hello World”示例聚焦于生产环境中必须考虑的细节如何选择协议、如何处理网络中断、如何设计前后端协作、以及如何优化用户体验。无论你使用的是 OpenAI 的 GPT 系列、开源的 Llama 或 Qwen还是其他任何提供流式接口的模型这里讨论的原理和模式都是相通的。2. 技术选型SSE 与 WebSocket 的深度对比实现流式通信主流有两种技术Server-Sent Events (SSE)和WebSocket。很多文章会简单地告诉你“用 SSE 就够了”但作为开发者我们必须理解背后的“为什么”才能做出最适合自己场景的选择。2.1 SSE为服务器到客户端的单向流而生SSE 是一种基于 HTTP 的轻量级协议。它的核心思想是客户端发起一个普通的 HTTP 请求但服务器不立即关闭连接而是保持连接打开并持续发送一系列格式化的“事件”数据。工作原理简述客户端浏览器使用EventSourceAPI 向一个特定端点发起 GET 请求并在请求头中设置Accept: text/event-stream。服务器响应时设置Content-Type: text/event-stream并保持连接不关闭。服务器通过这个持久的连接持续发送遵循特定格式的数据块。每个数据块以data:开头以两个换行符\n\n结束。例如data: 这是第一段流式内容\n\n data: 这是第二段内容\n\n客户端EventSource会自动解析这些消息并触发onmessage事件。SSE 的优势与局限优势协议简单基于 HTTP/HTTPS无需额外的协议升级握手兼容性极佳几乎不受防火墙或代理限制。自动重连EventSource内置了连接断开后的自动重连机制对于不稳定的网络环境非常友好。轻量级协议开销小特别适合服务器向客户端推送文本信息的场景如新闻推送、股票行情、以及我们这里的 LLM 文本流。局限单向通信只能从服务器向客户端推送数据。如果 Coding Agent 需要在生成过程中接收用户的实时干预指令例如“停重写这个函数”SSE 本身无法支持。通常需要配合另一个独立的 HTTP 请求通道来实现。文本协议虽然可以传输 JSON 字符串但原生设计是针对文本的。传输二进制数据如音频、视频流不是它的强项。连接数限制浏览器对同一个域名下的 HTTP 连接数有上限通常为6个一个持久的 SSE 连接会占用其中一个。对于需要大量并发长连接的场景需要谨慎。2.2 WebSocket全双工实时通信的瑞士军刀WebSocket 提供了在单个 TCP 连接上进行全双工通信的能力。连接建立后客户端和服务器可以随时相互发送数据。工作原理简述客户端发起一个带有Upgrade: websocket头的 HTTP 请求进行协议升级握手。握手成功后连接从 HTTP 协议切换为 WebSocket 协议。此后双方可以通过send方法发送数据并通过onmessage事件接收数据数据格式可以是文本或二进制。WebSocket 的优势与局限优势全双工双向实时通信是它的核心优势。对于需要复杂交互的 Coding Agent例如用户一边看代码生成一边可以发送“解释这行”、“重构变量名”等指令WebSocket 是更自然的选择。低延迟建立连接后数据传输头部开销极小延迟非常低。二进制支持原生支持二进制帧适合传输多种类型的数据。局限协议更复杂需要处理握手、帧解析、心跳保活等实现复杂度高于 SSE。无自动重连连接断开后需要开发者自己实现重连逻辑。可能遇到代理问题某些古老的或配置严格的代理服务器可能不支持 WebSocket 协议升级。2.3 决策指南为你的 Coding Agent 选择什么特性SSE (Server-Sent Events)WebSocket通信方向单向 (服务器 - 客户端)全双工 (双向)协议基础HTTP/HTTPS独立的 WebSocket 协议 (基于 TCP)实现复杂度低 (客户端使用EventSource服务器端格式简单)中高 (需处理握手、帧、心跳)自动重连内置支持需手动实现数据格式文本 (通常为text/event-stream)文本或二进制适用场景通知、日志流、LLM 文本流式输出聊天应用、实时协作、需要双向交互的复杂 Agent防火墙友好度高 (就是 HTTP)中 (可能被特殊策略拦截)我的经验与建议 对于大多数初、中阶的 Coding Agent 项目我强烈建议从 SSE 开始。原因如下场景匹配LLM 生成代码的过程本质上是服务器将生成的 token 流式推送给客户端这是一个典型的单向推送场景。SSE 就是为此而生的。简单可靠基于 HTTP意味着你可以复用现有的认证、负载均衡、监控体系。EventSource的自动重连能省去大量边缘情况处理代码。快速上手你可以在一个下午就搭建出可用的流式响应原型把精力集中在 Prompt 工程和 Agent 逻辑上而不是通信协议上。当你需要实现以下功能时才需要考虑升级到 WebSocket实时双向对话用户可以在 Agent 生成代码时随时插入评论或指令。多模态流需要同时流式传输代码和生成的图表、解释音频等。超高并发与低延迟对延迟有极致要求且能驾驭 WebSocket 的复杂性和状态管理。在本文的后续实现中我们将以SSE作为主要技术栈进行详解因为它是最贴合“让 LLM 流式响应”这一核心需求的、性价比最高的方案。3. 后端实现构建健壮的流式 API 端点后端是流式响应的发动机。我们的目标是构建一个能够稳定、高效地从 LLM 获取流式数据并将其规范化为 SSE 格式推送给前端的服务。这里以 Python 的 FastAPI 框架和 OpenAI 兼容的 API 为例但原理适用于任何语言和模型。3.1 依赖安装与基础设置首先确保你的环境已安装必要的库。我们将使用openai库或兼容的客户端如litellm来调用模型使用sse-starlette或sse_starlette来方便地生成 SSE 响应。pip install fastapi uvicorn openai sse-starlette3.2 核心 API 端点实现我们将创建一个/stream的 POST 端点它接收用户的请求如代码生成任务描述然后流式返回模型的响应。from fastapi import FastAPI, Request from fastapi.responses import StreamingResponse import openai import asyncio import json from sse_starlette.sse import EventSourceResponse app FastAPI() # 配置你的 LLM 客户端这里以 OpenAI 格式为例 # 实际可能是 OpenAI, Azure OpenAI, 或本地部署的 vLLM 等服务 client openai.AsyncOpenAI( api_keyyour-api-key, base_urlhttps://api.openai.com/v1 # 或你的本地/第三方端点 ) async def generate_streaming_response(prompt: str, model: str gpt-4): 核心生成器函数从 LLM 获取流式响应并转换为 SSE 格式。 try: # 调用 LLM 的流式接口 stream await client.chat.completions.create( modelmodel, messages[{role: user, content: prompt}], streamTrue, # 关键参数启用流式输出 temperature0.7, max_tokens2000, ) # 异步迭代流式响应 async for chunk in stream: # 提取 delta 中的内容 content chunk.choices[0].delta.content if content is not None: # 将内容封装为 SSE 事件数据 # 我们发送一个 JSON 字符串包含内容和可能的元数据如是否结束 event_data json.dumps({ content: content, finished: False }) yield event_data # 流结束时发送一个结束标记事件 yield json.dumps({content: , finished: True}) except Exception as e: # 发生错误时发送错误信息并标记结束 error_data json.dumps({ content: f\n\n[流式生成发生错误: {str(e)}], finished: True, error: True }) yield error_data app.post(/api/stream-code) async def stream_code(request: Request): 流式代码生成端点。 期望的请求体 JSON: {prompt: 用户的任务描述, model: 可选模型名称} data await request.json() prompt data.get(prompt, ) model data.get(model, gpt-4) if not prompt: # 对于错误请求也可以返回一个立即结束的流 async def error_stream(): yield json.dumps({content: 错误请求中未提供有效的 prompt。, finished: True, error: True}) return EventSourceResponse(error_stream()) # 使用 EventSourceResponse 包装我们的生成器 # 它会自动设置正确的 Content-Type: text/event-stream return EventSourceResponse( generate_streaming_response(prompt, model), ping15000 # 每15秒发送一个“ping”注释以保持连接活跃防止超时 )代码关键点解析streamTrue这是调用 LLM API 时开启流式响应的关键参数。没有它你会一直等待整个响应完成。异步生成器 (async for)我们使用async for来异步地消费 LLM 返回的流。这对于处理高并发请求至关重要它不会阻塞事件循环。SSE 数据格式我们每次yield一个 JSON 字符串。前端需要解析这个 JSON 来获取content和状态标记finished,error。这种结构比只发送纯文本更灵活便于扩展例如未来加入思考过程、工具调用等元数据。EventSourceResponse来自sse-starlette它帮我们处理了 SSE 协议细节包括自动添加data:前缀和双换行符以及可选的ping机制来保持连接。错误处理在try...except中包裹核心逻辑确保即使 LLM API 调用失败也能向客户端发送一个有意义的错误消息并正常结束流而不是直接断开连接导致前端无法感知。3.3 生产环境增强考虑上面的代码是一个可运行的原型但要用于生产还需要考虑以下几点超时与中断处理用户可能中途关闭页面或取消请求。后端需要监听连接断开事件并立即终止昂贵的 LLM 生成过程以节省资源。在 FastAPI 中你可以检查request.is_disconnected()在生成器内部定期检查比较麻烦通常需要结合背景任务和取消令牌。速率限制与缓存为 SSE 连接实现速率限制防止滥用。对于相同的提示可以考虑缓存首个 chunk 或完整响应但要注意流式场景下缓存的新鲜度。结构化数据与工具调用如果你的 Coding Agent 使用 ReAct 模式或 Function Calling流式响应可能需要传输更复杂的结构化数据如[THOUGHT]...[/THOUGHT]而不仅仅是纯文本。需要在数据协议设计时预留字段。使用更通用的客户端考虑使用litellm这样的库它统一了 OpenAI、Anthropic、Cohere、本地模型等众多接口让你的后端代码更容易切换模型提供商。4. 前端实现优雅地消费与展示流式内容前端的目标是创建一个流畅、用户友好的界面实时接收并渲染从后端 SSE 端点推送来的代码片段。我们将使用现代 JavaScript或 TypeScript和EventSourceAPI。4.1 基础 EventSource 连接首先我们创建一个函数来建立 SSE 连接并处理数据。class StreamingCodeAgent { constructor(apiEndpoint /api/stream-code) { this.apiEndpoint apiEndpoint; this.eventSource null; this.onDataCallback null; this.onFinishCallback null; this.onErrorCallback null; } /** * 开始流式代码生成 * param {string} prompt - 用户输入的提示词 * param {string} model - 选择的模型 */ startStreaming(prompt, model gpt-4) { // 先关闭可能存在的旧连接 this.close(); // 构建请求体 const requestBody { prompt, model }; // 使用 URLSearchParams 或直接发送 JSON注意后端接收方式 // 这里我们使用 POST 并发送 JSON但 EventSource 原生只支持 GET。 // 因此需要一个变通方案 // 方案A: 后端改为 GET参数放查询字符串有长度限制。 // 方案B: 使用 Fetch API 的流式响应放弃 EventSource。 // 方案C: 先 POST 创建一个会话返回一个带 token 的 GET 流式端点。 // 这里演示方案B因为它更灵活且支持POST。 this.useFetchStreaming(prompt, model); } /** * 使用 Fetch API 实现流式读取推荐支持POST */ async useFetchStreaming(prompt, model) { try { const response await fetch(this.apiEndpoint, { method: POST, headers: { Content-Type: application/json, }, body: JSON.stringify({ prompt, model }), }); if (!response.ok || !response.body) { throw new Error(HTTP error! status: ${response.status}); } // 获取可读流 const reader response.body.getReader(); const decoder new TextDecoder(utf-8); let buffer ; while (true) { const { done, value } await reader.read(); if (done) { // 流完全结束 if (this.onFinishCallback) this.onFinishCallback(); break; } // 解码 chunk 并添加到缓冲区 buffer decoder.decode(value, { stream: true }); // 解析缓冲区中的完整 SSE 行以 \n\n 分隔 const lines buffer.split(\n\n); // 最后一行可能是不完整的保留在缓冲区 buffer lines.pop() || ; for (const line of lines) { if (line.startsWith(data: )) { const dataStr line.slice(6).trim(); // 去掉 data: if (dataStr) { try { const data JSON.parse(dataStr); // 调用回调函数处理数据 if (data.error this.onErrorCallback) { this.onErrorCallback(data.content); } else if (this.onDataCallback) { this.onDataCallback(data.content, data.finished); } // 如果收到结束信号可以跳出循环但让外层循环自然结束更安全 if (data.finished) { reader.cancel(); // 可选提前取消读取 if (this.onFinishCallback) this.onFinishCallback(); return; } } catch (e) { console.error(Failed to parse SSE data:, e, Raw:, dataStr); } } } // 忽略 event:, id:, retry: 等其他 SSE 字段或处理 ping 注释 } } } catch (error) { console.error(Streaming failed:, error); if (this.onErrorCallback) this.onErrorCallback(连接失败: ${error.message}); } } // 注册回调函数 onData(callback) { this.onDataCallback callback; return this; // 支持链式调用 } onFinish(callback) { this.onFinishCallback callback; return this; } onError(callback) { this.onErrorCallback callback; return this; } // 关闭连接 close() { if (this.eventSource) { this.eventSource.close(); this.eventSource null; } } }为什么不用原生的EventSource原生EventSource不支持 POST 请求和自定义请求头这在传递较长的 prompt 或需要认证时很不方便。上面的useFetchStreaming方法使用 Fetch API 手动处理 SSE 流虽然代码量稍多但获得了完全的灵活性。4.2 与 UI 框架集成以 React 为例现在我们将这个流式客户端集成到一个 React 组件中。import React, { useState, useRef, useEffect } from react; function CodeGenerationView() { const [inputPrompt, setInputPrompt] useState(请用Python写一个快速排序函数并添加详细注释。); const [isStreaming, setIsStreaming] useState(false); const [generatedCode, setGeneratedCode] useState(); const [statusMessage, setStatusMessage] useState(); const streamAgentRef useRef(null); useEffect(() { // 初始化 Agent 实例 streamAgentRef.current new StreamingCodeAgent(http://your-backend.com/api/stream-code); // 设置回调 streamAgentRef.current .onData((chunk, isFinished) { // 收到数据块追加到生成的代码中 setGeneratedCode(prev prev chunk); if (isFinished) { setIsStreaming(false); setStatusMessage(生成完成); } }) .onFinish(() { setIsStreaming(false); setStatusMessage(流已结束。); }) .onError((errorMsg) { setIsStreaming(false); setStatusMessage(错误: ${errorMsg}); setGeneratedCode(prev prev \n\n--- 生成中断: ${errorMsg} ---); }); // 组件卸载时清理 return () { if (streamAgentRef.current) { streamAgentRef.current.close(); } }; }, []); const handleGenerate () { if (!inputPrompt.trim() || isStreaming) return; setGeneratedCode(); // 清空之前的内容 setStatusMessage(正在生成代码...); setIsStreaming(true); // 开始流式生成 streamAgentRef.current.startStreaming(inputPrompt); }; const handleStop () { if (streamAgentRef.current) { streamAgentRef.current.close(); setIsStreaming(false); setStatusMessage(已手动停止。); } }; return ( div classNamecode-gen-container h2智能代码生成助手/h2 div classNameinput-area textarea value{inputPrompt} onChange{(e) setInputPrompt(e.target.value)} placeholder描述你想要的代码功能... rows4 disabled{isStreaming} / div classNamebutton-group button onClick{handleGenerate} disabled{isStreaming || !inputPrompt.trim()} {isStreaming ? 生成中... : 开始生成} /button {isStreaming ( button onClick{handleStop} classNamestop-button 停止生成 /button )} /div div classNamestatus{statusMessage}/div /div div classNameoutput-area label生成的代码/label pre classNamecode-block code{generatedCode || // 代码将在这里实时显示...}/code /pre {/* 可以在这里添加一个代码编辑器组件如 Monaco Editor来获得高亮和编辑功能 */} /div /div ); } export default CodeGenerationView;4.3 用户体验优化技巧打字机效果与其一次性追加整个 chunk可以模拟打字效果逐个字符添加到 DOM。但这会增加前端复杂度对于长代码可能影响性能。一个折中方案是“行级”或“词级”流式渲染。自动滚动当代码不断追加时自动将滚动条保持在底部让用户始终看到最新内容。可以在onData回调中操作 DOM 元素的scrollTop属性。语法高亮流式过程中实时进行语法高亮是个挑战因为代码不完整。可以使用能处理不完整语法的前端高亮库如 Shiki、Prism.js 的某些插件。在高亮前对不完整的行或语法结构进行简单清理或占位。更简单的做法等流式结束后再进行一次完整的高亮。加载指示器在代码开始生成前可以显示一个闪烁的光标或“正在思考...”的动画给予用户即时反馈。错误状态与重试当流式中断或出错时除了显示错误信息还应提供一个“重试”按钮让用户可以重新发送请求而不是重新输入所有内容。5. 高级话题与生产环境挑战将流式响应投入生产环境意味着要面对更复杂的网络环境、更高的性能要求和更严苛的稳定性需求。5.1 连接稳定性与重连策略网络是不稳定的。SSE 连接可能因网络波动、代理超时、服务器重启而中断。后端保活Ping我们在后端代码中设置了ping15000这会在连接空闲时定期发送冒号开头的注释行:\n\n以保持 TCP 连接活跃防止被中间设备如负载均衡器、代理服务器因超时而关闭。这个时间间隔需要根据你的基础设施调整。前端自动重连如果使用原生EventSource它内置了重连逻辑。如果使用我们基于 Fetch 的实现需要自己实现。一个简单的策略是在onError或连接异常关闭后等待一个指数退避的时间如 1秒2秒4秒...然后重新调用startStreaming。注意对于用户主动停止的情况不应触发自动重连。5.2 流式传输中的上下文管理对于多轮对话的 Coding Agent你需要维护对话历史。在流式场景下这带来两个问题本次流式响应中引用历史这通常由后端处理。你需要将完整的对话历史包括本次用户提问作为 messages 列表发送给 LLM API。将本次流式结果加入历史供下一轮使用前端需要在流式完全结束后将最终生成的完整内容generatedCode发送回后端由后端存储到对话会话中。切勿在流式过程中每收到一个 chunk 就更新服务器端的历史记录这会产生大量无效请求。5.3 性能优化从 Token 到屏幕Chunk 合并与节流LLM API 可能以极快的速度返回 token每个 token 可能就是一个字符。如果每收到一个 token 就更新一次 React 状态setGeneratedCode会导致界面频繁重绘性能低下。一个常见的优化是“缓冲”在前端累积一小段时间如 50-100ms或一定数量字符如 20个字符的 chunk然后再批量更新状态。这能显著提升渲染性能。使用 Web Worker对于非常密集的流式更新和语法高亮计算可以考虑将流式数据的处理和组装放到 Web Worker 中避免阻塞主线程保持 UI 响应流畅。后端响应优化确保你的后端服务器如 Uvicorn/Gunicorn配置了合适的异步工作模式和超时设置能够高效处理大量并发的长连接SSE 连接。5.4 安全与监控认证与授权SSE 端点也需要保护。你可以在请求头中携带 Token如 JWT。对于基于 Fetch 的实现这很容易。对于希望用原生EventSource的情况可能需要通过 URL 查询参数传递 Token注意安全风险或者先通过一个普通 API 认证获取一个短期有效的、专用于 SSE 连接的令牌。速率限制根据用户 ID 或 IP 对/stream端点进行速率限制防止恶意用户耗尽你的 LLM API 额度或服务器资源。监控与日志记录流式请求的开始、结束、持续时间、消耗的 token 数。这对于成本核算、性能分析和故障排查至关重要。注意日志本身不应阻塞流式响应。6. 常见问题排查与实战技巧在实际开发中你一定会遇到各种“坑”。这里记录了一些典型问题及其解决方案。6.1 连接立即关闭或收不到数据检查响应头后端必须设置Content-Type: text/event-stream。如果使用了StreamingResponse或EventSourceResponse它们通常会帮你设置。检查 CORS如果前端与后端域名不同后端必须正确配置 CORS允许前端域名的请求并暴露必要的头如Content-Type。检查网络代理和防火墙某些企业网络环境可能会拦截或篡改长连接。尝试在简单网络环境下测试。确保你的负载均衡器如 Nginx配置了合适的超时时间例如proxy_read_timeout 300s;。查看浏览器开发者工具在 Network 标签页查看对你的 SSE 端点的请求。状态应该是 “Pending” 或 “200”并且类型是 “eventsource”。点击请求在 “Response” 或 “EventStream” 标签页查看是否有数据流进来。6.2 数据流中断或不完整后端生成器提前退出确保你的后端异步生成器函数 (generate_streaming_response) 中没有未捕获的异常并且yield语句被执行。在finally块中打印日志有助于调试。前端缓冲区解析错误我们手动解析 SSE 格式时如果数据块 (chunk) 的边界恰好切在\n\n中间会导致解析错误。代码中的buffer机制就是为了处理这种不完整帧。确保你的解析逻辑足够健壮。Nginx/Apache 缓冲反向代理服务器默认可能会缓冲上游你的后端应用的响应直到达到一定大小或超时后才发送给客户端。这会导致流式响应失去“实时性”。你需要在代理配置中禁用缓冲# Nginx 配置示例 location /api/stream-code { proxy_pass http://your_backend; proxy_set_header Connection ; proxy_http_version 1.1; chunked_transfer_encoding off; # 对于 SSE有时需要关闭分块编码 proxy_buffering off; # 关键关闭代理缓冲 proxy_cache off; # 关闭缓存 proxy_read_timeout 300s; # 设置长的读取超时 }6.3 流式内容格式混乱LLM 输出格式控制LLM 可能在流式输出中插入 Markdown 代码块标记如python。如果你的前端只是简单显示文本这些标记会显得很乱。你需要在 Prompt 中明确要求模型“直接输出纯代码不要包含任何 Markdown 标记”或者在前后端对输出进行后处理过滤掉这些标记。处理换行符LLM 输出的换行符 (\n) 在 HTML 中默认不显示为换行。你需要用 CSSwhite-space: pre-wrap;或white-space: pre;来保留格式或者将\n替换为br /。6.4 内存与资源泄漏及时关闭连接在 React 组件卸载、用户离开页面或主动停止时务必调用agent.close()或reader.cancel()来主动关闭连接和释放资源。后端连接管理对于每个活跃的 SSE 连接后端都会保持一个协程或线程。确保在连接断开时通过捕获GeneratorExit异常或检查客户端断开及时清理资源并终止 LLM 的生成调用如果 API 支持的话。一个关键的实操心得在开发初期不要过度优化。先让最基本的流式功能跑通后端能 yield 数据前端能收到并显示。然后逐步添加错误处理、连接管理、UI 优化。分阶段迭代会让你更容易定位问题。流式响应的核心价值在于“实时性”和“可中断性”只要实现了这两点你的 Coding Agent 体验就已经上了一个大台阶。