1. 流式输出的错觉你以为在拼字符串实际在拼半成品做 AI 应用的同学应该都有过这种体验没开流式输出之前一切岁月静好JSON.parse随便用。一旦把接口切到stream: true世界立刻支离破碎。模型最后吐出来明明是一个合法 JSON可你在每个 chunk 到达的瞬间去解析得到的永远是Unexpected end of JSON input。这篇文章专门聊这个大坑从流式输出的底层原理到 JSON、XML 的容错解析再到 Zod 校验和 Tool Calls 的增量拼装最后串成一条可以直接落地的生产级链路。适合正在做 AI 聊天应用、Agent、RAG 工具链或者任何需要一边流式渲染、一边结构化取数场景的前后端工程师参考前端为主涉及后端部分我会同步补原理两边都能照着抄。1.1 一次真实的流式输出 JSON现场事故我先还原一个自己踩过的现场。当时做一个内部数据分析助手后端用 vLLM 部署模型前端用fetch拉流式接口。初版代码长这个样子const res await fetch(/api/chat, { method: POST, headers: { Content-Type: application/json }, body: JSON.stringify({ messages, stream: true }) }); const reader res.body.getReader(); const decoder new TextDecoder(utf-8); while (true) { const { done, value } await reader.read(); if (done) break; const chunk decoder.decode(value); // 天真地以为每个 chunk 都是一个完整的 JSON const data JSON.parse(chunk); // 抛错整个前端逻辑直接崩 }第一次跑就哐哐报错。为什么因为流式接口返回的每个 chunk本质上只是 HTTP 响应体里新抵达的一段未定边界的字节流。网络包从哪里切取决于传输层、代理和缓冲区状态跟模型输出内容的语义毫无关系。模型打算输出{city:上海}第一个 chunk 完全可能是{city:上第二个 chunk 才是海}。在这种前提下对每个 chunk 单独做JSON.parse必然失败这是流式的物理特性不是配置问题。我当时的第一反应是攒起来流结束再解析但很快发现事情没那么简单。用户需要在模型还没说完的时候就看到文字在动这是产品体验的底线可一旦把显示和解析混在一个回调里处理你会发现显示层能接受半句话解析层却不能接受半个 JSON。这两条路径从一开始就该分开设计后面我在第六部分会给出完整拆分方案。1.2 chunk 边界与 UTF-8祸根在一开始就埋下了第一次写流式的人还会栽在另一个更隐蔽的坑里多字节字符被从中间切断。比如上海的海在 UTF-8 编码下占三个字节如果某个 chunk 恰好只包含它的前两个字节直接decoder.decode(value)就会得到一堆。解决方式其实一行代码let text ; while (true) { const { done, value } await reader.read(); if (done) break; text decoder.decode(value, { stream: true }); } text decoder.decode(); // 冲刷缓冲区把残留的半截字符收尾关键是这个{ stream: true }选项。它的作用是把解码器内部残余字节缓存起来等下一个 chunk 到达时再拼接成完整字符。很多人只看到文档里写了这个参数不知道它到底解决什么问题——它就是专门处理字节流可以从任意位切割这件事的。不过要牢记TextDecoder只解决编码层面的完整性它不会帮你把{city:上补成完整 JSON。于是问题依然存在我们拿到了一长串合法字符但里面的结构化数据是残缺的。这就是流式输出和结构化解析最核心的矛盾展示层需要增量解析层需要完整。想通这一点后面所有方案都是围绕如何调和这个矛盾展开的。2. JSON 天生脆弱把解析策略拆成攒齐和边收边补两层JSON 协议在设计时从来就没考虑过流式增量解析。它的合法性判断依赖你必须读到最后一个}才知道整体对不对中途任何一个字符错位整段全废。而流式场景对错误容忍度的要求又特别高所以第一步不是去找某个神奇库而是先调整架构预期。2.1 攒齐再解析 vs 边流边补两条路线的取舍我见过不少团队在这里走极端。一端是永远不解析等done之后一口气处理实现简单缺点也明显前端只是在假装流式因为整个 JSON 要在最后几千毫秒里才能被解析中间做不了任何联动。另一端是追求极端实时恨不得模型每生成一个字就更新一次结构化数据结果被各种边界 case 折磨到怀疑人生。我的建议是分场景。如果只是给前端展示用流式负责把字打出来攒齐后统一解析完全够用代码量最少也最好维护。如果确实需要边生成边渲染图表、表格、卡片那就必须引入增量解析但这不意味着要上一个完整的流式 JSON 解析器。很多实际场景用先攒一个局部 buffer每次只对 buffer 里完整的复合结构做提取就够了比如 buffer 里出现了}就尝试JSON.parse一次失败就继续等。这里有一个很实用的经验给解析操作加一个频率下限。不要每个 chunk 都去试解析而是用节流控制在 100~200 毫秒一次。流式场景下 token 到达速率一般不会太快100 毫秒的延迟用户完全感知不到却能省掉一大批无意义的解析失败和异常日志团队排障时也会清爽很多。2.2 兜底修复jsonrepair 与 partial-json-parser 对比决定边收边补之后光靠JSON.parse不够因为它在残缺 JSON 面前是一票否决。这里有两个主流思路我结合自己的使用体验整理成表方案原理适用场景短板攒齐后JSON.parse完整数据一次性解析简单场景、低并发中间完全拿不到结构化数据jsonrepair对残缺/非法 JSON 做启发式修复差一点就能解析的脏数据结构缺失严重时无能为力partial-json-parser流式 token 级解析返回已完整部分需要提前渲染嵌套结构实现复杂、边界 case 多JSON Mode / 约束解码服务端解码器强制产出合法 JSON后端可控、追求稳定依赖云端服务或 vLLM 等框架jsonrepair是我放在最后一层兜底用的它的修复能力覆盖了模型输出里绝大多数脏数据多余逗号、缺少括号、键名没引号、字符串里裸单引号、末尾被截断的数组/对象等等。用法非常简单import { repair } from jsonrepair; function safeParse(raw) { try { return { ok: true, data: JSON.parse(raw) }; } catch { try { return { ok: true, data: JSON.parse(repair(raw)) }; } catch { return { ok: false, error: json-repair-failed }; } } }注意我是整段攒齐之后才做 repair而不是流式过程中做。原因很直白jsonrepair面向的是基本完整的残缺数据如果数据只流到一半缺失的可能是整整一个对象任何启发式都不可能猜出模型后面要说什么硬修只会产出更离谱的错误数据。所以流式过程中不要修等流结束再修partial-json-parser这种流式方案则用来满足中间态渲染两者不是替代关系而是各自管好各自那一段。2.3 治本的一招JSON Mode 与约束解码前端写得再花哨都不如从源头掐断问题。如果后端是自己部署的模型比如 vLLM 或本地 Ollama优先开启约束解码。这也是vllm部署大模型相关话题里被反复提到的 key pointvLLM 的guided_json可以让模型在解码阶段就按照给定 JSON Schema 逐 token 约束输出把合法性前置到 token 采样阶段从机制上杜绝语法错误。from vllm import SamplingParams sampling_params SamplingParams( temperature0, guided_json{ type: object, properties: { city: {type: string}, temperature: {type: number}, }, required: [city, temperature], }, ) outputs llm.generate(prompt, sampling_params)如果调用云端大模型OpenAI 和 Anthropic 也都有 JSON modeOpenAI 是response_format: { type: json_object }Anthropic 类似。这类模式的核心价值是让模型在解码每个 token 时只能选择能继续构成合法 JSON的 token从源头消灭不完整和语法错误。但这里必须泼一盆冷水JSON Mode 只保证语法合法不保证结构正确。模型完全可以给你返回一个合法 JSON——比如{city: 123}——但你的业务要求 city 是字符串。JSON Mode 约束的是形式约束不了语义。这就会把问题带到更深的层次我们需要在能解析和符合预期之间加一道闸这就是第三部分要讲的 Zod。3. Zod 兜底JSON.parse 成功不等于数据能用很多工程团队把能 JSON.parse 出来当成拿到了结构化数据然后一头扎进data.city.someField直到线上暴露出undefined is not a function才意识到问题。大模型不是遵守类型契约的开发者它生成的字段随时可能缺、多、类型错。所以解析成功之后必须立刻做 schema 校验。3.1 为什么结构校验必须独立于语法解析我见过一个真实案例某个 Agent 应用让模型返回工具执行结果模型在正常输出之外加了一个多余的顶层字段导致前端按字段名取值永远拿到 undefined而且因为 JSON 本身合法后端日志里根本查不到异常。这就是典型的语法合法但结构不对。如果一开始就引入 schema 校验这种问题在上游就直接被拦住了。在 Node/TypeScript 生态里我首选 Zod原因有三。一是和 TypeScript 类型推断无缝衔接z.infertypeof Schema直接得到静态类型一份 schema 两处用二是safeParse不抛异常错误信息结构规整方便构建修正提示三是生态成熟和 OpenAPI、tRPC 都能互通。如果后端是 Python同等定位是 Pydantic思路完全一致。下面这段是给 Tool Calls 参数校验用的 Zod schema 示例后面第四部分还会继续用import { z } from zod; const GetWeatherArgs z.object({ city: z.string().min(1, 城市名不能为空), days: z.number().int().min(1).max(7).optional(), unit: z.enum([celsius, fahrenheit]).default(celsius), }); type GetWeatherArgs z.infertypeof GetWeatherArgs;注意unit用了default(celsius)这个能力很关键。模型经常漏掉可选项如果没有默认值你就得在业务代码里到处写args.unit ?? celsius有了 default校验通过后的数据就是自带默认值的最终值能少掉一大片判空代码。3.2 safeParse 不等于吞错误错误信息要变成修正弹药safeParse的返回值是一个 discriminated unionconst result GetWeatherArgs.safeParse(rawJson); if (!result.success) { // result.error 是 ZodError里面是 issues 数组 const summary result.error.issues .map((issue) 路径 ${issue.path.join(.)}${issue.message}) .join(); console.error(参数校验失败:, summary); } else { // result.data 已经是类型安全的数据 const { city, days, unit } result.data; }我特别想强调不要吞错误。很多同学校验失败就直接 return null前端弹一个解析失败的框这是最浪费的做法。Zod 给出的错误信息是一份非常宝贵的修正指引路径 city城市名不能为空这种信息完全可以原样拼到下一轮对话里让模型自己把输出改对。这是自修正循环能生效的唯一前提——模型必须知道它错在哪。3.3 自修正循环让模型自己把输出改对流式输出加大模型本身存在的偶发错误决定了你不可能要求一次生成、一次成功。行业里最务实的做法是给模型一两次纠错机会。核心逻辑是一段循环async function generateValidatedT( messages: ChatMessage[], schema: z.ZodTypeT, maxAttempts 3 ): PromiseT { for (let attempt 0; attempt maxAttempts; attempt) { const raw await streamOnce(messages); const json safeParse(raw); // 内含 jsonrepair 兜底 if (!json.ok) { messages.push({ role: assistant, content: raw }); messages.push({ role: user, content: 你的上一条输出无法被解析为合法 JSON请只输出原始 JSON不要任何说明文字。报错${json.error}, }); continue; } const checked schema.safeParse(json.data); if (checked.success) return checked.data; messages.push({ role: assistant, content: raw }); messages.push({ role: user, content: 你输出的内容通过了 JSON 解析但不符合要求。错误明细${summarizeZodError(checked.error)}。请重新生成。, }); } throw new Error(超过最大重试次数); }这里有三个必须注意的细节。第一每次修正尝试要把上一条原文以assistant消息形式放回对话历史模型才能定位到自己刚才的输出第二修正提示必须具体到字段和原因笼统说你错了基本没有效果第三maxAttempts一定设上限我默认给 3 次超过就降级到用户可见的兜底提示避免模型陷入无限自我否定。另外我还会记录每一轮失败原因这些日志是最便宜的回归测试集攒多了之后你会非常清楚自己接的模型容易在哪里翻车。4. Tool Calls 全面实战流式增量拼装与参数级校验聊完 JSON 和 Zod接下来是重头戏Tool Calls也常叫 Function Calling。它和让模型返回一段 JSON最大的区别在于Tool Calls 是模型协议层面对调用外部函数的一等公民支持而不是纯文本约定。很多教程只教你怎么发一次请求但流式场景下的 Tool Calls 有很多自己的坑尤其是增量拼装。4.1 Tool Calls 本质上是一条独立通道先看一次非流式响应里 Tool Calls 长什么样{ choices: [{ finish_reason: tool_calls, message: { role: assistant, content: null, tool_calls: [{ id: call_abc123, type: function, function: { name: get_weather, arguments: {\city\:\上海\,\days\:3} } }] } }] }注意三个关键点。第一content是null当模型决定调用工具时通常不再输出给用户的正文第二arguments是一个字符串化的 JSON你拿到之后还得再JSON.parse一次这就是双重解析的由来第三finish_reason是tool_calls业务层要靠它判断接下来该执行工具而不是继续对话。这三个点都理解到位再写流式处理才不会慌。4.2 流式增量拼装千万不要在 delta 上直接 JSON.parse流式响应里Tool Calls 的信息是碎片化推送的。OpenAI 兼容协议中每个 chunk 的delta里可能只有tool_calls数组的某个片段而且按 index 区分{choices:[{delta:{tool_calls:[{index:0,id:call_abc123,function:{name:get_weather,arguments:}}]}}]} {choices:[{delta:{tool_calls:[{index:0,function:{arguments:{\city\:}}]}}]} {choices:[{delta:{tool_calls:[{index:0,function:{arguments:\上海\}}}]}}]}也就是说arguments会被拆成好几段字符串每段单独JSON.parse必然失败。正确姿势是按 index 把碎片拼起来等整个流走完、finish_reason变为tool_calls之后再整体解析const accum: Recordnumber, { id?: string; name?: string; args: string } {}; // 每个 chunk 到达时的处理 function onChunk(chunk: OpenAIStreamChunk) { const tc chunk.choices?.[0]?.delta?.tool_calls?.[0]; if (!tc) return; const index tc.index ?? 0; accum[index] ?? { args: }; if (tc.id) accum[index].id tc.id; if (tc.function?.name) accum[index].name tc.function.name; if (tc.function?.arguments) accum[index].args tc.function.arguments; } // 流结束时统一处理 function finalizeToolCalls() { return Object.entries(accum).map(([index, item]) { const parsed safeParse(item.args); // 还是那套 jsonrepair 兜底 return { index: Number(index), id: item.id, name: item.name, arguments: parsed }; }); }这段代码有两个容易写错的点。第一accum[index] ?? { args: }必须放在处理 fragments 之前初始化否则第一个带 arguments 的 chunk 到来时args还不存在会直接变成undefined...第二同一个 index 的id和name只在第一个 chunk 出现一次后面全是纯 arguments 片段所以每个字段要单独判断不能合并成一个if全包进去。4.3 tool_choice 强制单工具把复杂度降一半如果你的业务场景一次只会调用一个工具大多数 Agent 的初始化阶段都这样我强烈建议用tool_choice把行为钉死{ tools: [{ type: function, function: { name: get_weather, description: 查询指定城市未来若干天的天气, parameters: { type: object, properties: { city: { type: string }, days: { type: integer } }, required: [city] } } }], tool_choice: { type: function, function: { name: get_weather } } }这样模型不会在要不要调用工具和调用哪个工具之间犹豫流式响应里只会出现 index 0 这一条 tool_call增量拼装逻辑可以砍掉一半。等业务发展到模型需要自己选工具时再把tool_choice换成auto。到那时多个 tool_call 并行出现的概率会明显上升前面那套按 index 累加的代码才算真正派上用场。顺带提醒一句有时候你会看到模型给了finish_reason: tool_calls但某个 tool_call 的 arguments 拼完后依然是空字符串或残缺的。这种情况在低温度下也偶有发生最稳妥的处理是把这当成一次参数生成失败走 Zod 校验失败那条路回灌给模型重新生成而不是自作主张填默认值。4.4 工具参数的 Zod 化从 prompt 定义到运行时校验共用一份 SchemaTool Calls 的parameters本质上是一份 JSON Schema它既会被放进请求里指导模型也应该在运行时被拿来校验模型的实际输出。与其前后端各维护一份不如让一份 Zod schema 成为唯一数据源再生成 JSON Schema 给请求用。Zod 生态里有zod-to-json-schema这类工具import { zodToJsonSchema } from zod-to-json-schema; const GetWeatherArgs z.object({ city: z.string().min(1), days: z.number().int().min(1).max(7), }); // 一份 schema两种用法 const jsonSchema zodToJsonSchema(GetWeatherArgs, get_weather_args); const checked GetWeatherArgs.safeParse(parsedArguments);这个做法的收益在长期维护上尤其明显以后改参数、加字段、调枚举只需要改一处不会出现文档里写 city 必填运行时检查却漏了的脱节。运行时校验通过后再放心把checked.data传给实际的函数执行器。5. 热搜词实战标签返回未完整怎么处理做流式结构化解析的人应该都搜过标签返回未完整怎么处理这个问题。它最常见的来源是你用大模型输出 XML比如 Claude 系的output.../output风格或者提示词里让模型用tag.../tag包裹结构化内容结果流式输出时标签对一直闭合不完整。这里把处理思路完整梳理一遍。5.1 2025 年了为什么 XML 仍然有它的位置你可能会问JSON 都研究得那么透了为什么还要用 XML原因很现实对于长文本中嵌入结构化片段的场景JSON 的可读性和可修复性都远不如带标签的 XML。比如让模型返回一篇带多个小节的文章JSON 的转义符会铺满全文而sectiontitle.../titlebody.../body/section这种结构模型生成时容错率高得多人在调试时一眼也能看懂。Anthropic 官方也很早就推荐用 XML 标签来组织提示词输出Claude 系列模型对这种格式的遵循度普遍很高。所以在 RAG、长报告生成这类场景里XML 输出 标签解析是比JSON 输出 流式校验更省心的组合。代价就是得自己处理标签未完整这个流式带来的副产品。5.2 标签未完整的标准姿势事件式拼接而不是正则硬啃我见过最惨烈的写法是拿一个巨型正则去匹配流中每个 chunk 里的tag.../tag。结果不用想标签被切成两半时正则直接失效而且正则面对嵌套标签时极易产生灾难性回溯。正确做法是事件式拼接——维护一个缓冲区持续扫描遇到闭合的完整标签就提取并触发回调class XmlTagStreamParser { private buffer ; private stack: string[] []; feed(chunk: string, onComplete: (tag: string, text: string) void) { this.buffer chunk; // 持续扫描 buffer把完整的 tag.../tag 提取出来 // 简化示例只处理无嵌套的平铺标签 const regex /([a-zA-Z_][\w-]*)([\s\S]*?)\/\1/g; let m: RegExpExecArray | null; while ((m regex.exec(this.buffer)) ! null) { onComplete(m[1], m[2]); this.buffer this.buffer.slice(m.index m[0].length); regex.lastIndex 0; // buffer 被重写了重置游标 } } remaining(): string { return this.buffer; } }这个简化版只覆盖平铺标签。真实项目里如果标签会嵌套就得改成真正的栈式解析开标签入栈闭合标签出栈只有栈空时才算一个完整块。无论哪套写法核心原则一致不要试图解析当前这一小段而是维护累积缓冲区等完整结构出现再动手。这跟第四部分流式拼装 arguments 的思路是同一个世界观。另外别忽略一个细节模型输出的 XML 内容里可能有lt;这样的转义实体也可能出现花括号等和 JSON 冲突的字符。我的习惯是在提取完整标签文本后做一次解除转义 剔除控制字符的清洗再进业务逻辑否则后续入库或渲染时会出现莫名其妙的错位。5.3 流结束时仍然缺闭合标签怎么办最麻烦的情况是流结束了栈里还有未闭合的标签。我按优先级给出一套降级策略如果标签内容是完整可用的比如title周报/title已经出现只是后面还有个body没闭合直接丢弃未闭合标签采用已完整内容如果栈里只有一个标签且内容明确可以按人工规则补上闭合标签比如模型输出到一半的summary本周完成三项任务栈里压着summary那就补一个/summary再解析如果嵌套两层以上且截断位置模糊就不要硬补了优先保数据正确性把整块标记为不完整走重试或交给用户确认。这里最忌讳的是不管三七二十一全部补闭合标签。XML 标签的合法性跟 JSON 一样补错一个闭合位置结果比不补更糟。判断原则是内容语义已经明确的才值得补内容本身是截断的半句话补了也是垃圾数据。6. 把整个链路串起来生产级流水线与我沉淀的经验前面几部分是单点拆解这一部分把它们串成一条能在生产环境跑的流水线并把那些只会在真实部署中踩到的细节一一列出来。我会以解析一个需要调用工具的流式响应为例因为它覆盖了 JSON、Zod、Tool Calls 三条主线XML 场景按第五部分的处理方式接入即可。6.1 一条完整的解析流水线流式响应进入 - 展示层chunk 文本直接追加到界面TextDecoder stream: true - 解析层按 index 累加 tool_calls 的 delta - 流结束 - finish_reason tool_calls ? 拼装 arguments : 使用 content - JSON.parse失败走 jsonrepair 修复 - Zod safeParse - 失败把 Zod 错误回灌重试最多 3 次 - 成功调用真实工具把结果作为新消息继续对话这条流水线的关键设计是展示层和解析层完全分离。展示层永远不会因为 JSON 解析失败而卡顿解析层也不会因为要迁就展示而被迫处理半截数据。两边的关注点完全不同显示关心快不快、顺不顺解析关心对不对、全不全。6.2 失败场景与降级策略对照表我把实际运行中遇到的高频失败场景整理成一张表建议直接贴到团队文档里失败场景表现处理策略流式 chunk 乱码中文变成TextDecoder加{ stream: true }结束后再 flush 一次积累的 JSON 不完整Unexpected end of JSON input流结束后用jsonrepair修复修复失败标记不完整JSON 合法但 Schema 不符字段缺失/类型错误ZodsafeParse拦截错误信息回灌重试tool_calls 的 arguments 残缺拼完仍是空/半截按 index 拼装后仍失败走参数重生成兜底XML 标签未闭合栈里有剩余标签按内容语义明确才补原则分层降级超过重试上限连续 3 次失败返回用户可见的兜底提示记录日志这张表的价值不在于处理策略那一列有多新奇而在于每一行的判定条件都很具体新接手的人不用靠猜就知道该走哪条路。6.3 踩坑之后留下的个人心得这几条是我自己反复踩过之后写进团队规范里的不保证绝对正确但至少能帮你少摔几次。第一永远给流式链路上限。无论是 token 数、buffer 大小还是重试次数都必须有硬上限。模型偶发话痨是常态没有上限的循环会在某次线上事故里给你上一课。第二日志里永远保留原始输出。我已经数不清有多少次靠原始 JSON 长什么样才定位到是前端拼装错了还是模型输出错了。第三接入新模型前先跑一轮坏样本回归。把你积攒的失败案例喂给新模型看它是不是同样翻车能提前发现一个模型的性格缺陷比上线后再救火省心得多。第四修正提示一定要具体。我观察到的规律是告诉模型你的 city 字段不是字符串比告诉它你返回的数据不符合要求有效得多成功率能差出一大截。说到底流式输出加结构化解析这件事本质是在跟不确定性共存。模型不会因为你在前端写了更漂亮的代码就变得百分之百可靠但你可以通过约束、校验、回灌、降级这一整套机制把不可靠性控制在产品可接受的范围内。我现在的默认态度是把每一次解析都当成可能失败来处理代码写得更悲观一点线上反而更稳。希望这份实战记录能帮你少走一些我走过的弯路。
