批量API使用教程

批量 API（Batch）怎么用？大批量调用省一半成本

实时 API 适合在线交互，批量 API 适合离线任务。两者的取舍不是功能差异，而是业务场景的匹配度。

批量 API 的核心逻辑很简单：把一堆请求打包成一个文件提交，API 在后台处理（最长 24 小时），完成后取结果文件。价格是同步 API 的五折，但响应是异步的，没法实时拿到结果。

这个组合对哪类场景值，对哪类场景反而麻烦，下面说清楚。

一、Batch API 的基本机制

以 OpenAI Batch API 为例，完整调用链路分四步：

关键参数：

• completion_window：批处理必须在多长时间内完成，目前固定 24 小时
• 结果文件最长在批次结束后保留 24 小时，超时未取会丢失
• 批量请求的配额与同步 API 独立，不互相占用

价格：所有模型统一打五折（输入和输出 token 均适用），这是官方公开的定价策略。

二、最小可用示例

完整的 Python 调用流程，用官方 SDK：

import time
import json
from openai import OpenAI

client = OpenAI(api_key="YOUR_API_KEY")

# ── 1. 构造请求文件（JSONL 格式）──
requests = [
{
"custom_id": f"task-{i}",
"method": "POST",
"url": "/v1/chat/completions",
"body": {
"model": "gpt-4o-mini",
"messages": [
{"role": "user", "content": f"用一句话总结第 {i} 篇文档的核心观点"}
],
"max_tokens": 100
}
}
for i in range(1, 51)
]

jsonl_path = "/tmp/batch_requests.jsonl"
with open(jsonl_path, "w", encoding="utf-8") as f:
for req in requests:
f.write(json.dumps(req, ensure_ascii=False) + "\\n")

# ── 2. 上传文件 ──
with open(jsonl_path, "rb") as f:
uploaded_file = client.files.create(
file=f,
purpose="batch"
)
file_id = uploaded_file.id
print(f"上传完成，file_id: {file_id}")

# ── 3. 创建批量任务 ──
batch = client.batches.create(
input_file_id=file_id,
endpoint="/v1/chat/completions",
completion_window="24h",
metadata={"description": "文档批量摘要任务"}
)
batch_id = batch.id
print(f"批次已创建，batch_id: {batch_id}，状态: {batch.status}")

# ── 4. 轮询等待完成 ──
while batch.status in ("validating", "in_progress", "finalizing"):
time.sleep(30) # 每 30 秒查一次，不要高频轮询
batch = client.batches.retrieve(batch_id)
print(f"当前状态: {batch.status}，进度: {batch.request_counts.completed}/{batch.request_counts.total}")

print(f"最终状态: {batch.status}")

# ── 5. 下载结果 ──
if batch.status == "completed" and batch.output_file_id:
result_file = client.files.content(batch.output_file_id)
# result_file.content 是一个形如 JSONL 的文本
print(result_file.content[:500])
elif batch.status == "failed":
# 查看失败原因
if batch.error_file_id:
err_file = client.files.content(batch.error_file_id)
print("错误详情:", err_file.content)

简化版（用 requests 库，不需要 SDK）：

import requests
import time
import json

API_KEY = "YOUR_API_KEY"
BASE_URL = "https://api.openai.com"

headers = {
"Authorization": f"Bearer {API_KEY}",
"Content-Type": "application/json"
}

# 1. 上传文件
with open("/tmp/batch_requests.jsonl", "rb") as f:
resp = requests.post(
f"{BASE_URL}/v1/files",
headers={"Authorization": f"Bearer {API_KEY}"},
files={"file": ("requests.jsonl", f, "application/jsonl")}
)
file_id = resp.json()["id"]

# 2. 创建批次
batch_resp = requests.post(
f"{BASE_URL}/v1/batches",
headers=headers,
json={
"input_file_id": file_id,
"endpoint": "/v1/chat/completions",
"completion_window": "24h"
}
)
batch_id = batch_resp.json()["id"]

# 3. 轮询
while True:
time.sleep(60)
status_resp = requests.get(f"{BASE_URL}/v1/batches/{batch_id}", headers=headers)
status = status_resp.json()["status"]
print(f"状态: {status}")
if status == "completed":
output_id = status_resp.json()["output_file_id"]
result = requests.get(f"{BASE_URL}/v1/files/{output_id}/content", headers=headers)
for line in result.text.strip().split("\\n"):
print(json.loads(line))
break
elif status in ("failed", "expired", "cancelled"):
break

