尧图网站建设 
做 AI 应用的朋友应该都有同感页面上的回答要像人打字一样逐字蹦出来后端要把大模型的 token 流持续推给前端同时前端最终还要拿到一个能被下游直接消费的 JSON 对象而不是一段夹杂着闲聊、Markdown 符号和多余解释的文本。这背后其实是三个问题绑在一起解决SSE 流式传输怎么做、LangChain 怎么接入流式模型、以及怎么把模型的文本输出稳定解析成结构化 JSON。这篇文章我把整套链路从头拆到尾后端怎么发包、前端怎么收包、LangChain 怎么约束输出、JSON 怎么兜底修复全都过一遍并附上我能直接复现的最小工程代码。适合看这篇内容的人是那些正在做 ChatBot、AI Agent、智能客服或任何要展示流式回答又要有结构化数据的开发者。如果你已经知道 SSE 是 Server-Sent Events但不太清楚它在 LangChain 场景里怎么落地又或者你遇到过模型输出的 JSON 总是带着 json 代码块、字段类型不对、流式传输一半连接被掐断这类问题那这篇文章可以让你少走一大截弯路。1. 先别急着写代码把 SSE 流式协议吃透1.1 一次 HTTP 请求响应体被切成了一串消息SSE 全称 Server-Sent Events中文常叫服务器推送事件。它没有引入新的协议底层依然是大家熟悉的 HTTP。普通请求里服务端把完整响应体一次性写完然后关掉连接SSE 请求则不同服务端在响应体里不断写入文本块连接保持打开客户端每收到一个块就可以立刻处理。这就是为什么它能做打字机效果的技术基础服务端每生成一个 token 就扔给前端前端收到一个 token 渲染一个 token。SSE 对响应格式有明确规定。服务端必须设置Content-Type: text/event-stream并且消息体遵循事件流语法。一个典型的事件流长这样id: 1 event: token data: {delta:你} id: 2 event: token data: {delta:好} id: 3 event: done data: {finished:true}这里有几个关键点。第一每个事件由两个连续换行符\n\n分隔也就是说空行是事件之间的边界。第二data:表示数据内容多行data:会被客户端自动合并成一行所以有些人会在每条 message 里故意把 JSON 折成多行但规范上到最后还是会拼在一起。第三event:用来声明事件类型客户端可以用addEventListener(token, ...)监听具体类型如果不写event:默认是message事件。第四id:表示事件编号客户端会记住最后一个收到的 id断线重连时把它放到Last-Event-ID请求头里发给服务端服务端借此实现增量续传。很多后端开发第一次写 SSE 容易忽略的细节是事件流中如果有多余的空格或者没有空行作为分隔客户端解析就会错位。前端浏览器里的EventSource解析器是严格按空行切事件的服务端该\n\n的地方一个都不能省。这个问题在 nginx 缓冲、代理转发、语言框架自动换行等场景里很容易被放大所以协议细节必须先搞清楚不然后面踩坑时不知道从哪查起。1.2 为什么这类场景选 SSE 而不是 WebSocket每次讲到流式传输总有人问为什么不用 WebSocket我的答案是业务场景决定选型大模型流式回答是一个单向推送、几乎不需要客户端反向实时发消息的场景SSE 比 WebSocket 有更明显的性价比优势。可以用一个表快速对比对比项SSEWebSocket底层协议HTTP兼容成本低独立 ws/wss 协议需要升级握手机制方向仅服务端到客户端单向推送全双工双向通信断线重连浏览器原生支持自动重连需要自己实现心跳和重连逻辑消息格式文本为主按事件流解析文本或二进制帧服务端实现难度普通 HTTP 响应保持不关闭即可需要专门处理握手、帧解析、连接状态调试方便度浏览器、curl 都可以直接看需要专用工具或代码辅助大模型流式接口的诉求很典型用户问了一句话之后大量数据是从模型往用户端单向流动。SSE 正好不用做双向通道客户端断开时服务端在写操作中能感知到BrokenPipe之类的异常也足够处理连接生命周期。反过来WebSocket 的全双工能力在这种场景里用不上反而多出心跳、消息帧格式、代理配置等一堆额外复杂度。还有一点SSE 基于 HTTP意味着它可以借用现有的请求头做鉴权。比如把 token 放在Authorization头里传给服务端这在 WebSocket 协议里反而要绕一圈。虽然浏览器原生EventSource不支持自定义请求头但后面我会讲到用fetch加ReadableStream完全可以绕开这个限制同时保留 SSE 协议本身的优点。1.3 合作之前先把事件格式设计好很多团队写 SSE 时候只往data:里塞一段纯文本以为能跑就行。短期的确能跑但业务一复杂就难受。比如你们后面要在原型之外加个追问 token 数量、附带工具调用 ID、返回最终结构化结果等字段纯文本格式就得折腾一遍协议。我的经验是让data:里的内容从第一天起就是 JSON 字符串并且明确约定事件类型。给一种我实际用下来比较舒服的通知格式模型每个增量 token 发一个event: token事件data是一个 JSON 对象里面包含delta新生成的文本片段当模型输出被完整消费完再发一个event: done事件data里带最终完整结果比如解析好的 JSON 对象或统计信息如果中间出错发一个event: error事件data里带code和message。这样设计的好处是前端可以用addEventListener(token, handler)分别监听业务逻辑不会把正常文本和错误信息混在一起同时data用 JSON 而不是裸字符串将来要扩展事件类型、协议版本、附加元数据时不用改动消息的分隔方式只需要在 JSON 里加字段。2. 后端怎么把大模型的 Token 变成 SSE 数据流2.1 FastAPI 实现最小 SSE 接口如果你在用 Python 做后端FastAPI 是这几天我见过的最适合写 SSE 的框架之一。它本身基于 Starlette自带StreamingResponse不需要额外引第三方库就能实现流式响应。先写一个不依赖 LangChain 的最小版本用生成器模拟事件流import asyncio import json from fastapi import FastAPI from fastapi.responses import StreamingResponse app FastAPI() async def event_generator(): for i, chunk in enumerate([你, 好, , 世, 界]): event { event: token, data: json.dumps({delta: chunk}, ensure_asciiFalse) } yield fevent: {event[event]}\ndata: {event[data]}\n\n await asyncio.sleep(0.05) yield event: done\ndata: {finished:true}\n\n app.get(/sse) async def sse_endpoint(): return StreamingResponse( event_generator(), media_typetext/event-stream, headers{ Cache-Control: no-cache, Connection: keep-alive, X-Accel-Buffering: no, }, )注意几个容易踩的坑。yield的内容必须是完整的 SSE 帧也就是说每条事件后面一定要加空行。media_typetext/event-stream不能省否则浏览器不会按事件流处理。Cache-Control: no-cache是 SSE 的推荐配置防止代理把响应缓存住导致客户端拿不到后续数据。其中X-Accel-Buffering: no是给 nginx 等代理看的告诉它不要缓冲这个响应。如果漏了这个头nginx 默认可能会攒一批数据再往后端转发前端看到的流式效果就变成每隔几秒吐一大段打字机手感直接没了。2.2 接入 LangChain用 stream() 而不是 invoke()LangChain 接入流式模型时很多人第一反应是model.invoke(prompt)但这会把模型所有输出都攒到结束才返回。要做打字机效果必须用流式接口。以ChatOpenAI为例老版本和新版本接口略有差异我建议以langchain_core的标准 API 为准因为它跨了多个模型供应商。基本写法是这样from langchain_openai import ChatOpenAI from langchain_core.prompts import ChatPromptTemplate model ChatOpenAI(modelgpt-4o-mini, temperature0, streamingTrue) prompt ChatPromptTemplate.from_messages([ (system, 你是一个助手只输出简洁的回答。), (human, {input}) ]) chain prompt | model async def llm_tokens(question: str): async for chunk in chain.astream({input: question}): content getattr(chunk, content, None) if content: yield content很多人在这一步有一个理解误区stream()返回的每个chunk不是一个汉字或一个 token的字符串而是模型 API 返回的一个结构化对象我们常说的增量文本一般藏在chunk.content字段里。不同的模型提供商chunk对象字段名可能不一样所以用getattr(chunk, content, None)去做兼容是稳妥的。LangChain 一些高版本还提供了callback方式的流式回调用法里面有个on_llm_new_token回调专门用来处理旧式 LLM 的 token 回调。如果你维护的是老项目看到它别慌思路一样拿到 token 就送进队列或直接 yield 出去。但新项目我更推荐直接用stream()/astream()代码更直观也不需要额外维护回调链。streamingTrue这个参数也要说一下。如果模型实例没开 streaming再调astream()可能只返回一个完整 chunk流式效果就变成一次性全量返回。所以实例化模型时一定要检查模型是否支持流式并在参数里显式开启。2.3 服务端封装三个必须处理的点后端把 LLM token 转 SSE 时我建议准备一个统一封装函数不要让业务接口到处裸写协议逻辑。封装时需要处理下面三件事。第一是连接关闭检测。SSE 连接不是永久的客户端刷新、断网、代理超时都会导致连接断开。如果服务端不感知模型还在继续生成白白浪费 token 费用。FastAPI 的StreamingResponse里如果客户端断开生成器在yield时通常会抛出异常所以要在生成器里捕获asyncio.CancelledError或Exception做清理并在日志里记录一次断流。第二是心跳机制。真实生产环境里模型在某些特殊问题上思考时间可能很长比如内部用到了检索或工具调用长时间没有 token 输出中间代理如果设置了空闲超时就会把连接掐断前端就会出现stream disconnected before completion: idle timeout waiting for sse。一个常规做法是给事件流发送 SSE 注释帧比如: keep-alive\n\n注释行会被客户端忽略但能保持连接上有数据传输避免被代理判定为空闲。第三是鉴权与 CORS。由于要设置Authorization头很多前端会放弃原生EventSource改用fetch流式读取这样就涉及跨域。服务端必须明确配置Access-Control-Allow-Headers: Authorization和Access-Control-Allow-Origin否则浏览器会在真正的流式数据到达之前直接把请求拦掉而且这种拦截在浏览器网络面板里看起来特别像连接失败。我封装的时候通常会定义一个生成器包装器把上面这些逻辑集中在一起业务代码只负责往队列里塞 tokenasync def sse_wrapper(queue: asyncio.Queue): try: while True: payload await queue.get() if payload is None: break yield fevent: token\ndata: {json.dumps(payload, ensure_asciiFalse)}\n\n except asyncio.CancelledError: pass finally: yield event: done\ndata: {finished:true}\n\n3. 前端打字机效果从字节流到逐字上屏3.1 用 fetch 加 ReadableStream 手工解析 SSE前端的标准方案是EventSource但它有两个死穴一是只支持 GET 请求且没法自定义请求头二是 POST 场景基本不可用。现在大模型应用几乎都要带 token 鉴权而且很多时候要传很长的 prompt所以主流做法是用fetch加ReadableStream手工解析 SSE 格式。核心思路是把fetch返回的response.body当成一个异步迭代器用TextDecoder把二进制块解码成字符串再按 SSE 协议切分事件。一段可以照着用的核心代码async function fetchSSE(url, options, handlers) { const response await fetch(url, options); if (!response.ok) { throw new Error(HTTP ${response.status}); } const reader response.body.getReader(); const decoder new TextDecoder(utf-8); let buffer ; while (true) { const { value, done } await reader.read(); if (done) break; buffer decoder.decode(value, { stream: true }); const frames buffer.split(/\r?\n\r?\n/); buffer frames.pop(); for (const frame of frames) { let eventType message; const dataLines []; const lines frame.split(/\r?\n/); for (const line of lines) { if (line.startsWith(event:)) { eventType line.slice(6).trim(); } else if (line.startsWith(data:)) { dataLines.push(line.slice(5).trimStart()); } } if (dataLines.length 0) { try { const parsed JSON.parse(dataLines.join(\n)); handlers[eventType]?.(parsed); } catch (e) { handlers.error?.({ type: parse_error, raw: dataLines.join(\n) }); } } } } }这段代码用了缓冲字符串buffer因为网络传输是字节流一次read()可能返回半个事件也可能返回多个事件必须在内存里把不完整的部分暂存下来等后续数据凑齐了再切分。TextDecoder.decode(value, { stream: true })也很关键它用来处理多字节字符被拆在两次读取中的情况。比如一个中文字符的 UTF-8 编码可能占 3 个字节如果服务端刚好在这个字符中间切分不加stream: true的话前端就会显示乱码。3.2 断流问题实录idle timeout waiting for SSEstream disconnected before completion: idle timeout waiting for sse 这个错误我最初看到时第一反应是服务端代码写错了后来排查发现是链路中某个代理节点在作祟。这类空闲超时出现的原理是SSE 连接建立后客户端和代理之间如果在一段时间内没有任何数据帧传输代理比如云厂商的负载均衡、nginx、网关就会主动断开连接节省连接资源。问题在于大模型在做工具调用或者跨多个子任务时真的可能几秒钟一个 token 都不出中间这一段正好触发代理的空闲超时。解决手段通常有三层。第一层是服务端主动发心跳帧让连接持续有数据流动我前面提到过: keep-alive\n\n的注释帧这是 SSE 协议里专门保活用的。第二层是调大代理的超时时间比如 nginx 通常会默认proxy_read_timeout 60s可以改成proxy_read_timeout 300s并且一定要开proxy_buffering off不然流式缓冲会严重影响体验。第三层是前端做好断线重连捕获到异常后携带Last-Event-ID重新请求服务端根据这个 ID 决定是否从断点续传。我在实际项目里会把这三层都做上因为生产环境链路太长任何一个环节都可能掐断连接。前端重连代码不能省即便有服务器保活用户自己也可能会切后台导致浏览器冻结连接。3.3 打字机渲染不只是收到就追加拿到事件流后最差的实现是每收到一个 token 就立刻操作 DOM 一次。在模型输出稍微密集的时候DOM 频繁更新会直接导致页面卡顿尤其是内容多、页面复杂的情况下。我推荐的方案是维护一个 token 队列渲染层用固定间隔批量消费队列里的内容。比如每 30ms 到 50ms 从队列里取一批增量文本一次性拼到内容节点上。这样视觉上依然是平滑的打字机效果但 DOM 的更新频率大幅降低。更讲究一点的还可以在中文标点处做停顿。中文打字机体验里标点符号后面通常是语义停顿点比如句号、感叹号、问号后面稍微停 80ms 到 120ms读起来更自然。这个逻辑可以在渲染循环里判断一下const pauseRegex /[。]/; let lastChar pendingText[pendingText.length - 1]; let extraDelay pauseRegex.test(lastChar) ? 100 : 0; setTimeout(flush, baseInterval extraDelay);另外要提醒一点流式渲染出来的文本如果是 HTML要用textContent而不是innerHTML否则模型输出内容里万一有img srcx onerror...之类的片段会直接变成 XSS 攻击向量。如果你真的需要渲染 Markdown也要等完整流结束之后再交给 Markdown 渲染库流式过程中老老实实用纯文本展示避免样式闪烁和标签断裂。4. LangChain 结构化输出怎么把文本稳固地变成 JSON4.1 大模型直接输出 JSON 的三个经典难题先说结论让大模型直接输出 JSON这件事在提示词层面做得到但不稳定。稳定性的短板来自模型本身是概率生成而不是严格的序列化器所以我在生产环境里从来不会只靠请你输出 JSON这句话。最常见的三类问题第一是模型把 JSON 包在 Markdown 代码块里比如返回好的这是结果 json {name: 张三, age: 18}这段字符串如果直接交给 json.loads() 必然报错。第二是模型在 JSON 前面或后面加了多余解释比如以下是你要的字段...、age 是年龄意思导致解析器无法定位 JSON 起始位置。第三是字段值类型不对比如传入 age: 18而 Pydantic 模型定义的是 int又或者枚举值大小写不对该 true 的给了 True。 有一个比较容易忽略的细节是大模型的输出往往不是一次成型。LangChain 在内部做代码生成、函数调用时经常会在最终 JSON 前输出一些中间思考过程如果提示词没说清楚这类思考痕迹就会混进最终结果里。所以结构化输出必须做两层既要在提示词层面约束格式又要在解析层做兜底修复。 ### 4.2 LangChain 输出解析器的正确用法 LangChain 提供了解析器体系常见的包括 StructuredOutputParser 和 PydanticOutputParser。这两者都不是魔法它们做的事很朴素把你定义的结构转成一段格式说明文本塞进提示词里引导模型按格式输出同时提供 parse() 方法在模型返回文本后按预期结构解析成对象。如果解析失败它们会抛异常。 PydanticOutputParser 是我更推荐的方式因为有 Pydantic 类型做背书后面接流程时不用担心字段缺失和类型错误。基本用法 python from typing import List, Literal from langchain_core.output_parsers import PydanticOutputParser from langchain_core.prompts import ChatPromptTemplate from pydantic import BaseModel, Field class Article(BaseModel): title: str Field(description文章标题) tags: List[str] Field(description文章标签) tone: Literal[positive, neutral, negative] Field(description整体情绪倾向) parser PydanticOutputParser(pydantic_objectArticle) prompt ChatPromptTemplate.from_messages([ (system, 根据用户的输入提取信息只输出 JSON不要输出任何解释。\n{format_instructions}), (human, {input}) ]) chain prompt | model | parser关键在于把parser.get_format_instructions()的输出拼接进提示词。format_instructions会包含字段说明、JSON 示例等模型看了之后输出合规 JSON 的概率会显著提高。注意它降低的是出错的概率而不是把概率变成 100%。LangChain 新版本里还有一个model.with_structured_output()的封装它会把 Pydantic 结构直接转成模型 API 的 JSON Schema 表单在底层让模型通过 function calling 方式返回结构化结果。这条路在某些模型上异常稳定但依赖厂商对 tool calling 的支持程度如果需要兼容多模型还是建议用解析器方案保底。4.3 JSON 兜底修复函数最后一公里不能省无论提示词怎么约束我还是会写一个尽力修复 JSON的函数把它放在解析器兜底位置。核心逻辑很固定先去 Markdown 围栏再提取第一个左花括号到最后一个右花括号之间的内容然后尝试json.loads()失败时尝试做常见修复。一个实际能用的 Python 版本import json import re def extract_json(text: str) - dict: if not text: raise ValueError(empty text) text re.sub(rjson||, , text.strip()) start, end text.find({), text.rfind(}) if start -1 or end -1 or end start: raise ValueError(no json object found) candidate text[start:end 1] try: return json.loads(candidate) except json.JSONDecodeError: pass candidate re.sub(r[\x00-\x1f], , candidate) candidate candidate.replace(, ) candidate re.sub(r(\w):, r\1:, candidate) candidate re.sub(r,\s*([}\]]), r\1, candidate) candidate candidate.replace(True, true).replace(False, false).replace(None, null) try: return json.loads(candidate) except json.JSONDecodeError as e: raise ValueError(fjson fix failed: {e}, raw{text[:200]})这个函数不是一个万能银弹单引号正则替换在值内部包含英文单引号时也可能误伤但它能覆盖绝大多数模型输出的常见毛病。我把它的定位说清楚它负责的是最后一道防线不是第一选择。第一选择永远是提示词约束和 LangChain 解析器这两层都失败了才轮到这个兜底函数出场。生产环境里我会把修复失败的内容连同原始文本一起记录到日志方便后面去调提示词。4.4 流式输出和 JSON 解析如何相处有人会问SSE 流式场景下能不能一边收 JSON 一边解析做到标题先显示出来字段一个个填充这个想法很好但实现起来要非常克制。流式输出时模型是一段一段生成 JSON 字符串的中间任何时刻这个字符串都可能是残缺的字段名写了一半、数组少一个括号、字符串值只出现开头一个引号。你当然可以用增量 JSON 解析器去试但实际效果大概率是频繁解析失败因为你不知道当前这个 token 结束后后面会不会立刻补上关键字符。我的建议流式传输和结构化解析不要在同一层做。SSE 流式层只管把模型输出的原始文本增量交给前端展示让用户看到打字机效果等done事件到达后后端或前端再拿完整内容做一次 JSON 解析解析成功则更新成结构化视图解析失败就走兜底修复。如果确实需要提前展示部分结构化字段一种折中方案是在提示词里要求模型先输出一个摘要字段最后再输出完整 JSON。前端拿到完整内容后依旧只解析一次但从用户体验上是先看到摘要再看到结构化结果而不会因为尝试解析半截 JSON 而引入大量 bug。5. 把完整链路串起来一个最小可复刻的工程示例5.1 服务端完整示例FastAPI LangChain SSE JSON 兜底下面给一个两端都能跑通的最小工程。服务端用 FastAPI模型用支持流式的 OpenAI 兼容接口收到用户问题后先流式返回 token最后再一次性返回结构化结果。import asyncio import json from typing import List from fastapi import FastAPI from fastapi.responses import StreamingResponse from langchain_openai import ChatOpenAI from langchain_core.prompts import ChatPromptTemplate from langchain_core.output_parsers import PydanticOutputParser from pydantic import BaseModel, Field app FastAPI() class Summary(BaseModel): key_points: List[str] Field(description总结出的关键点) tone: str Field(description整体语气) parser PydanticOutputParser(pydantic_objectSummary) model ChatOpenAI(modelgpt-4o-mini, temperature0, streamingTrue) async def ask(question: str): token_prompt ChatPromptTemplate.from_messages([ (system, 你是助手用大白话回答用户问题。), (human, {question}) ]) async for chunk in (token_prompt | model).astream({question: question}): content getattr(chunk, content, ) if content: yield fevent: token\ndata: {json.dumps({delta: content}, ensure_asciiFalse)}\n\n await asyncio.sleep(0) app.post(/chat) async def chat(question: str): async def gen(): async for frame in ask(question): yield frame try: final_prompt ChatPromptTemplate.from_messages([ (system, 从回答中提取关键信息只输出 JSON。\n{format_instructions}), (human, 问题{question}) ]) final_chain final_prompt | model | parser result await final_chain.ainvoke({question: question, format_instructions: parser.get_format_instructions()}) yield fevent: done\ndata: {json.dumps(result.model_dump(), ensure_asciiFalse)}\n\n except Exception as e: yield fevent: error\ndata: {json.dumps({message: str(e)}, ensure_asciiFalse)}\n\n return StreamingResponse(gen(), media_typetext/event-stream, headers{Cache-Control: no-cache, X-Accel-Buffering: no})这个例子为了可读性省略了 Nginx 心跳、鉴权和连接异常清理但主链路是完整的先astream流式吐字结束后用同一个模型再跑一次结构化提取返回done事件。这种先展示后结构化的设计在工程上是比较可控的。因为第二次调用用的是invoke全量结果不会把半截 JSON 暴露给下游。5.2 前端核心调用片段前端以 3.1 节提到的fetchSSE为基础只需要传入Authorization头并监听token和done事件async function main(question) { const handlers { token: (data) { pushToRenderQueue(data.delta); }, done: (data) { renderStructuredView(data); }, error: (err) { console.error(SSE error, err); handleReconnect(); } }; await fetchSSE(/chat, { method: POST, headers: { Content-Type: application/json, Authorization: Bearer ${localStorage.getItem(token)} }, body: JSON.stringify({ question }) }, handlers); }这里要特别注意fetchSSE实现的细节很多前端新手在解析 SSE 时习惯用res.text()拿完整文本这么写就完全没用到流式特性打字机效果直接变一次性打印。另外在服务端返回Content-Type: text/event-stream的前提下浏览器不会自动帮你去解析事件类型所有事件都必须手工按协议处理。5.3 我实际调试中的几条体会最后分享几条在实际项目里横跨后端和前端折腾出来的经验。第一先本地用 curl 验证 SSE再开前端。连接不稳定时十次有八次是代理或网络层的问题而不是业务代码的问题。用curl -N -H Accept: text/event-stream http://localhost:8000/chat可以确认服务端本身流式是否正常避免在前端 Debug 里无从下手。第二模型提示词里如果只是说输出 JSON很多时候不够。要同时给出字段名、字段含义、示例输出。LangChain 的format_instructions已经帮你生成这部分了但如果你自己手写 prompt也要保证示例输出和 Pydantic 定义严格一致。不同模型对 JSON Schema 的敏感度差异很大对某些小模型最好把示例 JSON 原样放进提示词它才会照葫芦画瓢。第三慎用流式拼接 JSON 再逐步解析。我见过团队试图在 SSE 中维护一个全局字符串每收到一个 token 就重新调用json.loads然后界面在解析成功和失败之间反复横跳体验反而更差。与其做这种不稳定的事不如老老实实等完整输出再走一次extract_json兜底。第四SSE 的心跳不是可选项而是必选项。千万不要认为大模型响应够快就不会有超时一旦你的应用里接入工具调用、Agent 规划或者多步推理模型在每一步中间都可能长时间不吐字心跳帧那 30 秒一发几乎救了我好几次线上事故。第五把事件流的协议版本号放在dataJSON 里方便将来升级。哪怕现在只有 v1 的解析逻辑提前写一个v: 1就够了。等将来要改消息格式时客户端可以根据版本分支处理而不是一口气推翻整个协议。 SSE与LangChain实战：大模型流式传输到结构化JSON的完整链路
做 AI 应用的朋友应该都有同感页面上的回答要像人打字一样逐字蹦出来后端要把大模型的 token 流持续推给前端同时前端最终还要拿到一个能被下游直接消费的 JSON 对象而不是一段夹杂着闲聊、Markdown 符号和多余解释的文本。这背后其实是三个问题绑在一起解决SSE 流式传输怎么做、LangChain 怎么接入流式模型、以及怎么把模型的文本输出稳定解析成结构化 JSON。这篇文章我把整套链路从头拆到尾后端怎么发包、前端怎么收包、LangChain 怎么约束输出、JSON 怎么兜底修复全都过一遍并附上我能直接复现的最小工程代码。适合看这篇内容的人是那些正在做 ChatBot、AI Agent、智能客服或任何要展示流式回答又要有结构化数据的开发者。如果你已经知道 SSE 是 Server-Sent Events但不太清楚它在 LangChain 场景里怎么落地又或者你遇到过模型输出的 JSON 总是带着 json 代码块、字段类型不对、流式传输一半连接被掐断这类问题那这篇文章可以让你少走一大截弯路。1. 先别急着写代码把 SSE 流式协议吃透1.1 一次 HTTP 请求响应体被切成了一串消息SSE 全称 Server-Sent Events中文常叫服务器推送事件。它没有引入新的协议底层依然是大家熟悉的 HTTP。普通请求里服务端把完整响应体一次性写完然后关掉连接SSE 请求则不同服务端在响应体里不断写入文本块连接保持打开客户端每收到一个块就可以立刻处理。这就是为什么它能做打字机效果的技术基础服务端每生成一个 token 就扔给前端前端收到一个 token 渲染一个 token。SSE 对响应格式有明确规定。服务端必须设置Content-Type: text/event-stream并且消息体遵循事件流语法。一个典型的事件流长这样id: 1 event: token data: {delta:你} id: 2 event: token data: {delta:好} id: 3 event: done data: {finished:true}这里有几个关键点。第一每个事件由两个连续换行符\n\n分隔也就是说空行是事件之间的边界。第二data:表示数据内容多行data:会被客户端自动合并成一行所以有些人会在每条 message 里故意把 JSON 折成多行但规范上到最后还是会拼在一起。第三event:用来声明事件类型客户端可以用addEventListener(token, ...)监听具体类型如果不写event:默认是message事件。第四id:表示事件编号客户端会记住最后一个收到的 id断线重连时把它放到Last-Event-ID请求头里发给服务端服务端借此实现增量续传。很多后端开发第一次写 SSE 容易忽略的细节是事件流中如果有多余的空格或者没有空行作为分隔客户端解析就会错位。前端浏览器里的EventSource解析器是严格按空行切事件的服务端该\n\n的地方一个都不能省。这个问题在 nginx 缓冲、代理转发、语言框架自动换行等场景里很容易被放大所以协议细节必须先搞清楚不然后面踩坑时不知道从哪查起。1.2 为什么这类场景选 SSE 而不是 WebSocket每次讲到流式传输总有人问为什么不用 WebSocket我的答案是业务场景决定选型大模型流式回答是一个单向推送、几乎不需要客户端反向实时发消息的场景SSE 比 WebSocket 有更明显的性价比优势。可以用一个表快速对比对比项SSEWebSocket底层协议HTTP兼容成本低独立 ws/wss 协议需要升级握手机制方向仅服务端到客户端单向推送全双工双向通信断线重连浏览器原生支持自动重连需要自己实现心跳和重连逻辑消息格式文本为主按事件流解析文本或二进制帧服务端实现难度普通 HTTP 响应保持不关闭即可需要专门处理握手、帧解析、连接状态调试方便度浏览器、curl 都可以直接看需要专用工具或代码辅助大模型流式接口的诉求很典型用户问了一句话之后大量数据是从模型往用户端单向流动。SSE 正好不用做双向通道客户端断开时服务端在写操作中能感知到BrokenPipe之类的异常也足够处理连接生命周期。反过来WebSocket 的全双工能力在这种场景里用不上反而多出心跳、消息帧格式、代理配置等一堆额外复杂度。还有一点SSE 基于 HTTP意味着它可以借用现有的请求头做鉴权。比如把 token 放在Authorization头里传给服务端这在 WebSocket 协议里反而要绕一圈。虽然浏览器原生EventSource不支持自定义请求头但后面我会讲到用fetch加ReadableStream完全可以绕开这个限制同时保留 SSE 协议本身的优点。1.3 合作之前先把事件格式设计好很多团队写 SSE 时候只往data:里塞一段纯文本以为能跑就行。短期的确能跑但业务一复杂就难受。比如你们后面要在原型之外加个追问 token 数量、附带工具调用 ID、返回最终结构化结果等字段纯文本格式就得折腾一遍协议。我的经验是让data:里的内容从第一天起就是 JSON 字符串并且明确约定事件类型。给一种我实际用下来比较舒服的通知格式模型每个增量 token 发一个event: token事件data是一个 JSON 对象里面包含delta新生成的文本片段当模型输出被完整消费完再发一个event: done事件data里带最终完整结果比如解析好的 JSON 对象或统计信息如果中间出错发一个event: error事件data里带code和message。这样设计的好处是前端可以用addEventListener(token, handler)分别监听业务逻辑不会把正常文本和错误信息混在一起同时data用 JSON 而不是裸字符串将来要扩展事件类型、协议版本、附加元数据时不用改动消息的分隔方式只需要在 JSON 里加字段。2. 后端怎么把大模型的 Token 变成 SSE 数据流2.1 FastAPI 实现最小 SSE 接口如果你在用 Python 做后端FastAPI 是这几天我见过的最适合写 SSE 的框架之一。它本身基于 Starlette自带StreamingResponse不需要额外引第三方库就能实现流式响应。先写一个不依赖 LangChain 的最小版本用生成器模拟事件流import asyncio import json from fastapi import FastAPI from fastapi.responses import StreamingResponse app FastAPI() async def event_generator(): for i, chunk in enumerate([你, 好, , 世, 界]): event { event: token, data: json.dumps({delta: chunk}, ensure_asciiFalse) } yield fevent: {event[event]}\ndata: {event[data]}\n\n await asyncio.sleep(0.05) yield event: done\ndata: {finished:true}\n\n app.get(/sse) async def sse_endpoint(): return StreamingResponse( event_generator(), media_typetext/event-stream, headers{ Cache-Control: no-cache, Connection: keep-alive, X-Accel-Buffering: no, }, )注意几个容易踩的坑。yield的内容必须是完整的 SSE 帧也就是说每条事件后面一定要加空行。media_typetext/event-stream不能省否则浏览器不会按事件流处理。Cache-Control: no-cache是 SSE 的推荐配置防止代理把响应缓存住导致客户端拿不到后续数据。其中X-Accel-Buffering: no是给 nginx 等代理看的告诉它不要缓冲这个响应。如果漏了这个头nginx 默认可能会攒一批数据再往后端转发前端看到的流式效果就变成每隔几秒吐一大段打字机手感直接没了。2.2 接入 LangChain用 stream() 而不是 invoke()LangChain 接入流式模型时很多人第一反应是model.invoke(prompt)但这会把模型所有输出都攒到结束才返回。要做打字机效果必须用流式接口。以ChatOpenAI为例老版本和新版本接口略有差异我建议以langchain_core的标准 API 为准因为它跨了多个模型供应商。基本写法是这样from langchain_openai import ChatOpenAI from langchain_core.prompts import ChatPromptTemplate model ChatOpenAI(modelgpt-4o-mini, temperature0, streamingTrue) prompt ChatPromptTemplate.from_messages([ (system, 你是一个助手只输出简洁的回答。), (human, {input}) ]) chain prompt | model async def llm_tokens(question: str): async for chunk in chain.astream({input: question}): content getattr(chunk, content, None) if content: yield content很多人在这一步有一个理解误区stream()返回的每个chunk不是一个汉字或一个 token的字符串而是模型 API 返回的一个结构化对象我们常说的增量文本一般藏在chunk.content字段里。不同的模型提供商chunk对象字段名可能不一样所以用getattr(chunk, content, None)去做兼容是稳妥的。LangChain 一些高版本还提供了callback方式的流式回调用法里面有个on_llm_new_token回调专门用来处理旧式 LLM 的 token 回调。如果你维护的是老项目看到它别慌思路一样拿到 token 就送进队列或直接 yield 出去。但新项目我更推荐直接用stream()/astream()代码更直观也不需要额外维护回调链。streamingTrue这个参数也要说一下。如果模型实例没开 streaming再调astream()可能只返回一个完整 chunk流式效果就变成一次性全量返回。所以实例化模型时一定要检查模型是否支持流式并在参数里显式开启。2.3 服务端封装三个必须处理的点后端把 LLM token 转 SSE 时我建议准备一个统一封装函数不要让业务接口到处裸写协议逻辑。封装时需要处理下面三件事。第一是连接关闭检测。SSE 连接不是永久的客户端刷新、断网、代理超时都会导致连接断开。如果服务端不感知模型还在继续生成白白浪费 token 费用。FastAPI 的StreamingResponse里如果客户端断开生成器在yield时通常会抛出异常所以要在生成器里捕获asyncio.CancelledError或Exception做清理并在日志里记录一次断流。第二是心跳机制。真实生产环境里模型在某些特殊问题上思考时间可能很长比如内部用到了检索或工具调用长时间没有 token 输出中间代理如果设置了空闲超时就会把连接掐断前端就会出现stream disconnected before completion: idle timeout waiting for sse。一个常规做法是给事件流发送 SSE 注释帧比如: keep-alive\n\n注释行会被客户端忽略但能保持连接上有数据传输避免被代理判定为空闲。第三是鉴权与 CORS。由于要设置Authorization头很多前端会放弃原生EventSource改用fetch流式读取这样就涉及跨域。服务端必须明确配置Access-Control-Allow-Headers: Authorization和Access-Control-Allow-Origin否则浏览器会在真正的流式数据到达之前直接把请求拦掉而且这种拦截在浏览器网络面板里看起来特别像连接失败。我封装的时候通常会定义一个生成器包装器把上面这些逻辑集中在一起业务代码只负责往队列里塞 tokenasync def sse_wrapper(queue: asyncio.Queue): try: while True: payload await queue.get() if payload is None: break yield fevent: token\ndata: {json.dumps(payload, ensure_asciiFalse)}\n\n except asyncio.CancelledError: pass finally: yield event: done\ndata: {finished:true}\n\n3. 前端打字机效果从字节流到逐字上屏3.1 用 fetch 加 ReadableStream 手工解析 SSE前端的标准方案是EventSource但它有两个死穴一是只支持 GET 请求且没法自定义请求头二是 POST 场景基本不可用。现在大模型应用几乎都要带 token 鉴权而且很多时候要传很长的 prompt所以主流做法是用fetch加ReadableStream手工解析 SSE 格式。核心思路是把fetch返回的response.body当成一个异步迭代器用TextDecoder把二进制块解码成字符串再按 SSE 协议切分事件。一段可以照着用的核心代码async function fetchSSE(url, options, handlers) { const response await fetch(url, options); if (!response.ok) { throw new Error(HTTP ${response.status}); } const reader response.body.getReader(); const decoder new TextDecoder(utf-8); let buffer ; while (true) { const { value, done } await reader.read(); if (done) break; buffer decoder.decode(value, { stream: true }); const frames buffer.split(/\r?\n\r?\n/); buffer frames.pop(); for (const frame of frames) { let eventType message; const dataLines []; const lines frame.split(/\r?\n/); for (const line of lines) { if (line.startsWith(event:)) { eventType line.slice(6).trim(); } else if (line.startsWith(data:)) { dataLines.push(line.slice(5).trimStart()); } } if (dataLines.length 0) { try { const parsed JSON.parse(dataLines.join(\n)); handlers[eventType]?.(parsed); } catch (e) { handlers.error?.({ type: parse_error, raw: dataLines.join(\n) }); } } } } }这段代码用了缓冲字符串buffer因为网络传输是字节流一次read()可能返回半个事件也可能返回多个事件必须在内存里把不完整的部分暂存下来等后续数据凑齐了再切分。TextDecoder.decode(value, { stream: true })也很关键它用来处理多字节字符被拆在两次读取中的情况。比如一个中文字符的 UTF-8 编码可能占 3 个字节如果服务端刚好在这个字符中间切分不加stream: true的话前端就会显示乱码。3.2 断流问题实录idle timeout waiting for SSEstream disconnected before completion: idle timeout waiting for sse 这个错误我最初看到时第一反应是服务端代码写错了后来排查发现是链路中某个代理节点在作祟。这类空闲超时出现的原理是SSE 连接建立后客户端和代理之间如果在一段时间内没有任何数据帧传输代理比如云厂商的负载均衡、nginx、网关就会主动断开连接节省连接资源。问题在于大模型在做工具调用或者跨多个子任务时真的可能几秒钟一个 token 都不出中间这一段正好触发代理的空闲超时。解决手段通常有三层。第一层是服务端主动发心跳帧让连接持续有数据流动我前面提到过: keep-alive\n\n的注释帧这是 SSE 协议里专门保活用的。第二层是调大代理的超时时间比如 nginx 通常会默认proxy_read_timeout 60s可以改成proxy_read_timeout 300s并且一定要开proxy_buffering off不然流式缓冲会严重影响体验。第三层是前端做好断线重连捕获到异常后携带Last-Event-ID重新请求服务端根据这个 ID 决定是否从断点续传。我在实际项目里会把这三层都做上因为生产环境链路太长任何一个环节都可能掐断连接。前端重连代码不能省即便有服务器保活用户自己也可能会切后台导致浏览器冻结连接。3.3 打字机渲染不只是收到就追加拿到事件流后最差的实现是每收到一个 token 就立刻操作 DOM 一次。在模型输出稍微密集的时候DOM 频繁更新会直接导致页面卡顿尤其是内容多、页面复杂的情况下。我推荐的方案是维护一个 token 队列渲染层用固定间隔批量消费队列里的内容。比如每 30ms 到 50ms 从队列里取一批增量文本一次性拼到内容节点上。这样视觉上依然是平滑的打字机效果但 DOM 的更新频率大幅降低。更讲究一点的还可以在中文标点处做停顿。中文打字机体验里标点符号后面通常是语义停顿点比如句号、感叹号、问号后面稍微停 80ms 到 120ms读起来更自然。这个逻辑可以在渲染循环里判断一下const pauseRegex /[。]/; let lastChar pendingText[pendingText.length - 1]; let extraDelay pauseRegex.test(lastChar) ? 100 : 0; setTimeout(flush, baseInterval extraDelay);另外要提醒一点流式渲染出来的文本如果是 HTML要用textContent而不是innerHTML否则模型输出内容里万一有img srcx onerror...之类的片段会直接变成 XSS 攻击向量。如果你真的需要渲染 Markdown也要等完整流结束之后再交给 Markdown 渲染库流式过程中老老实实用纯文本展示避免样式闪烁和标签断裂。4. LangChain 结构化输出怎么把文本稳固地变成 JSON4.1 大模型直接输出 JSON 的三个经典难题先说结论让大模型直接输出 JSON这件事在提示词层面做得到但不稳定。稳定性的短板来自模型本身是概率生成而不是严格的序列化器所以我在生产环境里从来不会只靠请你输出 JSON这句话。最常见的三类问题第一是模型把 JSON 包在 Markdown 代码块里比如返回好的这是结果 json {name: 张三, age: 18}这段字符串如果直接交给 json.loads() 必然报错。第二是模型在 JSON 前面或后面加了多余解释比如以下是你要的字段...、age 是年龄意思导致解析器无法定位 JSON 起始位置。第三是字段值类型不对比如传入 age: 18而 Pydantic 模型定义的是 int又或者枚举值大小写不对该 true 的给了 True。 有一个比较容易忽略的细节是大模型的输出往往不是一次成型。LangChain 在内部做代码生成、函数调用时经常会在最终 JSON 前输出一些中间思考过程如果提示词没说清楚这类思考痕迹就会混进最终结果里。所以结构化输出必须做两层既要在提示词层面约束格式又要在解析层做兜底修复。 ### 4.2 LangChain 输出解析器的正确用法 LangChain 提供了解析器体系常见的包括 StructuredOutputParser 和 PydanticOutputParser。这两者都不是魔法它们做的事很朴素把你定义的结构转成一段格式说明文本塞进提示词里引导模型按格式输出同时提供 parse() 方法在模型返回文本后按预期结构解析成对象。如果解析失败它们会抛异常。 PydanticOutputParser 是我更推荐的方式因为有 Pydantic 类型做背书后面接流程时不用担心字段缺失和类型错误。基本用法 python from typing import List, Literal from langchain_core.output_parsers import PydanticOutputParser from langchain_core.prompts import ChatPromptTemplate from pydantic import BaseModel, Field class Article(BaseModel): title: str Field(description文章标题) tags: List[str] Field(description文章标签) tone: Literal[positive, neutral, negative] Field(description整体情绪倾向) parser PydanticOutputParser(pydantic_objectArticle) prompt ChatPromptTemplate.from_messages([ (system, 根据用户的输入提取信息只输出 JSON不要输出任何解释。\n{format_instructions}), (human, {input}) ]) chain prompt | model | parser关键在于把parser.get_format_instructions()的输出拼接进提示词。format_instructions会包含字段说明、JSON 示例等模型看了之后输出合规 JSON 的概率会显著提高。注意它降低的是出错的概率而不是把概率变成 100%。LangChain 新版本里还有一个model.with_structured_output()的封装它会把 Pydantic 结构直接转成模型 API 的 JSON Schema 表单在底层让模型通过 function calling 方式返回结构化结果。这条路在某些模型上异常稳定但依赖厂商对 tool calling 的支持程度如果需要兼容多模型还是建议用解析器方案保底。4.3 JSON 兜底修复函数最后一公里不能省无论提示词怎么约束我还是会写一个尽力修复 JSON的函数把它放在解析器兜底位置。核心逻辑很固定先去 Markdown 围栏再提取第一个左花括号到最后一个右花括号之间的内容然后尝试json.loads()失败时尝试做常见修复。一个实际能用的 Python 版本import json import re def extract_json(text: str) - dict: if not text: raise ValueError(empty text) text re.sub(rjson||, , text.strip()) start, end text.find({), text.rfind(}) if start -1 or end -1 or end start: raise ValueError(no json object found) candidate text[start:end 1] try: return json.loads(candidate) except json.JSONDecodeError: pass candidate re.sub(r[\x00-\x1f], , candidate) candidate candidate.replace(, ) candidate re.sub(r(\w):, r\1:, candidate) candidate re.sub(r,\s*([}\]]), r\1, candidate) candidate candidate.replace(True, true).replace(False, false).replace(None, null) try: return json.loads(candidate) except json.JSONDecodeError as e: raise ValueError(fjson fix failed: {e}, raw{text[:200]})这个函数不是一个万能银弹单引号正则替换在值内部包含英文单引号时也可能误伤但它能覆盖绝大多数模型输出的常见毛病。我把它的定位说清楚它负责的是最后一道防线不是第一选择。第一选择永远是提示词约束和 LangChain 解析器这两层都失败了才轮到这个兜底函数出场。生产环境里我会把修复失败的内容连同原始文本一起记录到日志方便后面去调提示词。4.4 流式输出和 JSON 解析如何相处有人会问SSE 流式场景下能不能一边收 JSON 一边解析做到标题先显示出来字段一个个填充这个想法很好但实现起来要非常克制。流式输出时模型是一段一段生成 JSON 字符串的中间任何时刻这个字符串都可能是残缺的字段名写了一半、数组少一个括号、字符串值只出现开头一个引号。你当然可以用增量 JSON 解析器去试但实际效果大概率是频繁解析失败因为你不知道当前这个 token 结束后后面会不会立刻补上关键字符。我的建议流式传输和结构化解析不要在同一层做。SSE 流式层只管把模型输出的原始文本增量交给前端展示让用户看到打字机效果等done事件到达后后端或前端再拿完整内容做一次 JSON 解析解析成功则更新成结构化视图解析失败就走兜底修复。如果确实需要提前展示部分结构化字段一种折中方案是在提示词里要求模型先输出一个摘要字段最后再输出完整 JSON。前端拿到完整内容后依旧只解析一次但从用户体验上是先看到摘要再看到结构化结果而不会因为尝试解析半截 JSON 而引入大量 bug。5. 把完整链路串起来一个最小可复刻的工程示例5.1 服务端完整示例FastAPI LangChain SSE JSON 兜底下面给一个两端都能跑通的最小工程。服务端用 FastAPI模型用支持流式的 OpenAI 兼容接口收到用户问题后先流式返回 token最后再一次性返回结构化结果。import asyncio import json from typing import List from fastapi import FastAPI from fastapi.responses import StreamingResponse from langchain_openai import ChatOpenAI from langchain_core.prompts import ChatPromptTemplate from langchain_core.output_parsers import PydanticOutputParser from pydantic import BaseModel, Field app FastAPI() class Summary(BaseModel): key_points: List[str] Field(description总结出的关键点) tone: str Field(description整体语气) parser PydanticOutputParser(pydantic_objectSummary) model ChatOpenAI(modelgpt-4o-mini, temperature0, streamingTrue) async def ask(question: str): token_prompt ChatPromptTemplate.from_messages([ (system, 你是助手用大白话回答用户问题。), (human, {question}) ]) async for chunk in (token_prompt | model).astream({question: question}): content getattr(chunk, content, ) if content: yield fevent: token\ndata: {json.dumps({delta: content}, ensure_asciiFalse)}\n\n await asyncio.sleep(0) app.post(/chat) async def chat(question: str): async def gen(): async for frame in ask(question): yield frame try: final_prompt ChatPromptTemplate.from_messages([ (system, 从回答中提取关键信息只输出 JSON。\n{format_instructions}), (human, 问题{question}) ]) final_chain final_prompt | model | parser result await final_chain.ainvoke({question: question, format_instructions: parser.get_format_instructions()}) yield fevent: done\ndata: {json.dumps(result.model_dump(), ensure_asciiFalse)}\n\n except Exception as e: yield fevent: error\ndata: {json.dumps({message: str(e)}, ensure_asciiFalse)}\n\n return StreamingResponse(gen(), media_typetext/event-stream, headers{Cache-Control: no-cache, X-Accel-Buffering: no})这个例子为了可读性省略了 Nginx 心跳、鉴权和连接异常清理但主链路是完整的先astream流式吐字结束后用同一个模型再跑一次结构化提取返回done事件。这种先展示后结构化的设计在工程上是比较可控的。因为第二次调用用的是invoke全量结果不会把半截 JSON 暴露给下游。5.2 前端核心调用片段前端以 3.1 节提到的fetchSSE为基础只需要传入Authorization头并监听token和done事件async function main(question) { const handlers { token: (data) { pushToRenderQueue(data.delta); }, done: (data) { renderStructuredView(data); }, error: (err) { console.error(SSE error, err); handleReconnect(); } }; await fetchSSE(/chat, { method: POST, headers: { Content-Type: application/json, Authorization: Bearer ${localStorage.getItem(token)} }, body: JSON.stringify({ question }) }, handlers); }这里要特别注意fetchSSE实现的细节很多前端新手在解析 SSE 时习惯用res.text()拿完整文本这么写就完全没用到流式特性打字机效果直接变一次性打印。另外在服务端返回Content-Type: text/event-stream的前提下浏览器不会自动帮你去解析事件类型所有事件都必须手工按协议处理。5.3 我实际调试中的几条体会最后分享几条在实际项目里横跨后端和前端折腾出来的经验。第一先本地用 curl 验证 SSE再开前端。连接不稳定时十次有八次是代理或网络层的问题而不是业务代码的问题。用curl -N -H Accept: text/event-stream http://localhost:8000/chat可以确认服务端本身流式是否正常避免在前端 Debug 里无从下手。第二模型提示词里如果只是说输出 JSON很多时候不够。要同时给出字段名、字段含义、示例输出。LangChain 的format_instructions已经帮你生成这部分了但如果你自己手写 prompt也要保证示例输出和 Pydantic 定义严格一致。不同模型对 JSON Schema 的敏感度差异很大对某些小模型最好把示例 JSON 原样放进提示词它才会照葫芦画瓢。第三慎用流式拼接 JSON 再逐步解析。我见过团队试图在 SSE 中维护一个全局字符串每收到一个 token 就重新调用json.loads然后界面在解析成功和失败之间反复横跳体验反而更差。与其做这种不稳定的事不如老老实实等完整输出再走一次extract_json兜底。第四SSE 的心跳不是可选项而是必选项。千万不要认为大模型响应够快就不会有超时一旦你的应用里接入工具调用、Agent 规划或者多步推理模型在每一步中间都可能长时间不吐字心跳帧那 30 秒一发几乎救了我好几次线上事故。第五把事件流的协议版本号放在dataJSON 里方便将来升级。哪怕现在只有 v1 的解析逻辑提前写一个v: 1就够了。等将来要改消息格式时客户端可以根据版本分支处理而不是一口气推翻整个协议。