别等模型说完再解析：Node.js 流式结构化输出的增量解析

大模型的 Structured Output 在流式场景里，本质是「一段还在变长的 JSON 文本」，不是等 finish 才突然出现的完整对象。本文用 Node.js 把半截 JSON 补成可 JSON.parse 的快照，让下游尽早用上已经成型的字段。示例可直接运行；接到 Chat Completions 时只用官方文档里真实存在的字段，不拿虚构数字证明收益。

为什么要在流上解析

服务端若坚持等 finish_reason === "stop" 再一次性 JSON.parse，用户看到的就是：空白，然后整块结果一起跳出。纯聊天往往可以接受；「抽取字段、填表、商品列表、分步确认」这类结构化任务则很不合适。模型写 JSON 时通常先给出靠前的键，数组也是一个元素接着一个元素往外推。这些前缀里已经有能用的数据，只是文本还不能直接 parse。

增量解析的目标很具体：每来一批 token，就得到「当前能确定」的对象。标题字段一旦出现就可以渲染；items 每多一个完整元素，列表就多一行。中间快照允许缺字段、允许字符串还没写完；最后一次快照应与整段 JSON 的解析结果一致。

需要把预期说清楚：这不会让模型少生成 token，也不会让推理更快。它只是把「已经生成、已经到达」的那部分提前交给业务。收益来自更早开始用，而不是更早结束生成。若业务必须等齐所有字段才能决策（例如要校验合计金额），那仍该等流结束，增量层最多用来做 UI 占位。

流式 Structured Output 实际是什么

以 OpenAI Chat Completions 的流式接口为例：请求里设 stream: true 之后，每个 chunk 里业务真正要用的是 choices[0].delta.content。它是字符串碎片，不是按字段拆好的对象。即使加上 response_format: { type: "json_object" }，传输形态也不会变成「字段事件流」——约束的是最终文本应是一个 JSON 对象，不是把 JSON 的结构映射成协议帧。

因此「Structured Output」解决的是模型输出形状，增量解析是你在 Node.js 进程里自己做的事。其它提供商大多同样给出文本 delta，下面的解析层可以共用；本文不展开各家 SDK 的差异，避免把示例绑死在某一版客户端上。

有人会问：既然最终是列表，为什么不让模型改吐 JSONL，一行一个对象？可以，而且行级协议对「纯列表流」更省事——读到换行就 parse 一行。但 schema 是嵌套对象、字段有前后依赖、或必须遵守模型侧的 json_object / json_schema 时，你拿到的仍是单块 JSON。本文针对这一种。另外，json_object 只保证「是对象」，不保证字段名和类型符合你的业务定义；严格 schema 要看当前模型文档里的 Structured Outputs，那是另一层约束，和「流仍是文本碎片」不冲突。

把半截 JSON 补成对象

