Node.js 大目录 JSON 提取：分块策略与 Token 预算控制

Node.js 大目录 JSON 提取：分块策略与 Token 预算控制 | 前端进阶之旅

Full translation

完整中文译文

对于冗长的物流描述，简短的回答是：使用以检索优先的三遍流水线——在可测量的 token 预算内进行分块，用 embedding 生成候选证据列表，然后重新排序并只提取符合截止时间的段落。

对于混合产品目录，将检索优先作为默认方案。对于危险品、清关或合规记录等召回率比响应时间更重要的场景，保留"每个 chunk 都提取"作为备选方案。重要的一点不是选择哪个时髦的模型，而是明确的预算和证据追溯链。

Node.js 文本转 JSON 提取流水线在超时前应该做什么？

将超时视为需要分配的预算，而非最后才捕获的异常。长文档可能以两种独立方式失败：可能超出模型的输入限制，或者虽然没超出但剩余的墙上时钟时间太少无法完成有用工作。分块解决的是第一个约束。它不能自动解决第二个约束；每个 chunk 触发一个提取请求可能将 token 限制的修复变成延迟问题。

质量从输出契约开始。对于物流目录，先定义字段再选择 chunk 大小：SKU、重量、尺寸、危险品状态，以及支撑每个值的确切文本。一个语法有效但猜测量值的 JSON 对象，比一个标注 null 但保留了证据的部分对象更糟糕。提取后验证类型，在领域代码中拒绝不可能的组合，绝不让模型将缺失值静默转换为看似合理的值。延迟需要两个限制。为发送到提取阶段的段落设置 token 预算，为每个阶段设置时间预算。为 schema、指令、输出和传输留出余量。正确量取决于模型和分词器，所以我无法确定存在通用的 chunk 大小；用生产环境中实际使用的精确分词器和请求形状来测量。空白字符计数器适合测试编排，对生产环境的 token 计数不够精确。这些约束属于同一个设计评审，因为缩减所选证据可能改善延迟同时降低召回率，而允许更多段落可能以错过截止时间为代价恢复缺失的维度。在调优之前写下两个接受阈值。否则每次基准测试运行都可能在看到结果后切换指标而被宣布为胜利。

这就是基准测试发挥作用的地方。记录阶段耗时、输入 token 数、候选数量、所选 chunk ID、验证失败次数以及端到端截止时间错过次数。在标注目录集上比较 p50 和 p95 延迟以及字段级精确率和召回率。仅仅平均值会掩盖导致原始超时的过大描述。

第一个决策标准是证据召回率：短名单是否保留了请求字段所需的所有段落？第二个是检索后的剩余提取时间。如果短名单召回率低，在交换提取器之前改进 chunk 边界、查询或重排。如果检索消耗了大部分截止时间，在摄取期间预计算文档 embedding 并限制重排工作。这些诊断导致不同的修复方案，这就是为什么单一的端到端计时器是薄弱的可观测性。

重试需要同样的纪律。RFC 9110 定义了幂等方法并解释了客户端在通信故障后何时可以自动重试请求。保持稳定的 job ID 并使结果写入具有幂等性，这样重试不会创建两个目录版本。不要盲目重试每次超时：原始计算可能仍然完成了，另一条昂贵的请求可能使拥塞更严重。

用 TypeScript 实现三遍

下面的示例特意设计为适配器形状。它可以在 TypeScript 运行时下无包运行，使用确定性哈希向量使控制流可测试，并将生产环境的 embedding 和提取客户端留在核心流水线之外。这个玩具向量捕获 token 重叠而非语义。在生产环境中将其替换为与部署模型匹配的实现在空白 token 计数器旁使用。

每个 chunk 都带有其 ID 和原始文本。这个小选择使对账可审计并阻止重排将证据变成匿名字符串。

