用Batch API跑120万条数据，成本从90美元降到3.7美元

上个月我用 Batch API 跑了 120 万条数据。

账单出来的时候，我一口咖啡差点喷屏幕上——3.7 美元。就 3.7。

你猜三个月前同样的量用实时 API 多少钱？90 美元。不是差一点点，是整整 24 倍。我当时盯着那个数字看了足足十秒，然后开始骂自己：之前到底在干嘛？

所以今天我想聊聊 Batch API。不是那种官方文档翻译，是我这几个月在生产环境里实际用下来的经验——哪些场景真香，哪些坑踩得我想删库跑路。

先说说这玩意到底是个啥

OpenAI 在 2024 年 4 月推的 Batch API。原理不复杂：你把一堆请求打包成 JSONL 文件丢上去，OpenAI 承诺 24 小时内处理完，结果写回给你。因为它可以把这些任务塞到 GPU 集群闲的时候跑，所以给你打五折。

对所有模型都打五折。GPT-4o、GPT-4o-mini、GPT-4-turbo，全系半价。

拿 GPT-4o 举例，实时 API 输入 $2.50/1M tokens，输出 $10.00/1M tokens。Batch 直接砍半：输入 $1.25，输出 $5.00。

看着很美好对吧？但有个事官方文档提了但不够显眼：速率限制跟你的 Tier 等级挂钩。我当时 Tier 2，以为能随便提交，结果第一批 5 万条排了整整 18 个小时。18 个小时啊，我从上午等到半夜。

等等，这里我要更正一下——不是"提交不了"，是提交成功了但排队排到你怀疑人生。这个区别挺大的。能提交，就是慢。

哪些场景真值得用？

我给自己定了个判断标准，就一句话：不需要在 30 秒内拿到结果，就上 Batch。

说说我实际用上的三个场景。

大规模文本分类

公司有个项目，80 万条用户反馈要做情感分类和意图识别。之前用实时 API，不仅要处理速率限制，还得自己写重试逻辑、断点续传，光那部分代码就 200 多行。出了问题半夜爬起来修，那段时间我老婆问我是不是出轨了。

换成 Batch API 之后，代码砍到 60 行。成本从预估的 $400 降到 $180。关键是，我终于能睡整觉了。

提交的代码大概长这样：

import json

tasks = []
for idx, feedback in enumerate(feedback_list):
 task = {
 "custom_id": f"task-{idx}",
 "method": "POST",
 "url": "/v1/chat/completions",
 "body": {
 "model": "gpt-4o-mini",
 "messages": [
 {"role": "system", "content": "你是一个情感分类助手，将文本分类为：正面、负面、中性"},
 {"role": "user", "content": feedback}
 ],
 "max_tokens": 50
 }
 }
 tasks.append(task)

with open("batch_input.jsonl", "w", encoding="utf-8") as f:
 for task in tasks:
 f.write(json.dumps(task, ensure_ascii=False) + "\n")

注意那个 custom_id。这是你后面匹配请求和响应的唯一标识，千万别随便填。我习惯用 task-{index} 或者 {业务}-{时间戳}-{序号}，出问题了能快速定位到具体哪条数据。

数据清洗和结构化提取

这个场景特别爽。

我们从 PDF 里扒了 50 万条合同条款，需要提取甲乙方、金额、日期。用 Batch API，一次性扔上去，第二天早上喝着咖啡看结果。而且 Batch API 单文件最多支持 50,000 个请求，我分 10 个批次就全搞定了。

强烈建议提取类任务用 response_format 强制 JSON 输出：

{
 "body": {
 "model": "gpt-4o",
 "messages": [...],
 "response_format": {
 "type": "json_schema",
 "json_schema": {
 "name": "contract_extraction",
 "schema": {
 "type": "object",
 "properties": {
 "party_a": {"type": "string"},
 "party_b": {"type": "string"},
 "amount": {"type": "number"},
 "date": {"type": "string"}
 },
 "required": ["party_a", "party_b", "amount", "date"]
 }
 }
 }
 }
}

返回的是严格 JSON，后面省了多少正则的事。我之前没加这个参数，解析那块写了快 100 行，各种边界情况。加了之后，4 行。

模型评测和对比

这个场景可能比较少人提。