半截文本经常长这样：{"title":"降噪耳机","items":[{"id":1,"name":"A"},{"id":。直接 JSON.parse 会抛错。它几乎总是「合法 JSON 写到一半」：括号和引号尚未闭合，或者最后一个键、冒号、字面量只写了前缀。

处理策略可以很朴素，不必先上状态机：

先按原样 parse，成功就用。
失败则从尾部往回切；每切一截，补上未闭合的引号和括号，再 parse。
切到能 parse 为止。整段都救不回来就返回 undefined，等更多 token。

从尾部回退会顺手丢掉：不完整的键、悬空冒号、半截 true / null、尾随逗号。对象里已经写完的键值、数组里已经写完的元素会留下。结构化输出通常不大，反复 JSON.parse 足够清楚，也够用。若你要对几十兆的 JSON 做真正的流式扫描，应换专门的流式解析器，那是另一类问题，不要把下面的回退算法套过去。

未闭合字符串要单独看待。允许补全引号时，{"title":"降噪 会得到 { title: "降噪" }，适合 UI 逐字显示。这种中间态不要写进数据库，也不要当作 RPC 入参。流结束后，用未修复的原文再 parse 一次，作为最终结果；修复逻辑只服务于过程快照。

回退补全的边界

回退不是万能的。前缀若已经不是合法 JSON 的前缀（模型在 json 模式关闭时先说了一段话，或夹了 markdown 围栏），切尾巴也救不回来。下面的示例假设缓冲里已经是 JSON 文本——这在 json_object 下通常成立。若你无法打开 JSON 模式，需要先从全文里截出第一个 { 或 [，那是预处理，不是解析器该偷偷猜的。

可运行示例：从 token 流到对象快照

下面是一份单文件示例，无第三方依赖。保存为 stream-structured.mjs 后执行 node stream-structured.mjs 即可。它用定时器把一段完整 JSON 切成小片，模拟模型的 token 流；解析函数接到真实 delta.content 时用法相同。

// stream-structured.mjs
// 运行：node stream-structured.mjs

function closeBrackets(source) {
  let inString = false;
  let escape = false;
  const stack = [];

  for (const ch of source) {
    if (inString) {
      if (escape) {
        escape = false;
        continue;
      }
      if (ch === "\\") {
        escape = true;
        continue;
      }
      if (ch === '"') inString = false;
      continue;
    }
    if (ch === '"') {
      inString = true;
      continue;
    }
    if (ch === "{") stack.push("}");
    else if (ch === "[") stack.push("]");
    else if (ch === "}" || ch === "]") {
      const expected = stack.pop();
      if (expected !== ch) return null;
    }
  }

  // 末尾落在反斜杠上，补引号会把转义吃掉，这一刀不可用
  if (escape) return null;

  let out = source;
  if (inString) out += '"';
  for (let i = stack.length - 1; i >= 0; i--) out += stack[i];
  return out;
}

export function parsePartialJson(text, { allowPartialStrings = true } = {}) {
  const source = String(text).trim();
  if (!source) return undefined;

  try {
    return JSON.parse(source);
  } catch {
    // 走回退
  }

  for (let end = source.length; end > 0; end--) {
    const slice = source.slice(0, end);
    if (!allowPartialStrings) {
      // 禁止半截字符串时，若当前切片落在字符串内则跳过
      const closed = closeBrackets(slice);
      if (closed === null) continue;
      // 用「是否补了引号」判断：closeBrackets 在 inString 时会多一个 "
      // 这里改为：若原切片无法在不补引号的情况下闭合，则跳过
    }
    const candidate = closeBrackets(slice);
    if (candidate === null) continue;
    if (!allowPartialStrings && candidate.length > slice.length && candidate[slice.length] === '"') {
      continue;
    }
    try {
      return JSON.parse(candidate);
    } catch {
      // 继续切掉最后一个字符
    }
  }
  return undefined;
}

async function* mockModelStream(fullText, chunkSize = 3, delayMs = 30) {
  for (let i = 0; i < fullText.length; i += chunkSize) {
    await new Promise((resolve) => setTimeout(resolve, delayMs));
    yield fullText.slice(i, i + chunkSize);
  }
}

async function consumeTextStream(textStream) {
  let buffer = "";
  let lastSerialized = "";
  const snapshots = [];

  for await (const delta of textStream) {
    buffer += delta;
    const snap = parsePartialJson(buffer);
    if (snap === undefined) continue;
    const serialized = JSON.stringify(snap);
    if (serialized === lastSerialized) continue;
    lastSerialized = serialized;
    snapshots.push(snap);
    const itemCount = Array.isArray(snap.items) ? snap.items.length : 0;
    console.log(`--- 快照 items=${itemCount} title=${JSON.stringify(snap.title ?? null)}`);
    console.log(JSON.stringify(snap, null, 2));
  }

  const finalObject = JSON.parse(buffer);
  console.log("--- 最终结果（原文 parse，未走修复）");
  console.log(JSON.stringify(finalObject, null, 2));
  return { snapshots, finalObject };
}

const fullJson = JSON.stringify({
  title: "办公室降噪耳机怎么选",
  summary: "先看主动降噪和佩戴重量，再看连接稳定性。",
  items: [
    { id: "a1", name: "头戴式 A", price: 899 },
    { id: "b2", name: "入耳式 B", price: 499 },
  ],
});

await consumeTextStream(mockModelStream(fullJson));

这段程序做了三件小事：把半截 JSON 补成对象、用序列化结果去掉重复快照、在模拟流上打印每次变化。运行后应看到 title 先出现（可能先是前几个字），随后 items 从 0 条变成 1 条再变成 2 条，而不是等全部结束才出现。allowPartialStrings: false 时，未写完的字符串不会进快照，适合下游不能接受半截文案的场景。

去重很重要。模型连续吐出一个很长的数字或空白时，修复后的对象可能不变；若每次 token 都触发一次数据库或一次前端推送，意义不大。用 JSON.stringify 比较的是值，不是对象身份。键的插入顺序在 V8 的 JSON.parse 下是稳定的，对这种用途够用。

接到真实模型流时的边界

模拟流和真实流的差异只在「字符串从哪来」。官方 Node SDK 中，chat.completions.create({ stream: true }) 返回异步可迭代对象，循环里累加 delta.content 即可。密钥走环境变量 OPENAI_API_KEY，不要写进仓库。下面函数假定上一节的 parsePartialJson 已在同一作用域（或从该文件 import）。

import OpenAI from "openai";
import { parsePartialJson } from "./stream-structured.mjs";

export async function consumeChatCompletionJson({ messages, model = "gpt-4o-mini" }) {
  const client = new OpenAI();
  const stream = await client.chat.completions.create({
    model,
    messages,
    stream: true,
    response_format: { type: "json_object" },
  });

  let buffer = "";
  let lastSerialized = "";
  const snapshots = [];

  for await (const chunk of stream) {
    buffer += chunk.choices[0]?.delta?.content ?? "";
    const snap = parsePartialJson(buffer);
    if (snap === undefined) continue;
    const serialized = JSON.stringify(snap);
    if (serialized === lastSerialized) continue;
    lastSerialized = serialized;
    snapshots.push(snap);
  }

  return { snapshots, finalObject: JSON.parse(buffer), raw: buffer };
}

几条实际约束，建议写进代码审查清单，而不是当作优化项事后补：

第一个 chunk 常常只有 role、没有 content，?? "" 即可，不要把 undefined 拼进缓冲。
json_object 不保证业务 schema。增量快照会继承缺失字段和类型漂移。需要严格 schema 时，使用当前官方文档中的 json_schema / Structured Outputs；即便如此，流仍然是文本碎片，解析器不用换。
中间快照只用于展示或预取。持久化、扣费、写主库，一律等流结束并用原文 JSON.parse。修复函数可能补出「看起来合法、业务上未完成」的对象。
未开 JSON 模式时，模型可能输出 markdown 围栏或前言。解析前要先截出 JSON 本体；上面的回退补全不会魔法般跳过人类可读的开场白。
解析失败时返回「还没有对象」，不要把半截字符串当对象传给下游。下游一旦开始按字段取值，调试成本会高于解析器本身。
回退算法的代价随缓冲长度上升。对常见的小对象够用；若你让模型一次吐出超大文档，应重新评估，而不是先假设它会成为热点。

总结

增量解析解决的是「完整 JSON 尚未结束，但前缀里已经有可用数据」。实现上不必先引入框架：累加文本、尝试 parse、失败则从尾部回退并补上引号和括号。把它接到 Chat Completions 的 delta.content 上即可，模拟流和真实流共用同一层。

记住两件事。第一，中间快照可以不完整，最终结果必须用原文再 parse 一次，修复逻辑不得成为写入路径的唯一来源。第二，模型侧的 JSON 模式约束的是最终文本，并不改变流的形状——你处理的始终是字符串。能早一点把已成型的字段交给 UI 或预取逻辑，这件事本身足够；把它说成降低了模型延迟，则不准确。

别等模型说完再解析：Node.js 流式结构化输出的增量解析

为什么要在流上解析

流式 Structured Output 实际是什么

把半截 JSON 补成对象

回退补全的边界

可运行示例：从 token 流到对象快照

接到真实模型流时的边界

总结

相关文章

别让代码智能体把终端当小说读：用 Node.js 解析 ANSI、表格与交互提示

别让文件工具逃出工作区：用 Node.js 做路径规范化与写入根限制

别把本地模型当成换地址的云 API：用 Node.js 管排队、窗口与内存

评论