type CatalogRecord = {
  sku: string | null;
  massKg: number | null;
  dimensionsCm: [number, number, number] | null;
  hazmat: boolean | null;
  evidence: Array<{ chunkId: number; text: string }>;
};

type Chunk = { id: number; text: string; tokens: number };
type RankedChunk = Chunk & { score: number };

const countTokens = (text: string): number =>
  text.trim() === "" ? 0 : text.trim().split(/\s+/).length;

function chunkText(text: string, maxTokens: number, overlapTokens: number): Chunk[] {
  if (maxTokens <= overlapTokens || overlapTokens < 0) {
    throw new Error("INVALID_CHUNK_BUDGET");
  }

  const words = text.trim().split(/\s+/);
  const chunks: Chunk[] = [];
  const step = maxTokens - overlapTokens;

  for (let start = 0, id = 0; start < words.length; start += step, id += 1) {
    const chunkText = words.slice(start, start + maxTokens).join(" ");
    chunks.push({ id, text: chunkText, tokens: countTokens(chunkText) });
    if (start + maxTokens >= words.length) break;
  }

  return chunks;
}

// Deterministic test adapter. Use a model-compatible embedding in production.
function testEmbedding(text: string, dimensions = 64): number[] {
  const vector = Array<number>(dimensions).fill(0);
  for (const token of text.toLowerCase().match(/[a-z0-9.]+/g) ?? []) {
    let hash = 2166136261;
    for (const char of token) {
      hash ^= char.charCodeAt(0);
      hash = Math.imul(hash, 16777619);
    }
    vector[(hash >>> 0) % dimensions] += 1;
  }
  const norm = Math.hypot(...vector) || 1;
  return vector.map((value) => value / norm);
}

function cosine(a: number[], b: number[]): number {
  return a.reduce((sum, value, index) => sum + value * b[index], 0);
}

function retrieve(chunks: Chunk[], query: string, limit: number): RankedChunk[] {
  const queryVector = testEmbedding(query);
  return chunks
    .map((chunk) => ({
      ...chunk,
      score: cosine(queryVector, testEmbedding(chunk.text)),
    }))
    .sort((a, b) => b.score - a.score)
    .slice(0, limit);
}

function rerank(candidates: RankedChunk[], terms: string[]): RankedChunk[] {
  const normalizedTerms = terms.map((term) => term.toLowerCase());
  return candidates
    .map((chunk) => {
      const haystack = chunk.text.toLowerCase();
      const lexicalHits = normalizedTerms.filter((term) => haystack.includes(term)).length;
      return { ...chunk, score: chunk.score + lexicalHits / normalizedTerms.length };
    })
    .sort((a, b) => b.score - a.score);
}

function extractDeterministically(chunks: RankedChunk[]): CatalogRecord {
  const joined = chunks.map((chunk) => chunk.text).join("\n");
  const sku = joined.match(/\bSKU[:\s]+([A-Z0-9-]+)/i)?.[1] ?? null;
  const mass = joined.match(/\b(?:mass|weight)[:\s]+([0-9.]+)\s*kg\b/i)?.[1];
  const dimensions = joined.match(
    /\b(?:dimensions|size)[:\s]+([0-9.]+)\s*x\s*([0-9.]+)\s*x\s*([0-9.]+)\s*cm\b/i,
  );
  const hazmatMatch = joined.match(/\bhazmat[:\s]+(yes|no)\b/i)?.[1];

  return {
    sku,
    massKg: mass === undefined ? null : Number(mass),
    dimensionsCm: dimensions
      ? [Number(dimensions[1]), Number(dimensions[2]), Number(dimensions[3])]
      : null,
    hazmat: hazmatMatch === undefined ? null : hazmatMatch.toLowerCase() === "yes",
    evidence: chunks.map((chunk) => ({ chunkId: chunk.id, text: chunk.text })),
  };
}