我去年底做模型选型（对，就是 2024 年 11 月那波 GPT-4o 更新之后），需要对比 GPT-4o、GPT-4o-mini 和 GPT-4-turbo 在 1000 个测试用例上的表现。用 Batch API 同时提交三批，每个模型跑一遍，第二天直接算准确率。

成本让我挺意外的：

• GPT-4o Batch：$6.20
• GPT-4o-mini Batch：$0.45
• GPT-4-turbo Batch：$8.10

最后发现 GPT-4o-mini 在我们分类任务上准确率只比 GPT-4o 低 2.3 个百分点。2.3 个百分点。但成本差了 13 倍。

我把这个数据甩到老板面前，他沉默了三秒，然后说"切到 mini"。就这么简单。省下的钱够我们团队吃一个月海底捞。

那些让我头秃的坑

说完成功案例，该聊聊那些让我加班的坑了。

坑一：速率限制比你想的复杂

OpenAI 的限制分好几个维度。Batch API 虽然独立于实时 API，但也有自己的天花板。

我当时的情况是这样的：Tier 2，以为可以无限提交。第一批 5 万条成功提交，正在那儿美呢，再提交第二批直接报错。查了半天文档才发现，正在处理的批次数量有限制——Tier 2 好像是 10 个，我现在升到 Tier 3 了，但当时确实被这个卡住了。

解决办法：写了个队列管理脚本。

import time
from openai import OpenAI

client = OpenAI()

def submit_batches_with_retry(file_paths, max_concurrent=5):
 active_batches = []
 completed_batches = []
 
 for file_path in file_paths:
 while len(active_batches) >= max_concurrent:
 for batch in active_batches[:]:
 status = client.batches.retrieve(batch.id)
 if status.status in ["completed", "failed", "expired"]:
 active_batches.remove(batch)
 completed_batches.append(status)
 time.sleep(30)
 
 batch_input_file = client.files.create(
 file=open(file_path, "rb"),
 purpose="batch"
 )
 
 batch = client.batches.create(
 input_file_id=batch_input_file.id,
 endpoint="/v1/chat/completions",
 completion_window="24h"
 )
 active_batches.append(batch)
 print(f"Submitted batch {batch.id} for {file_path}")
 
 return completed_batches

自从用了这个，再也没触发过限制。其实就是个简单的生产者-消费者模式，但确实管用。

坑二：JSONL 格式问题，一错全错

这个坑真的恶心。

JSONL 要求每行一个完整 JSON，不能有空行，不能有格式错误。我有一次在最后一行多敲了个回车，就一个回车，整个批次验证阶段就挂了。

更要命的是，如果中间某条数据 JSON 格式有问题，OpenAI 不会告诉你具体哪一行。就给你一个笼统的报错，让你自己去猜。

我后来被迫写了个验证脚本：

def validate_jsonl(file_path):
 errors = []
 with open(file_path, 'r', encoding='utf-8') as f:
 for line_num, line in enumerate(f, 1):
 line = line.strip()
 if not line:
 errors.append(f"Line {line_num}: Empty line")
 continue
 try:
 data = json.loads(line)
 if "custom_id" not in data:
 errors.append(f"Line {line_num}: Missing custom_id")
 if "body" not in data:
 errors.append(f"Line {line_num}: Missing body")
 except json.JSONDecodeError as e:
 errors.append(f"Line {line_num}: JSON decode error - {str(e)}")
 
 if errors:
 print(f"Found {len(errors)} errors:")
 for error in errors[:10]:
 print(f" {error}")
 return False
 return True

现在每次上传前先跑一遍，省了无数重试时间。我觉得这应该是官方提供的基础功能，但他们没做，只能自己搞。

坑三：24 小时不是合同，是愿望

官方说 24 小时内完成。我信了。

然后 GPT-4o 刚发布那周，我的任务排了 30 多个小时。大家都在测新模型，队列挤爆了。另一次是文件太大，180MB，处理时间明显更长。

嗯...这个其实比较复杂。它跟模型热度、文件大小、当前队列长度都有关系。我现在给自己定了个死规矩：需要 4 小时内出结果，就用实时 API 加并发控制。可以等到第二天，才走 Batch。

别抱侥幸心理。我抱过，翻车了。

坑四：结果顺序是乱的

这个坑纯属我没仔细看文档。

