让 AI 一口气处理 100 万条请求：openai-batch 入门与实战
想象一下，你的邮箱里躺着 10 万封邮件，需要 AI 逐一总结；或者你手头有 50 万份产品评论，需要 AI 判断每条是好评还是差评。
如果按常规方式一条一条调用模型，不仅慢，还可能因为并发限制而被限速。
这时候，批处理（batch） 就派上用场了：把请求打包成文件，一次性交给云端，让 AI 在后台慢慢跑，你只需等待最终结果。
本文将带你认识 openai-batch——一款把“打包、提交、等待、下载”四步简化为几行代码的小工具，并手把手演示如何用它向 OpenAI 和 Parasail 发送大批量任务。
为什么要用批处理？
| 传统逐条调用 | 批处理 |
|---|---|
| 每次都要握手、鉴权 | 一次鉴权，批量提交 |
| 网络延迟累积 | 后台排队，几乎零额外延迟 |
| 容易被限速 | 官方优先保障批处理带宽 |
| 需要自己写重试、重连 | 官方帮你重试，失败可单独重跑 |
一句话：省钱、省时、省心。
openai-batch 是什么？
openai-batch 是一个轻量级 Python 库，把 OpenAI 官方批处理 API 的“文件上传→创建任务→轮询状态→下载结果”做成了极简的高阶接口，同时兼容 Parasail 提供的 HuggingFace 模型托管服务。
核心特点：
-
自动识别模型归属：写 gpt-4o-mini就用 OpenAI，写meta-llama/Meta-Llama-3-8B-Instruct就用 Parasail。 -
支持聊天补全（chat completion）和文本嵌入（embedding）。 -
提供“一口气跑完”和“分步可控”两种模式。 -
断点续跑：任务中途关机也不怕，记下 batch_id即可续命。
五分钟上手
1. 安装
pip install openai-batch
确保你已准备好以下环境变量之一：
# 想用 OpenAI
export OPENAI_API_KEY="sk-xxxxxxxxxxxxxxxx"
# 想用 Parasail
export PARASAIL_API_KEY="psl_xxxxxxxxxxxxxxxx"
2. 最简单的一口气模式
下面这段代码会：
-
随机生成 100 条“讲个笑话”的请求； -
自动上传到 Parasail； -
后台排队、运行； -
完成后把结果 JSONL 下载到本地； -
返回本地路径，方便你继续处理。
import random
from openai_batch import Batch
objects = ["猫", "机器人", "咖啡杯", "宇宙飞船", "香蕉"]
with Batch() as batch:
for _ in range(100):
batch.add_to_batch(
model="meta-llama/Meta-Llama-3-8B-Instruct",
temperature=0.7,
max_completion_tokens=1000,
messages=[{"role": "user",
"content": f"请讲一个关于{random.choice(objects)}的笑话"}]
)
result, output_path, error_path = batch.submit_wait_download()
print("任务状态:", result.status)
print("结果文件:", output_path)
print("错误文件:", error_path or "无")
运行完你会得到两个文件：
-
batch_output.jsonl：每条请求的模型回复，按原顺序排列。 -
batch_errors.jsonl：万一哪条请求格式不对或超时，会把错误详情写在这里，方便单独重跑。
进阶：把流程拆成四步
如果你需要更细的控制，例如：
-
想在白天提交任务，夜里再下载结果； -
想监控进度，给老板汇报百分比； -
想断点续跑，防止笔记本合盖就凉凉。
可以把“一口气模式”拆成四步：
from openai_batch import Batch
import time
# 1. 先声明文件路径，方便自己管理
batch = Batch(
submission_input_file="my_input.jsonl",
output_file="my_output.jsonl",
error_file="my_errors.jsonl"
)
# 2. 逐条添加请求
for obj in ["猫", "机器人", "咖啡杯", "宇宙飞船", "香蕉"]:
batch.add_to_batch(
model="gpt-4o-mini",
messages=[{"role": "user", "content": f"请用三句话介绍{obj}"}]
)
# 3. 提交，并获得 batch_id
batch_id = batch.submit()
print("任务已提交，ID:", batch_id)
# 4. 轮询状态直到完成
while True:
status = batch.status()
print("当前状态:", status.status)
if status.status in ["completed", "failed", "expired", "cancelled"]:
break
time.sleep(60)
# 5. 下载结果
output_path, error_path = batch.download()
print("结果文件:", output_path)
把 batch_id 记到小本本，下次电脑重启后只需：
batch = Batch(batch_id="batch_abc123")
# 继续上面的轮询和下载
场景示例
1. 文本嵌入（embedding）
把两段话转成向量，供后续检索或聚类：
with Batch() as batch:
for doc in ["敏捷的棕色狐狸跳过懒狗", "机器学习模型可以理解自然语言"]:
batch.add_to_batch(
model="text-embedding-3-small",
input=doc
)
_, output_path, _ = batch.submit_wait_download()
输出同样是 JSONL，每行包含 embedding 字段。
2. 显式指定供应商
当模型名称冲突或想强制走某一家时：
from openai_batch.providers import get_provider_by_name
provider = get_provider_by_name("parasail")
batch = Batch(provider=provider)
batch.add_to_batch(
model="microsoft/DialoGPT-small",
messages=[{"role": "user", "content": "你好"}]
)
命令行：不写代码也能跑
如果你更习惯终端，openai-batch 附带了两条命令：
-
生产示例输入
python -m openai_batch.example_prompts会输出若干 JSONL 格式的示例请求，方便你照猫画虎。
-
一键跑批
# 先生成输入文件 python -m openai_batch.example_prompts | \ python -m openai_batch.create_batch --model gpt-4o-mini > input.jsonl # 提交并等待 python -m openai_batch.run input.jsonl
常用参数：
-
-c：只提交，不等待，适合在 CI/CD 里异步跑。 -
--resume <BATCH_ID>：续跑。 -
--dry-run：空跑检查格式，不上传。
常见疑问 Q&A
Q1：文件格式有什么讲究？
A：必须是 JSON Lines（.jsonl），每条一行。openai-batch 会在内存里自动帮你生成，无需手写。
Q2：一次最多能塞多少条？
A：OpenAI 官方限制单文件 2 GB、1000 万 tokens；Parasail 限制略宽松。通常 100 万条短文本毫无压力。
Q3：价格怎么算？
A：批处理单价通常比实时调用便宜 50%，且不计流式返回的额外费用。具体以官方账单为准。
Q4：失败了怎么办？
A：下载的 *_errors.jsonl 会告诉你哪几行出错，修正后可单独打包成新文件重新跑。
小结
-
把 10 万条请求一次性丢给 AI，再也不是梦。 -
openai-batch 用不到 100 行核心代码，把繁琐的文件上传、状态轮询、错误重试都包圆了。 -
无论你是 Python 爱好者还是命令行党，都能在三分钟内完成第一次批量调用。
现在就试试：打开终端，装好库，把“请总结这封邮件”的请求攒成一万条，让 AI 夜里默默跑完，明早收结果。你会发现，批处理才是大模型时代的“懒人神器”。
祝你玩得开心，跑批愉快！
