1. 流式输出的本质为什么我们需要 SSE1.1 从一次糟糕的等待说起如果你做过任何跟大模型对接的前端项目一定遇到过这种场景用户点下发送按钮界面卡住转圈圈转了七八秒然后一大段文字“啪”地一下全部蹦出来。体验非常割裂用户会怀疑是不是网络断了甚至反复点击按钮。这个问题的根源在于传统的 HTTP 请求-响应模型是“一问一答”式的服务端必须把完整的结果算完才能一次性返回给客户端。大模型的推理过程恰恰是逐 token 生成的它天然就是流式的。如果我们硬要等它全部生成完再返回等于白白浪费了中间过程的时间。SSEServer-Sent Events就是解决这个矛盾的关键技术。它允许服务端在一个长连接上持续向客户端推送数据片段客户端收到一片就渲染一片于是就有了我们熟悉的“打字机效果”。这里要先厘清一个概念SSE 不是 WebSocket。很多人一提到实时推送就想到 WebSocket但两者定位完全不同。WebSocket 是双向通信适合聊天室、协同编辑这类需要客户端频繁发消息的场景而 SSE 是单向的服务端推、客户端收正好匹配大模型“我问一次、它答一长串”的模式。而且 SSE 基于普通 HTTP 协议不需要额外的握手升级浏览器原生支持EventSource服务端实现也简单得多。1.2 SSE 的数据格式到底长什么样SSE 的协议格式其实非常朴素它就是一段纯文本用特定的字段名来组织。一个典型的事件流长这样data: {content: 你} data: {content: 好} data: {content: 世界} data: [DONE]每个事件之间用两个换行符分隔。data:后面跟的是实际内容可以是纯文本也可以是 JSON 字符串。除了data还有event自定义事件类型、id事件 ID用于断线重连时定位、retry重连间隔毫秒数这几个字段。实际对接大模型 API 时最常用的就是data字段服务端把每个 token 或一小段文本包成一个事件推过来。注意SSE 规范里以冒号开头的行是注释行常被用作心跳保活。很多服务端会每隔十几秒推一个: keep-alive来防止连接被中间层掐断。这个细节在后面排查“idle timeout”问题时非常关键。1.3 为什么大模型服务偏爱 SSE从工程角度看SSE 有几个天然优势。第一是实现成本低服务端只要设置Content-Type: text/event-stream然后往响应流里写数据就行不需要维护复杂的连接状态。第二是兼容性好它跑在标准 HTTP 之上各种反向代理、负载均衡基本都能透传不像 WebSocket 有时会被某些网关拦截。第三是自动重连浏览器端的EventSource内置了断线重连机制服务端可以通过retry字段控制重连节奏。当然它也有短板。SSE 是单向的客户端没法通过同一个连接发消息另外在 HTTP/1.1 下浏览器对同一域名的并发连接数有限制如果同时开很多 SSE 连接会互相挤占。不过对于大模型对话这种场景通常一个会话一条流问题不大。理解了这些底层特性后面无论是自己封装流式接口还是排查连接中断问题心里都有底。2. 手写一个 SSE 流式接口从服务端到前端2.1 服务端用 FastAPI 把大模型输出“挤”出来现在假设我们要用 FastAPI 搭一个流式接口把大模型的输出逐段推给前端。核心思路是返回一个生成器generatorFastAPI 会用StreamingResponse把它包装成流式响应。下面是一个可以直接跑的骨架from fastapi import FastAPI from fastapi.responses import StreamingResponse import json, asyncio app FastAPI() async def event_generator(prompt: str): # 模拟大模型逐 token 输出 fake_tokens [流, 式, 输, 出, 的, 核, 心, 是, 逐, 段, 返, 回] for token in fake_tokens: payload {content: token} # SSE 格式data: xxx\n\n yield fdata: {json.dumps(payload, ensure_asciiFalse)}\n\n await asyncio.sleep(0.1) # 结束标记 yield data: [DONE]\n\n app.get(/stream) async def stream(prompt: str): return StreamingResponse( event_generator(prompt), media_typetext/event-stream, headers{ Cache-Control: no-cache, Connection: keep-alive, X-Accel-Buffering: no, # 关键禁用 Nginx 缓冲 }, )这段代码里有几个容易被忽略但极其关键的细节。media_type必须是text/event-stream否则浏览器不会按 SSE 解析。Cache-Control: no-cache防止中间层缓存流式内容。而X-Accel-Buffering: no这个响应头是给 Nginx 看的如果不加Nginx 默认会缓冲响应导致前端收到的数据是一坨一坨的打字机效果直接失效。我踩过这个坑本地测试一切正常一上生产就变成“批量蹦字”排查了半天才发现是网关缓冲。2.2 前端EventSource 与 fetch 流式读取的取舍前端接收 SSE 有两条路。第一条是用浏览器原生的EventSourceconst es new EventSource(/stream?prompt你好); es.onmessage (event) { if (event.data [DONE]) { es.close(); return; } const { content } JSON.parse(event.data); appendToUI(content); }; es.onerror (err) { console.error(SSE 连接异常, err); es.close(); };EventSource的优点是简单、自带重连但它有个硬伤只支持 GET 请求不能自定义请求头。这意味着你没法在 Header 里带认证 token也没法发 POST 请求传复杂的参数体。实际项目里对话内容往往很长塞进 URL 查询参数既不优雅也可能超长。所以更通用的做法是用fetch配合ReadableStream手动解析const response await fetch(/stream, { method: POST, headers: { Content-Type: application/json, Authorization: Bearer xxx }, body: JSON.stringify({ prompt: 你好 }), }); const reader response.body.getReader(); const decoder new TextDecoder(utf-8); let buffer ; while (true) { const { done, value } await reader.read(); if (done) break; buffer decoder.decode(value, { stream: true }); // 按双换行切分事件 const parts buffer.split(\n\n); buffer parts.pop(); // 最后一段可能不完整留到下次 for (const part of parts) { if (!part.startsWith(data:)) continue; const data part.slice(5).trim(); if (data [DONE]) return; const { content } JSON.parse(data); appendToUI(content); } }这里有个必须注意的点网络传输是分片的一个 SSE 事件可能被拆到两个 chunk 里。所以不能假设每次reader.read()拿到的都是完整事件必须维护一个 buffer用split(\n\n)切分后把最后一段不完整的留在缓冲区。这个细节如果处理不好会出现 JSON 解析报错而且报错是偶发的非常难查。2.3 打字机效果的渲染节奏控制拿到 token 之后直接往 DOM 里塞其实还不够“像打字机”。因为大模型吐 token 的速度是不均匀的有时候一次来好几个字有时候卡一下。如果直接渲染视觉上会一顿一顿的。我的做法是加一个简单的渲染队列用requestAnimationFrame或者定时器平滑消费const queue []; let rendering false; function enqueue(text) { queue.push(...text.split()); if (!rendering) renderLoop(); } function renderLoop() { if (queue.length 0) { rendering false; return; } rendering true; const char queue.shift(); appendToUI(char); setTimeout(renderLoop, 20); // 每 20ms 渲染一个字 }这样即使后端一次推来五个字前端也会以稳定的节奏逐字显示观感更顺滑。20ms 这个值可以调太快了看不出打字感太慢了会拖慢整体阅读速度。我一般用 15 到 30ms 之间根据内容长度动态调整——短回答快一点长回答慢一点。3. LangChain 结构化输出让模型吐出可解析的 JSON3.1 为什么自由文本不够用流式输出解决了“看得爽”的问题但很多场景下我们要的不只是给人看的文字而是给程序用的数据。比如用户说“帮我订明天下午三点从北京到上海的机票”后端需要提取出{date: 明天, time: 15:00, from: 北京, to: 上海}这样的结构化对象才能去调用订票接口。如果模型返回的是一段自然语言你还得再写正则去抠既脆弱又容易出错。LangChain 的结构化输出Structured Output就是干这个的。它让模型按照你给定的 schema 返回 JSONLangChain 负责把 schema 转换成模型能理解的格式比如 JSON Schema 或函数调用定义再把模型返回的结果解析成 Python 对象。这样一来模型输出就从“散文”变成了“表格”程序可以直接消费。3.2 三种主流实现路径的对比LangChain 里实现结构化输出有好几种方式我按推荐程度排个序并说明各自适用场景。方式原理优点缺点适用场景with_structured_output利用模型原生函数调用能力最稳定解析成功率高依赖模型支持 function calling主流商用模型PydanticOutputParser把 schema 写进 prompt解析返回文本兼容所有模型模型可能不按格式返回开源小模型JsonOutputParser类似上者但更轻量简单直接无类型校验快速原型with_structured_output是首选因为它走的是模型原生的工具调用通道模型在训练时就见过大量这种格式遵循度最高。用法也很直观from langchain_openai import ChatOpenAI from pydantic import BaseModel, Field class FlightInfo(BaseModel): date: str Field(description出发日期) time: str Field(description出发时间24小时制) from_city: str Field(description出发城市) to_city: str Field(description到达城市) llm ChatOpenAI(modelgpt-4o-mini) structured_llm llm.with_structured_output(FlightInfo) result structured_llm.invoke(帮我订明天下午三点从北京到上海的机票) print(result) # FlightInfo(date明天, time15:00, from_city北京, to_city上海)注意Field里的description不是可有可无的装饰。模型就是靠这些描述来理解每个字段该填什么的。描述写得越清楚抽取准确率越高。我见过有人把所有字段描述都写成“这是一个字段”结果模型乱填一气。这就像给新人派活你交代得越具体他干得越对。3.3 结构化输出与流式的矛盾怎么破这里有个天然的矛盾结构化输出要求返回完整的 JSON 才能解析而流式输出是逐段来的JSON 没闭合之前根本没法json.loads。那能不能既要流式又要结构化可以但要用“部分解析”的思路。LangChain 提供了JsonOutputParser的流式版本它能在 JSON 还没闭合时尽可能解析出已经完整的字段。比如模型正在输出{date: 明天, time: 15:00, from_city: 北解析器能先返回{date: 明天, time: 15:00}等后续字段补齐再更新。前端就可以做到“字段逐个填充”的效果比等整个 JSON 出来再渲染体验好很多。from langchain_core.output_parsers import JsonOutputParser parser JsonOutputParser(pydantic_objectFlightInfo) chain prompt | llm | parser for chunk in chain.stream({input: 帮我订明天下午三点从北京到上海的机票}): print(chunk) # 逐步输出已解析的字段不过要提醒一句部分解析对 JSON 格式的容错要求很高如果模型输出的 JSON 有语法错误比如多了个逗号解析器可能中途抛异常。生产环境里我一般会加一层 try-except解析失败就跳过这个 chunk等下一个完整的再试。4. 实战踩坑那些文档里不会写的细节4.1 “stream disconnected before completion” 到底怎么回事这个报错在对接大模型流式接口时出现频率极高字面意思是“流在完成前断开了”。它背后的原因通常有三类。第一类是空闲超时也就是标题里提到的idle timeout waiting for sse。很多网关或负载均衡器有个默认规则如果一条连接超过 N 秒没有数据传输就判定为空闲并掐断。大模型在“思考”阶段尤其是推理模型可能十几秒不吐一个字正好触发这个规则。解决办法是在服务端加心跳。前面提到 SSE 支持注释行我们可以起一个后台任务每隔 10 秒往流里写一个: ping\n\nasync def event_generator(prompt: str): heartbeat_task asyncio.create_task(send_heartbeat()) try: async for token in llm.astream(prompt): yield fdata: {json.dumps({content: token})}\n\n finally: heartbeat_task.cancel() yield data: [DONE]\n\n async def send_heartbeat(): while True: await asyncio.sleep(10) yield : ping\n\n # 注意这里需要配合队列才能真正推出去实际实现时心跳和业务数据往往要共用一个队列用一个生产者-消费者模型来协调否则两个协程同时往响应流里写会出问题。这个坑我在第一次做流式接口时踩得很惨心跳和 token 交错写入导致前端解析出一堆乱码。第二类原因是代理层缓冲。有些反向代理会等缓冲区满了才转发如果流很小又很慢可能一直不转发最后超时。前面提到的X-Accel-Buffering: no就是治这个的。第三类是客户端主动断开比如用户切走了页面浏览器关闭了连接服务端写入时就会报错。这种情况属于正常现象捕获异常静默处理即可。4.2 JSON 解析的容错处理清单模型返回的 JSON 再规范也架不住偶尔抽风。下面这些容错手段是我在实际项目里逐步攒出来的按优先级排列。去除 markdown 代码块标记模型经常把 JSON 包在json ...里解析前先 strip 掉。处理尾随逗号{a: 1,}这种在 Python 里是非法的可以用正则或json5库容错。截断修复流式场景下 JSON 可能不完整需要补全缺失的引号和括号。字段缺失兜底用 Pydantic 的默认值或 Optional 类型避免某个字段没抽到就整个失败。重试机制解析失败时把原始输出和错误信息再喂给模型让它修正格式通常一次就能修好。提示不要试图用正则去解析 JSON。JSON 是递归结构正则处理不了嵌套。要么用json.loads要么用专门的容错库别自己造轮子。4.3 流式接口的封装复用思路一个项目里往往有多个地方要调流式接口如果每个地方都写一遍 fetch 解析逻辑维护起来是灾难。我的做法是封装一个通用的流式客户端把连接管理、事件解析、错误重试、心跳检测都收进去业务层只关心“收到一个 token 该干嘛”。class SSEClient { constructor(url, options {}) { this.url url; this.options options; this.controller null; } async connect(body, { onMessage, onError, onComplete }) { this.controller new AbortController(); try { const resp await fetch(this.url, { method: POST, headers: { Content-Type: application/json, ...this.options.headers }, body: JSON.stringify(body), signal: this.controller.signal, }); const reader resp.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.split(\n\n); buffer events.pop(); for (const evt of events) { if (evt.startsWith(:)) continue; // 心跳 if (!evt.startsWith(data:)) continue; const data evt.slice(5).trim(); if (data [DONE]) { onComplete?.(); return; } onMessage?.(JSON.parse(data)); } } } catch (err) { if (err.name ! AbortError) onError?.(err); } } abort() { this.controller?.abort(); } }这个类把心跳行过滤、buffer 切分、中断控制都处理了。业务层调用时只需要传onMessage回调非常清爽。而且abort方法让用户可以随时点“停止生成”这在对话产品里是标配功能。5. 从单轮到 Agent流式与结构化的进阶组合5.1 Agent 场景下的流式事件类型当你从简单的问答升级到 Agent智能体流式输出的内容就不再只是文本 token 了。Agent 可能会先“思考”再决定调用某个工具工具返回结果后再继续生成。这整个过程如果都流式推给前端事件类型就变得多样有思考片段、有工具调用请求、有工具执行结果、有最终回答。LangChain 和 LangGraph 在这方面提供了事件流接口可以按事件类型分别处理。前端收到tool_call事件时可以显示“正在查询天气...”收到tool_result时显示结果收到token时继续打字机渲染。这种分类型的流式处理能让用户清楚看到 Agent 在干什么而不是干等。async for event in agent.astream_events({input: query}, versionv2): kind event[event] if kind on_chat_model_stream: yield sse_pack(token, event[data][chunk].content) elif kind on_tool_start: yield sse_pack(tool_start, {name: event[name]}) elif kind on_tool_end: yield sse_pack(tool_end, {output: event[data][output]})这里的关键是统一事件包装格式前端按type字段分发处理。我一般用{type: token, data: {...}}这样的结构简单清晰。5.2 结构化输出在 Agent 工具调用中的角色Agent 调用工具时参数必须是结构化的。比如“查天气”工具需要{city: 北京, date: 今天}模型得把用户那句“北京今天天气咋样”转成这个结构。这正好是结构化输出的用武之地。LangChain 的工具定义本身就带 schema模型通过 function calling 生成参数框架自动解析后传给工具函数。实际开发中工具参数抽取的准确率是 Agent 好不好用的分水岭。我的经验是工具描述和参数描述要写得像给同事交代任务一样具体。比如date字段不要只写“日期”要写“日期格式 YYYY-MM-DD如果用户说‘今天’则填当天日期”。模型看到这种描述抽取准确率能明显提升。5.3 端到端链路的性能考量把 SSE 流式、LangChain 结构化输出、Agent 工具调用串起来之后链路的延迟点会变多。我实测下来主要的耗时分布在三块模型首 token 延迟、工具执行时间、以及各层之间的序列化开销。首 token 延迟主要看模型和网络优化空间有限工具执行时间取决于外部 API可以加缓存序列化开销则可以通过精简事件结构来降低。有个容易被忽视的点是背压backpressure。如果前端渲染速度跟不上后端推送速度数据会在缓冲区堆积内存上涨。虽然大模型输出速度通常不快但在 Agent 场景下工具结果可能一次性推来一大段这时候前端要有节流机制。我的做法是在渲染队列里设一个上限超过就丢弃中间帧只保留最新的保证界面不卡死。6. 我个人的一些实操体会流式接口这东西看起来简单真正做稳需要处理的边界情况比想象中多。我最初以为只要把StreamingResponse一包就完事了结果上线后遇到各种断流、乱码、卡顿。后来慢慢总结出一套自己的检查清单上线前必测弱网环境、必测长文本、必测用户中途取消、必测网关超时配置。这四项过了基本就不会出大问题。结构化输出方面我的建议是能用原生 function calling 就别用 prompt 解析。前者是模型的原生能力稳定性和准确率都高一个档次。只有在用不支持工具调用的开源模型时才退而求其次用 prompt 方案并且一定要加解析失败的重试兜底。最后分享一个小技巧调试流式接口时用curl -N命令直接看原始输出比在浏览器里看 Network 面板清楚得多。-N参数禁用 curl 自己的缓冲能实时看到每个 SSE 事件到达的时机排查心跳和超时问题特别有用。这个命令帮我定位过好几次“看起来是前端问题、实际是网关缓冲”的疑难杂症。