三、同步 API vs 批量 API：差异对比

什么时候选 Batch，什么时候不用：

• 需要实时反馈 → 用同步 API，Batch 的异步特性帮不上忙
• 100 条以上的同类请求 → Batch，通常能省 50% 成本
• 任务能接受等几分钟到 24 小时 → Batch
• 请求之间有依赖（比如上一条的结果影响下一条输入） → 同步 API 或流式 API，Batch 不支持链式调用

四、适用场景

场景一：文档批量处理

把一个文件夹里的 500 篇技术文档逐一摘要，每篇丢给模型生成 3 句话总结。

• 同步调用：500 次请求，可能触发限流，需要加延迟，总费用高
• Batch 调用：构造 500 行 JSONL，上传，睡觉等结果，五折

场景二：大模型评测

用标准测试集（如 2000 道选择题）批量跑模型，准确率统计。

• Batch 天然适合评测场景：评测不需要实时，结果拿到后统一分析
• 可以同时提交多个批次并行处理

场景三：数据标注

对一批用户评论做情感分类或实体识别。

• 评论量级大（几千到几万条），同步调用成本高且容易触发限流
• Batch 稳定出结果，错误行单独从 error_file 里捞

场景四：批量翻译

本地化场景，批量将一批 UI 文本翻译成多语言。

• 翻译通常对实时性要求低，Batch 是性价比最高的选择

五、常见坑

坑 1：文件格式严格出错

JSONL 每行必须是合法的 JSON，不能有尾部逗号、空行、UTF-8 BOM 头。

# 正确
with open("requests.jsonl", "w", encoding="utf-8") as f:
f.write(json.dumps(req) + "\\n")

# 常见错误：多打了个逗号或末尾换行

验证文件格式：

with open("requests.jsonl") as f:
for i, line in enumerate(f, 1):
try:
json.loads(line)
except json.JSONDecodeError as e:
print(f"第 {i} 行格式错误: {e}")

坑 2：24 小时延迟

Batch 不保证秒级响应，测试阶段别等急了。官方建议用 Webhook 回调，但目前 OpenAI Batch API 尚未开放 Webhook，轮询是主要方式。轮询间隔 30-60 秒即可，不要低于 10 秒，容易被当攻击。

坑 3：结果文件过期

completed 状态出来后，结果文件只保留 24 小时。拿到 batch_id 后尽快取结果，或者在提交时就设计好取结果的后续流程（比如任务完成后自动触发下载）。

坑 4：配额计算错误

Batch 的输入输出 token 价格虽然五折，但总费用 = (输入 token 数 × 0.5 + 输出 token 数 × 0.5) × 单价。如果输出 token 占比高（如大量生成的评测结果），实际节省幅度比想象的小。先在 Playground 跑几条样本估算 token 量，再决定是否用 Batch。

坑 5：错误重试逻辑

Batch 里单条请求失败（如模型不支持该请求格式），整批不会全部失败，错误行会出现在 error_file_id 里。需要主动读取并重试失败的请求：

if batch.error_file_id:
err_resp = requests.get(
f"{BASE_URL}/v1/files/{batch.error_file_id}/content",
headers=headers
)
failed_requests = []
for line in err_resp.text.strip().split("\\n"):
err_obj = json.loads(line)
failed_requests.append(err_obj["custom_id"])
print(f"失败请求 ID: {failed_requests}")

坑 6：endpoint 前缀不匹配

批量请求的 url 字段必须和创建批次时传的 endpoint 参数前缀匹配。如果创建时写的是 /v1/chat/completions，JSONL 文件里每条请求的 url 也必须以 /v1/chat/completions 开头（可以带查询参数），否则报错 The URL provided for this request does not prefix-match the batch endpoint。

六、快速排错表

七、配置检查清单

开始批量任务前确认以下事项：

• 已通过官方定价页确认目标模型的 Batch 折扣比例
• JSONL 文件已逐行验证，无多余逗号、空行、特殊字符
• 每条请求的 custom_id 唯一且有业务含义（方便后续对账）
• endpoint 参数与 JSONL 文件中每条 url 前缀一致
• 预估 token 总量：输入 token × 0.5 × 单价 + 输出 token × 0.5 × 单价，与预算对比
• 轮询逻辑已有超时处理（建议轮询总时长不超过 25 小时）
• 任务完成后有自动下载结果的后续流程，不依赖人工操作
• 错误文件（error_file_id）的读取和重试逻辑已实现
• 如果结果量大，确认本地有足够的磁盘空间存放结果文件