Batch API 返回的结果文件是 JSONL，每行有 custom_id、response 和 error。但返回顺序不一定和输入顺序一致。我以为按顺序返回，写了个按索引匹配的逻辑，结果全乱了。全乱了。

正确做法是用 custom_id 建字典：

def parse_batch_results(output_file_path):
 results = {}
 with open(output_file_path, 'r', encoding='utf-8') as f:
 for line in f:
 data = json.loads(line)
 custom_id = data["custom_id"]
 if data.get("error"):
 results[custom_id] = {"error": data["error"]}
 else:
 response_body = data["response"]["body"]
 content = response_body["choices"][0]["message"]["content"]
 results[custom_id] = {"content": content, "usage": response_body.get("usage", {})}
 return results

现在学乖了，拿到结果第一件事就是建映射表。

一个完整案例：100 万条客服对话

让我给你看个真实项目。100 万条客服对话，提取关键信息和情感标签。

方案 A：实时 API + GPT-4o

• 总 tokens：输入 850M + 输出 120M
• 预估成本：$2,125 + $1,200 = $3,325
• 需要自己写并发控制，预计跑 3-4 天
• 代码里全是重试逻辑和断点续传，看着就头疼

方案 B：Batch API + GPT-4o

• 实际成本：$1,062.50 + $600 = $1,662.50
• 分 20 个批次，30 小时内全部完成
• 代码量：大概 200 行，清清爽爽

方案 C：Batch API + GPT-4o-mini

• 实际成本：$127.50 + $72 = $199.50
• 准确率比 GPT-4o 低 1.8%
• 但成本只有 1/8

我们选了方案 C。1.8% 的准确率损失在客服对话场景下完全可以接受——据我了解，人工标注的误差率大概也在 3%-5%，1.8% 根本不叫事。

这个决策省了 90% 的成本。老板很高兴，真请我吃了顿饭。不是什么高级餐厅，就是楼下那家湘菜馆，但好歹不用自己掏钱。

我总结的一些实践习惯

经过这几个月折腾，形成了一些固定做法：

文件组织：

• 每个 JSONL 控制在 10,000-30,000 条，别贪多
• 文件大小不超过 100MB，留点余地
• custom_id 用有意义的命名，出问题能溯源

错误处理：

• 上传前必须跑验证脚本
• 解析结果用 custom_id 做映射，别信顺序
• 失败率超过 5% 就要排查，别得过且过

成本控制：

• 优先考虑 GPT-4o-mini，大多数分类和提取够用了
• 用 max_tokens 卡住输出长度
• 定期看 usage 数据，我每个月 5 号固定检查一次

监控：

• 记录每个批次的提交时间、完成时间
• 统计平均处理时长，方便下次预估
• 失败任务单独记录，分析原因

我写了个小脚本，批次跑完自动发 Slack 通知：

def notify_batch_completion(batch_id):
 batch = client.batches.retrieve(batch_id)
 status = batch.status
 request_counts = batch.request_counts
 
 message = f"""
 📊 Batch {batch_id} 完成
 状态: {status}
 总请求: {request_counts.total}
 成功: {request_counts.completed}
 失败: {request_counts.failed}
 完成时间: {batch.completed_at}
 """
 
 send_slack_notification(message)

不用守着屏幕刷新，该干嘛干嘛去。

也别无脑上 Batch

说了这么多好话，也得说说不适用的场景：

• 实时对话。用户等不了 24 小时，这不用解释。
• 小批量高频。每次几十条但一天提交很多次，管理批次的开销可能比省的钱还大。
• 需要流式输出。Batch API 不支持 streaming，没法做打字机效果。
• 串行依赖任务。每条请求依赖前一条结果的那种，Batch 搞不定。

别为了省钱把业务体验搞崩了。我在这一点上吃过亏——当时把一个本该实时的小功能硬上了 Batch，被用户骂了整整一周。

最后

Batch API 真的是个被严重低估的功能。

我身边好多开发者还在用实时 API 跑大批量任务，一问要么不知道有 Batch，要么以为很复杂懒得搞。其实从实时 API 切到 Batch，代码改动量通常不超过 20%，成本直接砍半。

如果你手头有大批量文本处理的需求，这周就试试。真的。

你们平时怎么优化 LLM API 成本？有没有踩过其他 Batch API 的坑？评论区聊聊，我看到都会回。

标签： #OpenAI #BatchAPI #成本优化 #GPT-4o #开发者工具