别让一次 429 毁掉整批任务：写一个能重试的 AI API 客户端

30 秒速查

• 只重试 429、5xx、超时和网络错误
• 3 次上限 + 指数退避 + 随机抖动
• 总预算 deadlineMs，每次重试重算剩余时间
• JSON 解析失败当失败处理，日志记 token 成本

这篇教程解决什么问题

批量跑 AI 任务时，最常见的失败不是模型不会回答，而是：

• 请求超时；
• 触发限流，返回 429；
• 服务端偶发 5xx；
• 返回了 JSON，但字段缺失或类型不对。

如果每失败一次就整批重跑，时间和费用都会翻倍。这篇写一个最小但够用的客户端：有超时、有限重试、能读 Retry-After、会校验结果，并记录成本。

设计原则

先定四条规则，代码只是它们的实现：

1. 只重试可恢复的错误：网络错误、超时、429、5xx。400、401、403 不重试。
2. 重试次数有限：默认 3 次，且总耗时不超过上限。
3. 退避带随机抖动：避免所有任务在同一时刻再次撞上限流。
4. 成功响应也要校验：能解析不等于结构正确。

完整脚本

新建 scripts/llm-client/client.mjs，下面是全文——超时、重试、退避、截止时间、校验、成本日志、并发池都在里面，后面几节逐个说明为什么这么写：

export const DEFAULT_TIMEOUT_MS = 60_000;
export const DEFAULT_MAX_RETRIES = 3;
export const DEFAULT_BASE_DELAY_MS = 800;

function sleep(ms) {
  return new Promise((resolve) => setTimeout(resolve, ms));
}

export function isRetryableStatus(status) {
  return status === 429 || (status >= 500 && status <= 599);
}

export function parseRetryAfter(value) {
  if (!value) return null;

  const seconds = Number(value);
  if (Number.isFinite(seconds)) return seconds * 1000;

  const date = Date.parse(value);
  if (!Number.isNaN(date)) return Math.max(0, date - Date.now());

  return null;
}

export async function callLLM({
  baseUrl,
  apiKey,
  body,
  timeoutMs = DEFAULT_TIMEOUT_MS,
  maxRetries = DEFAULT_MAX_RETRIES,
  baseDelayMs = DEFAULT_BASE_DELAY_MS,
  deadlineMs,
}) {
  const startedAt = Date.now();
  let lastError;

  for (let attempt = 0; attempt <= maxRetries; attempt += 1) {
    const attemptStartedAt = Date.now();
    const remaining = deadlineMs
      ? deadlineMs - (Date.now() - startedAt)
      : Number.POSITIVE_INFINITY;

    if (remaining <= 0) {
      throw lastError ?? new Error(`deadline of ${deadlineMs}ms exceeded`);
    }

    try {
      const response = await fetch(`${baseUrl}/chat/completions`, {
        method: "POST",
        headers: {
          "content-type": "application/json",
          authorization: `Bearer ${apiKey}`,
        },
        body: JSON.stringify(body),
        signal: AbortSignal.timeout(Math.min(timeoutMs, remaining)),
      });

      if (!response.ok) {
        const text = await response.text();
        const error = new Error(`HTTP ${response.status}: ${text.slice(0, 300)}`);
        error.status = response.status;
        error.retryAfterMs = parseRetryAfter(response.headers.get("retry-after"));
        throw error;
      }

      const data = await response.json();

      return {
        data,
        meta: {
          attempt,
          durationMs: Date.now() - attemptStartedAt,
          usage: data.usage ?? null,
        },
      };
    } catch (error) {
      lastError = error;

      const retryable =
        error.name === "TimeoutError" ||
        error.name === "AbortError" ||
        error instanceof TypeError ||
        (typeof error.status === "number" && isRetryableStatus(error.status));

      if (!retryable || attempt === maxRetries) {
        throw error;
      }

      const exponential = baseDelayMs * 2 ** attempt;
      const jitter = Math.random() * baseDelayMs;
      const waitMs = error.retryAfterMs ?? exponential + jitter;

      if (deadlineMs && Date.now() - startedAt + waitMs >= deadlineMs) {
        throw error;
      }

      console.warn(
        `attempt ${attempt + 1} failed (${error.message}), retrying in ${Math.round(waitMs)}ms`,
      );

      await sleep(waitMs);
    }
  }

  throw lastError;
}

export function callLLMWithDeadline({ deadlineMs = 180_000, ...options }) {
  return callLLM({ ...options, deadlineMs });
}

