OpenAI Batch API 用 custom_id 如何对应原始请求

我第一次把一批不同业务的请求交给 OpenAI Batch API 时，最容易犯的错是把输出数组的第几行当成输入的第几行。这个假设不可靠：Batch 输出行顺序可能变化，真正用于回配原始请求的是每行唯一的 custom_id。把它设计成业务关联键，再用它更新本地任务表，批量结果就不会因为乱序而串单。

官方地址：https://developers.openai.com/api/docs/guides/batch

先把 custom_id 当成业务关联键

custom_id 不是模型生成的结果 ID，也不是创建批次后才得到的 batch_...。它由开发者在输入 JSONL 中提供，并且在同一个输入文件里必须唯一。我的做法是让它携带“业务对象 + 尝试号”，例如 invoice-20260914-0007-a1，同时在本地清单保留原始发票号、输入文件行号和请求目的。

在 JSONL 中保存可回溯的请求身份

先生成本地任务清单，再写 Batch 输入文件。下面是两个请求的最小形态。JSON 对象中的注释会破坏 JSONL 格式，所以字段含义放在代码块外解释：custom_id 负责关联，body 只放目标 endpoint 所需参数。

{"custom_id":"invoice-20260914-0007-a1","method":"POST","url":"/v1/responses","body":{"model":"gpt-5","input":"提取发票号和金额"}}
{"custom_id":"invoice-20260914-0008-a1","method":"POST","url":"/v1/responses","body":{"model":"gpt-5","input":"提取发票号和金额"}}

上传时文件用途要设置为 batch，创建批次时再使用返回的 input_file_id。不要把客户姓名、长文本或内部密码塞进 custom_id；它会出现在结果文件里，短而稳定的内部任务键更容易检索和脱敏。

from openai import OpenAI

client = OpenAI()

# 以 batch 用途上传 JSONL，返回的文件 ID 只代表输入文件。
with open("batchinput.jsonl", "rb") as input_file:
    uploaded = client.files.create(file=input_file, purpose="batch")

# 用输入文件 ID 创建批次；custom_id 仍由每行请求自己携带。
batch = client.batches.create(
    input_file_id=uploaded.id,
    endpoint="/v1/responses",
    completion_window="24h",
)
print(batch.id)

按 custom_id 解析成功与失败结果

批次完成后，从 output_file_id 读取成功结果。官方说明明确提醒：输出文件的行顺序可能与输入不同。因此解析器应该把每一行先转成以 custom_id 为键的字典，再去更新业务状态，而不是用递增数组下标。

import json

def reconcile(lines, task_by_custom_id):
    # 逐行按 custom_id 回配，允许服务端以不同顺序返回结果。
    for line in lines:
        item = json.loads(line)
        custom_id = item["custom_id"]
        task = task_by_custom_id.get(custom_id)
        if task is None:
            # 未知键先记录，避免误写入另一条业务任务。
            continue
        if item.get("response") is not None:
            # 成功响应保留底层 request_id，业务结果写入原任务。
            task["status"] = "succeeded"
            task["request_id"] = item["response"].get("request_id")
            task["result"] = item["response"].get("body")
        else:
            # 错误行只改变当前任务，后续是否重试由业务策略决定。
            task["status"] = "failed"
            task["error"] = item.get("error")

tasks = {
    "invoice-20260914-0007-a1": {"status": "pending"},
    "invoice-20260914-0008-a1": {"status": "pending"},
}
with open("batch_output.jsonl", encoding="utf-8") as output_file:
    reconcile(output_file, tasks)

这个写法还有一个好处：成功行通常包含 response，非 HTTP 类失败会在 error 中描述原因；如果任务失败，不能因为“这一行没有 response”就把它当成空结果。生产落库时可给状态更新加幂等条件，例如只有 pending 或 running 才写入，防止重复下载文件造成覆盖。

重试、重复和结果保留要分开处理

同一个 custom_id 不应在同一个输入文件里重复。若一条任务需要重试，不要原地覆盖旧结果，可以把尝试号递增为 invoice-20260914-0007-a2，并在本地表中用 parent_task_id 关联两次尝试。这样既能知道最终采用哪条结果，也能保留原始错误。

此外，batch_id 适合观察整批状态，output_file_id 和 error_file_id 适合定位结果文件；它们都不是逐请求的业务键。批次过期或取消时，已完成请求仍可能进入输出文件，未完成请求则应从错误文件读取对应的 custom_id，不要用“整批失败”覆盖所有行。

常见问题

custom_id 可以直接使用数据库自增 ID 吗？

可以，但最好加上业务类型和尝试号，形成可读且全局不易混淆的键；数据库 ID 本身只在你的业务表内有意义。

能不能按输入 JSONL 的行号读取输出？

不能把行号当成可靠映射。输出顺序可能变化，应该使用每行的 custom_id，本地还可以记录原始行号用于排查。

成功结果里的 request_id 能替代 custom_id 吗？

不建议。request_id 更适合日志、支持排查和审计；回配原始业务请求应优先使用开发者事先写入的 custom_id。

只记住一句话：Batch API 负责异步处理，custom_id 负责把异步结果带回你的业务。先把这个键设计稳定，再处理乱序、失败和重试，批量调用才不会变成一张难以还原的结果表。