LLM Batch API 离线推理生产实战：用 JSONL 队列、custom_id 与结果回灌降低非实时任务成本

背景：不是所有大模型请求都应该走在线接口

很多团队接入大模型后，会把所有任务都塞进同一套在线调用链路：用户请求、后台分类、批量摘要、历史数据补标签、知识库嵌入、评测集回放、风控内容审核，全部用同步 API 发起。刚开始请求量不大，这种做法简单直接；一旦数据规模上来，问题就会集中暴露。

第一类问题是成本与限流。 后台任务通常不需要秒级响应，却会和在线用户请求争抢同步接口额度。如果一个夜间分类任务把限流额度打满，白天用户请求的延迟和失败率就会被无关任务拖高。

第二类问题是可恢复性。 同步脚本跑到一半失败，常见结果是没人说得清哪些样本已经处理、哪些样本需要重试、哪些输出已经写回业务库。重跑一遍看似简单，实际可能造成重复计费、重复写入和结果漂移。

第三类问题是结果对账。 离线任务通常以万级、百万级样本为单位。它需要的是作业视角：输入文件版本、任务批次、完成率、失败原因、Token 使用量、结果落库状态，而不是一条条同步请求的临时日志。

因此，Batch API 的价值不只是”便宜一点”。更准确地说，它把非实时 LLM 请求从在线调用模式，改造成一种可排队、可追踪、可恢复、可对账的离线推理流水线。

Batch API 的核心原理

主流模型平台的 Batch API 设计大同小异：开发者准备一批请求，通常以 JSONL 文件或请求数组提交；平台异步处理；开发者轮询作业状态；作业完成后下载结果文件或读取结果流；最后用自定义 ID 把结果写回内部系统。

平台	输入格式	ID 字段	关键状态	特点
OpenAI	JSONL 文件	custom_id	validating → in_progress → finalizing → completed	支持评测、分类、嵌入等场景
Anthropic	Messages 请求数组	custom_id	in_progress → ended	明确列出 succeeded/errored/canceled/expired
Gemini	inline 请求或 JSONL	key	pending → running → succeeded/failed/cancelled/expired	支持取消 batch job

这几个实现的细节不同，但工程抽象可以统一成五个核心对象：

• Input Manifest：冻结的输入清单，包含业务主键、输入哈希、prompt 版本、模型参数
• Batch Job：提交到模型平台的异步作业
• Status Poller：状态轮询器，将平台状态映射为内部标准状态机
• Result Sink：结果落地层，先存原始响应再做业务回灌
• Retry Ledger：重试账本，按错误类型分类处理失败

适合 Batch API 的任务类型

Batch API 最适合处理”吞吐量比即时性更重要”的任务。

典型场景：

• 离线评测：对黄金集、回归集、红队样本批量跑模型输出，并按版本归档
• 大规模分类：给工单、评论、商品、知识库片段批量打标签
• 批量摘要：对历史客服会话、会议纪要、长文档生成摘要或结构化字段
• Embedding 构建：对内容库做初始化嵌入或周期性重算
• 数据清洗：用模型辅助判断脏数据、重复数据、异常字段或合规风险
• 非实时内容生成：批量生成商品描述、SEO 草稿、内部知识问答初稿

不适合的场景也很明确：在线聊天、实时语音、边输入边输出的用户交互、需要工具调用实时确认的 Agent 流程、支付风控等强实时链路。

生产架构：把 Batch 当成作业系统，而不是脚本

一个可维护的 Batch API 流水线通常包含七层。

1. 任务入口：把业务需求转成 batch_task

不要让业务脚本直接生成 JSONL 并调用模型平台。更稳妥的做法是先创建内部 batch_task 记录，记录任务类型、模型、提示词版本、输入数据版本、预估请求数、预算上限、发起人、审批状态和过期策略。

