Node.js LLM 提取：429 限流下的 Schema 校验与重试策略

当 LLM 返回 429 限流时，电商代码审查的结构化数据提取会危险地失败——但一个到达 Pull Request 的、违反 Schema 的发现结果更糟。它可能指向错误的文件、遗漏证据，或者把一个建议变成阻塞性问题。

简短回答：将每个 LLM 提取请求经过区域本地的队列， 用有限退避重试 429 响应， 且仅在返回的 JSON 通过严格 Schema 校验后才确认工作完成。对可延迟的仓库扫描使用批量 API， 而非对开发者等待阅读的反馈使用。

这个顺序很重要。退避保护容量，校验保护产品。

本构建日志假设已有一个同步提取器。迁移顺序是：Schema 校验门第一，队列执行第二，区域或批量分区最后。每个阶段都可以回滚而不削弱输出契约。

迁移从一种可发布状态开始

从重要的状态转换入手：发现结果要么经过校验并可发布，要么它还不是一个发现结果。HTTP 成功、可解析的 JSON 和非空数组都是中间观察结果。没有任何一个能获得标注 Pull Request 的权利。

对于这个系统，验收需要预期的 Schema 版本、与进入 worker 时相同的 change ID， 以及其 file、line、severity、summary 和 evidence 字段通过运行时检查的发现结果。这比 TypeScript 单独校验更严格。编译时类型无法说明进程运行期间收到的未知字节。

在队列之前写下失败账本。四个终态标签足以覆盖迁移：transport_rejected、retry_exhausted、schema_rejected 和 accepted。只有最后一个标签才能触发发布。将它们区分开来可以避免仪表盘上一堆"已完成"请求掩盖不可用的输出， 并且让 Schema 变更有一个可衡量的影响范围， 而不假装每个失败都来自限流。

在修改调度器之前，先在当前同步调用周围部署这个校验门。观察拒绝标签而不发布新行为， 将接受的输出与现有路径比较， 并为意外 Schema 拒绝设置回滚条件。这是改变架构的约束。队列吞吐量变得次要于那个转换的完整性。

阶段 1 替换充满希望的解析

有用的抽象是适配器， 而非散布在队列中的供应商 SDK。Worker 需要一个操作：接收文本并返回未知数据。其后的一切都是本地代码， 这使得正确性路径易于测试， 并将配置膨胀挡在调用点之外。

type Region = "us" | "eu";
type Severity = "info" | "warning" | "blocking";

type ReviewFinding = {
  file: string;
  line: number;
  severity: Severity;
  summary: string;
  evidence: string;
};

type ReviewResult = {
  schemaVersion: 1;
  changeId: string;
  findings: ReviewFinding[];
};

type ExtractionRequest = {
  changeId: string;
  region: Region;
  diff: string;
};

type ModelAdapter = {
  extract(input: string, signal: AbortSignal): Promise<unknown>;
};

const isString = (value: unknown): value is string =>
  typeof value === "string" && value.length > 0;

function isFinding(value: unknown): value is ReviewFinding {
  if (typeof value !== "object" || value === null) return false;
  const row = value as Record<string, unknown>;
  return (
    isString(row.file) &&
    Number.isInteger(row.line) &&
    (row.line as number) >= 1 &&
    ["info", "warning", "blocking"].includes(String(row.severity)) &&
    isString(row.summary) &&
    isString(row.evidence)
  );
}

function parseReviewResult(value: unknown, changeId: string): ReviewResult {
  if (typeof value !== "object" || value === null) {
    throw new Error("invalid_result: expected an object");
  }

  const row = value as Record<string, unknown>;
  if (
    row.schemaVersion !== 1 ||
    row.changeId !== changeId ||
    !Array.isArray(row.findings) ||
    !row.findings.every(isFinding)
  ) {
    throw new Error("invalid_result: schema or correlation mismatch");
  }

  return row as ReviewResult;
}

关联性检查很容易被忽略。没有它， 一个语法完美的响应仍可能在异步重试或批量协调后被挂到错误的 Change 上。因此 changeId 是接受输出的一部分， 而不是在模型边界消失的元数据。

考虑一个修改了税费显示和库存预留的假设结账 Diff。提取器返回两个格式良好的发现结果， 但它的 changeId 属于上一个提交， 因为异步结果按到达顺序被协调了。每个字段校验器都通过了。发布那个对象会把看起来准确的证据放到错误的代码上——这正是结构性正确性包含身份而非在 JSON 形状处停步的原因。Worker 拒绝整个对象， 记录 schema_rejected， 并保持发布不动；然后运营商可以比较提交的清单、返回的 ID 和 Schema 版本， 而无需猜测哪个队列位置移动了。这个示例不需要特殊的队列功能。它需要一个在迁移期间使用的每个调度器都能存活的契约。

这个校验器故意做得很无聊。对于更大的 Schema， 从一个签入的 Schema 定义生成运行时校验， 这样 TypeScript 类型、测试固件和生产校验门就不会各行其是。验收规则保持不变：未知进入；经过校验的域对象离开。

在这个阶段， 不要添加重试或区域路由。换一个边界， 运行契约固件， 并保留旧的发布开关作为回滚点。一次发布同时改变解析、调度和放置的迁移不会给被拒绝的发现结果留下任何清晰解释。

Node.js 如何通过 429 限流队列迁移 LLM 提取？

重试必须保留原始 item ID、区域、尝试次数和截止时间。它不能跳入另一条车道。它也不能在等待时持有并发许可证， 因为睡眠中的 Worker 会把瞬时限流变成自我造成的饥饿。