export function assertShape(value, shape, path = "root") {
  for (const [key, type] of Object.entries(shape)) {
    const actual = value?.[key];
    const ok =
      type === "array"
        ? Array.isArray(actual)
        : type === "string"
          ? typeof actual === "string" && actual.length > 0
          : typeof actual === type;

    if (!ok) {
      throw new Error(`invalid response at ${path}.${key}: expected ${type}`);
    }
  }

  return value;
}

export function extractJson(content) {
  if (typeof content !== "string") {
    throw new Error("message content is not a string");
  }

  const fenced = content.match(/```(?:json)?\s*([\s\S]*?)```/i);
  const candidate = (fenced ? fenced[1] : content).trim();

  try {
    return JSON.parse(candidate);
  } catch (error) {
    throw new Error(`failed to parse JSON: ${error.message}`);
  }
}

export function logUsage(meta, label = "request") {
  const usage = meta.usage ?? {};
  console.log(
    JSON.stringify({
      label,
      attempt: meta.attempt,
      durationMs: meta.durationMs,
      inputTokens: usage.prompt_tokens ?? null,
      outputTokens: usage.completion_tokens ?? null,
      totalTokens: usage.total_tokens ?? null,
    }),
  );
}

export async function mapWithConcurrency(items, limit, worker) {
  const results = new Array(items.length);
  let cursor = 0;

  async function next() {
    while (cursor < items.length) {
      const index = cursor;
      cursor += 1;
      results[index] = await worker(items[index], index);
    }
  }

  await Promise.all(
    Array.from({ length: Math.min(limit, items.length) }, () => next()),
  );

  return results;
}

调用示例见 resources/llm-client/example.mjs。核心调用是 callLLMWithDeadline({ baseUrl, apiKey, body, deadlineMs })，成功时返回 { data, meta: { attempt, durationMs, usage } }。

为什么要随机抖动

假设 100 个任务同时被限流，固定等 2 秒会让它们在 2 秒后同时再次请求，形成第二波拥塞。加上随机抖动后，请求会被打散。脚本里的算法是 baseDelayMs * 2 ** attempt + Math.random() * baseDelayMs：指数部分让等待越来越长，随机部分把同时被限流的任务错开。

Retry-After 优先级高于指数退避。服务端明确告诉你等多久时，就按它来。

加上总时长上限

单次超时不够。如果每次都在 60 秒超时，重试 3 次就是 4 分钟，批量任务会被一个坏请求拖住。

核心是给整批任务一个总预算，而不是让单个请求无限占用。

脚本里 callLLM 接收 deadlineMs，每次重试前用“截止时间 − 已用时间”算出剩余预算，再取 min(timeoutMs, 剩余时间) 作为本次超时；连下一次退避都会超出预算时直接抛错。callLLMWithDeadline 只是给它补了个 180 秒的默认值。关键是给整批任务设预算，而不是让单个请求无限占用。

校验结构化输出

模型返回 200 不代表结果可用。脚本里的 assertShape 是不引入依赖的校验：按 shape 描述逐层检查类型，缺字段或类型不对直接抛错；extractJson 负责从可能带代码围栏的文本里把 JSON 抠出来。

如果需要更严格的约束，用 Zod 或 JSON Schema。原则是：解析失败和字段缺失都当作失败处理，不要带着半截数据继续跑。

记录成本

把每次请求的 token 和耗时写进日志，月底才能知道钱花在哪。脚本里的 logUsage 从 meta.usage 取 prompt_tokens / completion_tokens / total_tokens，连同耗时和标签打一行 JSON。

批量任务建议按批次汇总，而不是每个请求打一行。否则日志会比结果还大。

并发要克制

不要用 Promise.all 一次打 100 个请求。脚本里的 mapWithConcurrency(items, limit, worker) 就是个固定并发池：最多同时跑 limit 个，完成一个补一个，结果按原顺序返回。建议同时 3 到 5 个。

并发数不是越高越好。限流通常按分钟计算，盲目提高并发只会增加 429 和重试成本。

上线前检查

• API Key 只存在服务端环境变量里，绝不进入前端代码。
• 日志里不打印完整 prompt，避免泄露用户数据。
• 对非幂等操作做去重，重试前确认不会重复扣费。
• 给整批任务设总预算和总时长，失败的任务单独落盘，方便补跑。
• 记录模型名和版本，服务商升级模型时能追溯结果变化。

这套客户端不依赖任何特定服务商。把 baseUrl、请求体和响应解析换掉，就能接到兼容 OpenAI 接口的其他服务上。

下一步读什么

请求层稳了，返回内容还要校验：AI 返回的 JSON 不稳。