CREATE TABLE llm_batch_task (
  id TEXT PRIMARY KEY,
  task_type TEXT NOT NULL,
  model_name TEXT NOT NULL,
  prompt_version TEXT NOT NULL,
  input_manifest_uri TEXT NOT NULL,
  status TEXT NOT NULL,
  total_items INTEGER NOT NULL,
  estimated_input_tokens BIGINT,
  estimated_output_tokens BIGINT,
  budget_cents BIGINT,
  provider_batch_id TEXT,
  created_at TIMESTAMP NOT NULL,
  updated_at TIMESTAMP NOT NULL
);

这里的关键是：Batch 作业必须先成为内部系统中的可审计对象，再成为外部模型平台上的异步任务。

2. 输入清单：用 manifest 固定输入版本

Batch 任务最怕”输入不确定”。如果直接从数据库实时查询待处理数据，重试时可能查到另一批数据，导致结果不可复现。

更好的方式是生成 Input Manifest：把本次任务的输入记录冻结成一个清单，包含业务主键、输入哈希、prompt 版本、模型参数、输出 schema 版本和自定义 ID。

custom_id 不应使用随机 UUID 了事。推荐结构是：

{task_id}:{item_type}:{business_id}:{input_hash_prefix}

例如：task_20260706_ticket_label:ticket:TK2026003912:9f3a21bc

这样做有三个好处：

1. 可以从结果反查业务记录
2. 可以识别同一业务记录是否因输入变化而重新处理
3. 可以在回灌时做幂等判断

3. JSONL 构建：逐行校验，不要整体失败

JSONL 的优势是简单、可流式写入、可分片上传。它的风险是单行格式错误可能导致整个 batch 验证失败。

生成 JSONL 时，应在本地先做三类校验：

• 结构校验：每行必须是合法 JSON，并包含平台要求的字段
• 预算校验：估算输入 token 和最大输出 token，防止单条请求异常膨胀
• 策略校验：禁止把敏感字段、未脱敏原文或不该出域的数据写入 batch 文件

一个简化的 JSONL 行示例：

{"custom_id":"task_20260706_ticket_label:ticket:TK2026003912:9f3a21bc","method":"POST","url":"/v1/responses","body":{"model":"gpt-5.5-mini","input":"请将以下工单归类为 billing、technical 或 account。\\n\\n工单内容：...","max_output_tokens":200}}

4. 提交作业：分片、命名和配额保护

生产环境不要把所有请求塞进一个巨型 batch。更稳妥的做法是按任务类型、模型、输入大小和业务优先级分片。例如一个百万级分类任务可以拆成 100 个 shard，每个 shard 独立提交、轮询和回灌。

分片的好处：

优势	说明
故障隔离	单个 shard 失败时，不影响其他 shard
并发控制	可以控制并发提交数量
成本统计	可以按 shard 统计失败率和成本
优先级调度	可以把高优先级任务插队，而不是等待一个巨型作业结束

提交前应检查外部平台的 batch 限制（单批请求数、文件大小、排队 token 上限、创建频率、作业有效期等），把这些限制放进 provider adapter 配置，不要写死在业务代码里。

5. 状态轮询：状态机必须显式

Batch 作业天然是异步的，状态机比一次性脚本更重要。内部状态可以设计为：

CREATED → VALIDATING → SUBMITTED → RUNNING → FINALIZING → COMPLETED
                                                      → FAILED
                                                      → EXPIRED
                                                      → CANCELLED

轮询器要做三件事：

1. 定期查询 provider batch 状态，并把原始状态映射成内部标准状态
2. 记录状态变化时间，计算排队时长、运行时长、最终完成时长
3. 处理卡住的作业——如果某个任务长时间停留在提交状态，应进入人工检查或自动拆分重试，而不是无限轮询

6. 结果回灌：按 custom_id 对账，不按行号对账

结果回灌是 Batch API 最容易出错的地方。输出文件通常也是 JSONL，但输出顺序不一定等于输入顺序。正确做法是读取每一行，解析 custom_id 或 key，查找内部 manifest，确认输入哈希、任务 ID、业务 ID 和输出 schema 版本都匹配后再写回。

结果表建议拆成两层：