阶段 2 只将经过校验的请求移到队列后面， 并保持发布契约不变。在转移实时工作之前， 用相同的固定固件同时运行同步和队列调度器。比较目标是按 changeId 接受的发现结果， 而不仅仅是成功请求的相等数量。

以下是调度核心。适配器可以从它使用的任何传输层暴露类型化的限流错误；队列不需要知道供应商路由或响应形状。

class RateLimitError extends Error {
  constructor(readonly retryAfterMs?: number) {
    super("rate_limited");
  }
}

const delay = (milliseconds: number): Promise<void> =>
  new Promise((resolve) => setTimeout(resolve, milliseconds));

function retryDelayMs(attempt: number, retryAfterMs?: number): number {
  if (retryAfterMs !== undefined) return Math.min(retryAfterMs, 30_000);
  const ceiling = Math.min(500 * 2 ** attempt, 30_000);
  return Math.floor(Math.random() * ceiling);
}

async function runExtraction(
  request: ExtractionRequest,
  adapter: ModelAdapter,
  signal: AbortSignal,
): Promise<ReviewResult> {
  const maxAttempts = 5;

  for (let attempt = 0; attempt < maxAttempts; attempt += 1) {
    try {
      const raw = await adapter.extract(request.diff, signal);
      return parseReviewResult(raw, request.changeId);
    } catch (error) {
      const canRetry = error instanceof RateLimitError && attempt < maxAttempts - 1;
      if (!canRetry) throw error;
      await delay(retryDelayMs(attempt, error.retryAfterMs));
    }
  }

  throw new Error("retry_budget_exhausted");
}

五次尝试和 30 秒上限是示例策略值， 不是通用调优建议。用实际工作负载做基准测试。按车道记录队列等待、活跃时间、尝试次数、校验拒绝次数和端到端延迟； 仅看平均值会掩盖用户感受到的突发流量。我更关心 p95 队列等待和无效输出率， 而不是漂亮的每秒请求数， 因为这两个指标暴露了过载和正确性损失。

注意代码不重试什么：schema_rejected。盲目重放相同的无效提取会消耗容量而不会改变契约或输入。仅在修复 Prompt 和验收标准被单独测试的情况下， 才有条件地将该结果路由到有限的修复策略。否则， 显式失败， 并在适用的区域保留策略下保存原始响应以供诊断。

在美国、欧盟和批量交接期间，身份在哪里？

将入场、执行和验收视为独立状态。入出场决定代码 Diff 属于美国车道、欧盟车道还是延迟车道。执行拥有并发和 429 恢复。验收在所有地方应用相同的 JSON 契约。

在阶段 3 添加这些分区。安全的迁移键是 (region, changeId, schemaVersion)；它在调度机制改变时保持稳定， 因此排水期间的结果不会与更新的工作混淆。

区域选择必须在入队之前进行。不要因为某个队列恰好更短就把被拒绝的 EU 条目发送到 US Worker；运营便利不是数据位置策略。将 payload 存储、日志、重试元数据和结果处理保持在同一区域边界内。确切的法律和部署要求各不相同， 我不确定一个通用架构图能解决它们。一份有文档记录的数据流审查可以。

批量适用于调用方在代码审查交互期间不需要结果时：夜间仓库扫描、迁移审计或回填可以用延迟换取更平滑的入场。为每个输入提供一个稳定的 changeId， 持久化提交的清单， 并按 ID 而不是数组位置协调输出。部分完成必须可见。一个缺失的条目不能静默地将 1000 个变更的扫描变成 999 个表面上的成功。

当开发者在等待时、 当发现结果阻塞合并时、 或者当 Diff 在延迟结果返回前可能变陈旧时， 坚持使用交互式车道。问题是运维重复：实时队列和批量作业需要相同的 Schema 校验门、固件、区域规则和可观测性。如果批量路径有第二个解析器， 两条路径最终会对"有效"的含义产生分歧。

当上游系统无法安全协调异步 ID 时， 批量也不适用。先修复那个所有权边界。更快的批量提交不会修复模糊的身份。

发布门在自适应入场之前

第一个变更应该是每个区域的自适应入场， 由观察到的节流和队列延迟驱动， 而非单一硬编码的并发值。我会保持一个硬上限作为安全护栏， 当 429 响应上升时减少许可， 并逐步恢复它们。控制器需要阻尼； 一个过于积极的开环增减会振荡并制造突发。

接下来， 我会对输出契约进行版本控制， 并从代表性的电商变更中构建一个固定评估集：价格计算、库存预留、税费显示和结账状态转换。部署应在校验通过确切的结构检查后才能迁移流量。语义层面的审查质量需要单独的评分标准， 因为有效的 JSON 仍可能包含一个弱的发现结果。传输成功、 结构正确性和审查有用性是三个不同的信号。

局限性是存在的。队列不能创造供应商容量， Schema 校验不能证明一个发现结果是真实的， 区域车道本身也不能建立合规。对于小型低容量仓库， 三个持久化队列可能是多余的 machinery；进程内限流器加严格校验可能是更好的选择。对于大容量或受监管的工作， 持久化状态和明确的区域所有权证明了它们运维重量的合理性。这是我会用的决策线。

最终架构故意做得很朴素：分类、入队、执行、校验、发布。一次迁移一个动词。每个额外的层必须改善其中之一并在基准测试中证明它。如果做不到， 就删除配置。