OpenAI Batch API 怎样安排离线批量请求

OpenAI Batch API 适合不要求即时返回的离线任务：把多个请求写入 JSONL，上传为批次输入文件，创建一个 24 小时完成窗口的任务，完成后再下载结果。它的关键不是“把循环改成异步”，而是让每条请求有唯一 custom_id，这样输出顺序变化时仍能准确回填业务记录。

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

先把离线任务切成可追踪的 JSONL

Batch API 的输入不是一个 JSON 数组，而是 JSON Lines：每一行都是一次底层 API 请求。行中的 method、url 和 body 对应实际端点；custom_id 是业务侧的稳定键，建议直接包含业务表主键或任务编号，但不要放密钥和个人隐私。

{"custom_id":"ticket-1001","method":"POST","url":"/v1/responses","body":{"model":"gpt-5-mini","input":"请把这条工单归类为产品、支付或物流，并返回一个标签。"}}
{"custom_id":"ticket-1002","method":"POST","url":"/v1/responses","body":{"model":"gpt-5-mini","input":"请把这条工单归类为产品、支付或物流，并返回一个标签。"}}

这里的 JSON 必须保持严格有效，不能像 JavaScript 一样插入注释。生成文件时要校验每行都能独立解析，并确认所有 custom_id 不重复。官方文档还要求同一个输入文件只能使用一个模型；若任务需要不同模型，应拆成多个批次。

上传文件后再创建 24 小时批次

文件上传和批次创建是两个动作。Python SDK 的最小组织方式如下，上传阶段的 purpose 必须是 batch，创建阶段则把返回的文件 ID 放入 input_file_id。

from openai import OpenAI

client = OpenAI()  # 从环境变量读取密钥，不把密钥写进脚本

with open("batchinput.jsonl", "rb") as stream:
    input_file = client.files.create(file=stream, purpose="batch")  # 上传批次输入文件

batch = client.batches.create(
    input_file_id=input_file.id,       # 只引用已上传的批次文件
    endpoint="/v1/responses",         # 必须和 JSONL 每行的 url 保持一致
    completion_window="24h",          # 允许服务在离线窗口内完成
    metadata={"job": "ticket-labeling"},  # 用业务标签追踪批次
)
print(batch.id, batch.status)  # 先保存 batch_id，再进入状态查询

Batch API 官方说明的优势是相对同步 API 低 50% 成本、拥有独立且更高的限流空间，并以 24 小时作为清晰的完成窗口。它不适合聊天框、实时风控或必须立即展示答案的交互；这些场景的等待成本通常比折扣更重要。

用状态和结果文件判断真正的完成

创建成功只说明任务进入队列，不代表每条请求都已得到答案。状态通常会经历 validating、in_progress、finalizing，最终可能是 completed、failed、expired 或 cancelled。轮询时保存上一次状态，设置合理间隔，不要高频请求。

import time
from openai import OpenAI

client = OpenAI()
batch_id = "batch_请替换为实际ID"

while True:
    current = client.batches.retrieve(batch_id)  # 读取服务端状态
    if current.status in {"completed", "failed", "expired", "cancelled"}:
        break
    time.sleep(30)  # 离线任务不需要秒级轮询

print(current.status, current.request_counts)  # 区分总数、成功数和失败数
if current.status == "completed" and current.output_file_id:
    result = client.files.content(current.output_file_id)  # 下载成功与失败混合结果文件
    print(result.text)
elif current.error_file_id:
    errors = client.files.content(current.error_file_id)  # 失败行单独保存，供重试使用
    print(errors.text)

输出文件的行顺序不保证与输入一致，因此不要用数组下标直接回填。读取每行后，以 custom_id 查找业务记录；包含 response 的行进入成功处理，包含 error 的行进入失败表。批次过期时，已完成请求仍可从输出文件取得，未完成请求会进入错误文件，不能把整个批次简单当成全量失败。

提交前的边界检查清单

另外，流式参数 stream=true 不适合 Batch 请求；如果只想重试失败项，应从错误文件抽出新的 JSONL，并为新的批次使用新的任务标识。不要因为某一行失败就重发全部成功项。

常见问题

Batch API 能不能立即拿到每条响应？

不能按同步接口的方式等待单条响应。提交后应保存 batch_id，等状态进入终态，再从 output_file_id 下载结果。

为什么输出顺序和输入顺序不一样？

批处理是异步执行，完成先后会改变输出行顺序。custom_id 才是稳定关联键，不能用行号当业务主键。

一个批次可以混用不同模型吗？

官方指南要求一个输入文件只使用一个模型。不同模型的任务拆成不同 JSONL 和批次，并分别记录配置。