CREATE TABLE llm_batch_result_raw (
  id TEXT PRIMARY KEY,
  task_id TEXT NOT NULL,
  custom_id TEXT NOT NULL,
  provider_request_id TEXT,
  result_type TEXT NOT NULL,
  raw_json TEXT NOT NULL,
  input_tokens INTEGER,
  output_tokens INTEGER,
  created_at TIMESTAMP NOT NULL
);

CREATE TABLE llm_batch_result_applied (
  id TEXT PRIMARY KEY,
  task_id TEXT NOT NULL,
  business_id TEXT NOT NULL,
  custom_id TEXT NOT NULL,
  output_version TEXT NOT NULL,
  applied_status TEXT NOT NULL,
  applied_at TIMESTAMP
);

第一张表保留 provider 原始结果，便于审计和重新解析。第二张表记录是否已经写回业务系统，避免重复写入。

7. 重试账本：失败不是一个状态，而是一组原因

失败结果至少应分为五类：

错误类型	说明	处理策略
validation_error	输入结构或参数错误	修正请求后重提
rate_or_capacity_error	平台容量或排队限制	拆分或延后重试
expired	作业窗口内未完成	降低 shard 大小或调整提交节奏
content_policy_error	内容被安全策略拦截	进入合规处理
internal_error	平台或网络异常	按幂等策略重试

不要对所有失败简单重跑。validation_error 重跑只会再次失败；content_policy_error 盲目重跑可能扩大合规风险；expired 直接重跑可能继续过期。重试系统应基于错误类型给出不同动作。

常见误区

误区一：把 Batch API 等同于在线 dynamic batching

Batch API 是平台提供的异步作业接口，面向非实时任务。Continuous batching 或 in-flight batching 是推理服务内部的调度策略，面向在线吞吐优化。两者名字都叫 batching，但工程边界完全不同。本文讨论的是前者：如何把大量非实时请求组织成可追踪的离线作业。

误区二：只看折扣，不看资金占用和过期风险

Batch API 往往有成本优势，但这不等于可以无限提交。批量任务可能占用预算、触发排队上限，甚至因为任务过大而过期。生产系统应在提交前估算 token 和预算，在运行中监控完成率，在过期后只重试未完成样本。

误区三：用自增行号做结果映射

行号在输入文件里看起来稳定，但在输出文件里不一定可靠。结果应该用 custom_id、key 或内部业务主键映射。否则一旦输出顺序变化，就会把 A 用户的结果写到 B 用户记录上。

误区四：把 prompt 写在脚本里

离线批处理通常会反复运行。如果 prompt 写死在脚本里，半年后很难解释某批结果为什么生成成那样。prompt、模型、温度、最大输出、输出 schema 都应版本化，并写入 task 和 manifest。

误区五：只保存最终结果，不保存原始响应

只保存业务字段会丢失排障信息。原始响应里的 request id、usage、finish reason、错误类型、模型版本都可能成为审计和成本分析依据。建议先落 raw，再做解析和业务回灌。

推荐的离线推理流水线

完整流程如下：

业务系统 → 创建 batch_task → 生成 input_manifest → 构建 JSONL shard
→ 上传并提交 provider batch → 轮询 batch 状态 → 下载 output/error JSONL
→ 写入 raw_result → 解析结构化结果 → 幂等回灌业务表
→ 生成成本、失败率和质量报告

四个关键控制点：

1. 提交前门禁：检查输入规模、敏感字段、prompt 版本、模型可用性、预算上限、输出 schema 和 dry-run 样本
2. 运行中观测：监控 submitted、running、completed、expired、failed 的 shard 数量，统计排队时长、完成率和预计剩余时间
3. 完成后对账：输入 manifest 有多少条，成功多少条，失败多少条，过期多少条，缺失多少条，业务回灌多少条，都要能对上
4. 重试与复盘：只重试可重试项，并把错误原因、重试次数、最终状态写入 ledger

代码示例：用 custom_id 做幂等回灌

下面示例演示核心思想：读取 provider 输出 JSONL，按 custom_id 写入 raw 结果，再交给业务回灌器处理。

