OpenAI Batch API 教程：半价跑批量任务

{"custom_id": "faq-001", "method": "POST", "url": "/v1/responses", "body": {"model": "gpt-4.1-mini", "input": "把这段话改写成一句话摘要：退款会在 3 到 5 个工作日到账。"}}
{"custom_id": "faq-002", "method": "POST", "url": "/v1/responses", "body": {"model": "gpt-4.1-mini", "input": "把这段话改写成一句话摘要：发票需要提供公司名称和税号。"}}

# 1. 上传，purpose 必须是 batch
curl https://api.openai.com/v1/files \
  -H "Authorization: Bearer $OPENAI_API_KEY" \
  -F purpose="batch" \
  -F file="@batch-input.jsonl"

# 2. 创建批次
curl https://api.openai.com/v1/batches \
  -H "Authorization: Bearer $OPENAI_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "input_file_id": "file-abc123",
    "endpoint": "/v1/responses",
    "completion_window": "24h"
  }'

# 3. 查状态
curl https://api.openai.com/v1/batches/batch_abc123 \
  -H "Authorization: Bearer $OPENAI_API_KEY"

# 4. 下载结果文件
curl https://api.openai.com/v1/files/file-output-abc123/content \
  -H "Authorization: Bearer $OPENAI_API_KEY" \
  -o batch-output.jsonl

import time
from openai import OpenAI

client = OpenAI()

with open("batch-input.jsonl", "rb") as fh:
    upload = client.files.create(file=fh, purpose="batch")

batch = client.batches.create(
    input_file_id=upload.id,
    endpoint="/v1/responses",
    completion_window="24h",
)

while batch.status not in ("completed", "failed", "expired", "cancelled"):
    time.sleep(60)
    batch = client.batches.retrieve(batch.id)
    print(batch.status, batch.request_counts)

# 结果文件
content = client.files.content(batch.output_file_id).text
open("batch-output.jsonl", "w").write(content)

import json

def load_jsonl(path):
    with open(path, encoding="utf-8") as fh:
        for line in fh:
            line = line.strip()
            if line:
                yield json.loads(line)

results = {}
for row in load_jsonl("batch-output.jsonl"):
    custom_id = row.get("custom_id")
    response = row.get("response") or {}
    if row.get("error"):
        results[custom_id] = {"ok": False, "error": row["error"]}
        continue
    if response.get("status_code") == 200:
        body = response.get("body", {})
        results[custom_id] = {"ok": True, "body": body}
    else:
        results[custom_id] = {"ok": False, "error": response}

# 和输入比对，找出既没成功也没失败的漏项
inputs = [row["custom_id"] for row in load_jsonl("batch-input.jsonl")]
missing = [cid for cid in inputs if cid not in results]
print("总数", len(inputs), "有结果", len(results), "缺失", len(missing))