function extractCatalogRecord(description: string): CatalogRecord {
  const chunks = chunkText(description, 24, 6);
  const candidates = retrieve(
    chunks,
    "SKU mass weight dimensions size hazmat dangerous goods",
    4,
  );
  const selected = rerank(candidates, ["sku", "kg", "cm", "hazmat"]).slice(0, 2);
  const extractionTokens = selected.reduce((sum, chunk) => sum + chunk.tokens, 0);

  if (extractionTokens > 48) throw new Error("EXTRACTION_TOKEN_BUDGET_EXCEEDED");
  return extractDeterministically(selected);
}

const description = [
  "Warehouse note: blue replacement pump for regional depot stock.",
  "SKU: PUMP-4815. Packed mass: 12.5 kg.",
  "Dimensions: 42 x 3",
];

尽管代码保持小巧，但这里有三遍。chunkText 创建有界限的候选。retrieve廉价地缩小范围。rerank 在提取前对含字段文本花费更多注意力。在生产环境中，最终适配器可以向模型请求结构化 JSON，但其响应仍需要 schema 验证和针对 chunk ID 的对账。

不要通过复制来调优示例中的数字。它们的存在是为了使测试可执行。构建一个包含以下内容的小评估集：简洁描述、单位变体、重复属性、矛盾的供应商备注，以及跨 chunk 边界的重要字段。扫掠 chunk 大小、重叠、检索数量、重排数量和提取预算。然后选择能清除运营实际所需字段召回阈值的 fastest configuration。

棘手的失败是矛盾而非解析。描述可能在供应商块中说 mass: 12.5 kg，随后在 shipping weight: 14 kg。两个值在不同的字段定义下都可能有效。保留两个段落，精确界定目标字段，将未解决的冲突发送到审核而不是让 chunk 顺序决定。这会在精确的静默目录腐败可能进入系统的时刻增加摩擦——有用的摩擦。

Embedding 和重排何时输给另一种方法？

当每个条款都可能改变结果时，检索优先并不适用。危险品申报、海关文件和合同限制通常需要全面覆盖。使用"每个 chunk 都提取"、合并类型化部分结果并运行矛盾检查，即使延迟随文档长度增加。问题是操作层面的：更多调用意味着更多部分完成的机会，因此为每个 chunk 设置检查点并使聚合可重复。

当输入严格有界、舒适地符合测量的请求预算并在负载测试中达到延迟目标时，坚持一个完整 prompt。额外的分块和 embedding 基础设施会增加配置、追踪表面和故障状态，而不会获得有用的余量。这里的 DX 很重要。如果没人能解释哪个指标移动了每个阈值，那么具有五个可调阈值的流水线就是一项负债。

当精确标识符携带信号时，embedding 也会失效。SKU、UN 编号、尺寸和单位标记非常适合词法检索或确定性解析器。混合短名单可以取精确匹配候选和向量候选的并集，然后进行一次重排。这防止了语义弱但操作关键字符串因向量相似性低而消失。

检索优先会丢失证据吗？

会的。该设计以较低的召回率上限换取有界限的提取工作。公开测量这种损失。如果标注评估表明短名单丢弃了所需证据，提高候选预算、改进每个字段的查询、添加精确匹配或选择穷举备选方案。不要通过更大的上下文窗口来隐藏权衡。

部署规则很简洁：只有在同一组保留文档上同时通过字段级质量和截止时间合规两个关卡时，才发布配置。在发布后监控这些关卡，因为目录混合会发生变化。独立于模型适配器回滚阈值；将它们耦合会将常规调优更改变成完整的集成发布。

RFC 9110: HTTP Semantics，包括幂等方法和重试注意事项：https://www.rfc-editor.org/rfc/rfc9110

LiteLLM，一个开源自托管 LLM 网关，可作为可能的适配器边界之一：https://github.com/BerriAI/litellm

HTTP 方法语义和重试行为：https://www.rfc-editor.org/rfc/rfc9110

用于学习适配器边界的开源网关实现：https://github.com/BerriAI/litellm