import fs from "node:fs";
import readline from "node:readline";

type BatchLine = {
  custom_id?: string;
  key?: string;
  response?: unknown;
  result?: unknown;
  error?: unknown;
};

type ManifestItem = {
  taskId: string;
  customId: string;
  businessId: string;
  inputHash: string;
};

async function loadManifestByCustomId(customId: string): Promise<ManifestItem | null> {
  // 生产环境中应查询数据库或对象存储中的 manifest 索引
  return null;
}

async function saveRawResult(args: {
  taskId: string;
  customId: string;
  businessId: string;
  resultType: "succeeded" | "errored" | "expired" | "unknown";
  rawJson: string;
}) {
  // 先写 raw_result，并对 taskId + customId 建唯一索引，保证幂等
}

function classifyResult(line: BatchLine): "succeeded" | "errored" | "expired" | "unknown" {
  if (line.error) return "errored";
  const resultType = (line.result as any)?.type;
  if (resultType === "succeeded") return "succeeded";
  if (resultType === "errored") return "errored";
  if (resultType === "expired") return "expired";
  if (line.response) return "succeeded";
  return "unknown";
}

export async function importBatchOutput(filePath: string) {
  const stream = fs.createReadStream(filePath, { encoding: "utf8" });
  const rl = readline.createInterface({ input: stream, crlfDelay: Infinity });

  for await (const rawLine of rl) {
    if (!rawLine.trim()) continue;
    const line = JSON.parse(rawLine) as BatchLine;
    const customId = line.custom_id ?? line.key;
    if (!customId) {
      throw new Error(`missing custom_id/key in output line: ${rawLine.slice(0, 200)}`);
    }
    const manifest = await loadManifestByCustomId(customId);
    if (!manifest) {
      throw new Error(`orphan batch result: ${customId}`);
    }
    await saveRawResult({
      taskId: manifest.taskId,
      customId,
      businessId: manifest.businessId,
      resultType: classifyResult(line),
      rawJson: rawLine,
    });
  }
}

这段代码刻意没有把结果直接写进业务表。生产系统应先完成 raw 落库，再由解析器按输出 schema 做结构化校验，最后由回灌器幂等写入业务字段。

上线检查清单

上线前建议逐项检查：

1. 任务建模：是否有 batch_task、manifest、raw_result、applied_result 和 retry_ledger
2. 输入冻结：是否能复现某次 batch 的完整输入、prompt、模型和参数
3. ID 设计：custom_id 是否能反查业务记录，是否符合各平台字符和长度限制
4. 本地校验：JSONL 是否逐行校验，是否预估 token 和预算
5. 状态轮询：是否处理 running、completed、failed、expired、cancelled 等状态
6. 结果映射：是否只按 custom_id/key 对账，完全不依赖输出顺序
7. 错误分类：是否区分验证错误、平台错误、过期、取消和内容策略错误
8. 幂等回灌：重复导入同一结果文件是否不会重复写业务表
9. 预算保护：是否有任务级、租户级、模型级预算上限
10. 数据保留：是否理解平台对 Batch 输入输出的保留策略，并按业务合规要求处理敏感数据
11. 人工复核：高风险任务是否抽样复核，是否有回滚或撤销标记机制
12. 报表：是否能输出成功率、失败率、过期率、token 消耗、单位成本和业务回灌完成率

总结

Batch API 不只是”便宜一点的在线 API”——它是一种完全不同的工程范式。把非实时 LLM 任务从同步调用模式迁移到离线推理流水线，需要建立完整的作业生命周期管理：从任务建模、输入冻结、JSONL 分片、状态轮询，到 custom_id 结果对账、错误分类重试和幂等回灌。当你的系统能以作业视角回答”这批任务处理了多少、成功了哪些、失败了哪些、花了多少钱”时，才算真正驾驭了 Batch API 的生产实践。

参考资料

• OpenAI Batch API
• Anthropic Claude Message Batches
• Google Gemini Batch API
• Google Cloud Gemini Batch Inference