力 力达云
教程聚合APItoken成本国产模型资讯工具关于内测申请
← 返回资讯

Batch API 批量调用省钱指南：非实时任务立省 50%

2026-07-08

对于不需要实时返回结果的 AI 任务，切换到 Batch API 是单次改动收益最高的成本优化之一——同样的模型、同样的 prompt，价格直接打对折。我自己带团队做过一次迁移：原本用实时接口跑一批商品评论分类，脚本里就是一个 for 循环挨个调用，账单一直居高不下；改成 Batch 之后除了多写十几行提交和轮询的代码，模型本身没换、prompt 一个字没动，当月账单直接腰斩。这不是玄学，是厂商在定价上明摆着告诉你：只要你能接受”晚点给结果”，它就愿意把这部分算力挪到低峰期排产，成本让利给你。

什么是 Batch API

Batch API（批量异步接口）允许开发者将大量请求打包为一个任务文件统一提交，模型在后台处理完毕后统一返回结果。与实时调用相比：

对比维度实时调用（同步）Batch API（异步）
响应方式立即返回（秒级）延迟返回（通常 1–24 小时）
价格标准价约 50% 折扣（截至 2026-06，以官方为准）
适用场景用户交互、实时问答离线批处理、数据标注、报告生成
并发限制受 RPM 限制通常有独立队列，限制宽松
结果获取直接获取轮询状态 → 下载结果文件

目前 OpenAI 和 Anthropic 均已推出 Batch API，国内部分厂商也在跟进，使用前请查阅各家文档确认支持情况。这里有个容易被忽略的细节：两家大厂的叫法和心智模型并不完全一样。OpenAI 叫 Batch API，你上传的是一个 JSONL 文件，拿到的是 batch.id；Anthropic 那边叫 Message Batches API，提交的是一个请求数组而不是文件，返回的对象结构和字段命名也不同（比如状态字段、结果的拉取方式）。如果你的项目要同时兼容两家，千万别偷懒直接复用同一套解析逻辑，先去翻一遍各自最新的接口文档，字段对不上是真的会在生产环境炸的那种坑，不是本地跑一次就能发现的。

哪些任务适合切换到 Batch API

满足以下条件的任务，切换 Batch API 几乎没有副作用：

  • 离线数据处理：批量对文章、评论、商品描述做分类、打标、摘要
  • 定时报告生成：每天/每周的自动汇总邮件、分析报告
  • 数据标注与清洗：用大模型标注训练数据集
  • 批量翻译：对文档库、产品文案做语言本地化
  • Embedding 批量生成：为 RAG 知识库建索引（部分厂商 Batch Embedding 也有折扣）

不适合的场景：实时客服、流式对话、用户等待结果的所有交互式场景。

判断一个任务能不能切 Batch，我通常就问自己一个问题：结果晚几个小时给到，会不会有人在页面前干等？ 如果答案是”不会，反正是后台跑的定时任务”，那基本可以无脑切。反过来，哪怕看起来是”批量”的活儿，只要产品形态是用户点了一下按钮就盯着转圈等结果，那本质还是实时场景，别为了省钱硬凑 Batch，体验崩了得不偿失。

成本差距到底有多大

拿一个真实量级举例：假设你要对 10 万条用户评论做情感分类，每条评论加上 prompt 大概 150 tokens 输入、20 tokens 输出，模型选 gpt-4o-mini。三种做法的相对成本大致是这样（具体单价会变，以官方最新价目表为准，这里只看倍数关系）：

方案相对成本说明
实时调用旗舰模型基准 100%谁都不推荐，纯浪费
实时调用轻量模型约 10%–15%换模型省的钱最多
Batch 调用轻量模型约 5%–8%换模型 + 打折，双重叠加

能看出来，真正的大头省钱来自”换成更小的模型”，Batch 的 50% 折扣是在此基础上的再叠加，两者不冲突、可以一起用。如果你现在还在用旗舰模型做离线批处理，这是我见过最典型的”钱花在不该花的地方”——批处理任务对响应速度完全不敏感，没理由为速度付费。

基本使用流程

以 OpenAI Batch API 为例，完整流程分三步：

第一步：构造请求文件（JSONL 格式）

{"custom_id": "req-001", "method": "POST", "url": "/v1/chat/completions", "body": {"model": "gpt-4o-mini", "messages": [{"role": "user", "content": "请对以下评论情感分类：好评还是差评？「发货很快，质量不错」"}], "max_tokens": 20}}
{"custom_id": "req-002", "method": "POST", "url": "/v1/chat/completions", "body": {"model": "gpt-4o-mini", "messages": [{"role": "user", "content": "请对以下评论情感分类：好评还是差评？「物流太慢，包装破损」"}], "max_tokens": 20}}

这个 custom_id 是整个流程里最容易被忽视、但出问题最要命的一个字段。它的作用是让你在结果文件里能对上号——结果文件里各行的顺序和你提交的输入文件顺序不保证一致，模型后台是并行处理这批请求的，处理完的顺序取决于各条请求的实际耗时，谁先跑完谁先落盘。所以拿到结果后你必须按 custom_id 做匹配，不能假设第 N 行结果对应第 N 行输入，我见过至少两个团队踩过这个坑：本地小批量测试时凑巧顺序没乱，一上生产环境跑大批次就全对错位了，分类结果和评论文本对不上，还得靠人工回查才发现是这个原因。另外 custom_id 在同一个批次内必须唯一，如果你是用循环里的临时变量拼的（比如某次忘了重置计数器），提交时会直接报错，报错信息通常类似 Duplicate custom_id detected，看到这句话第一反应就是去检查生成 ID 的逻辑有没有跨批次复用。

第二步：提交批次并轮询状态

import openai, time

client = openai.OpenAI()

# 上传文件
with open("requests.jsonl", "rb") as f:
    file = client.files.create(file=f, purpose="batch")

# 创建批次
batch = client.batches.create(
    input_file_id=file.id,
    endpoint="/v1/chat/completions",
    completion_window="24h"
)

# 轮询等待完成
while True:
    batch = client.batches.retrieve(batch.id)
    if batch.status in ("completed", "failed", "expired"):
        break
    time.sleep(60)

批次提交成功后，你应该先看到状态是 validating——这一步模型服务在校验你的 JSONL 格式对不对，一般几分钟内会变成 in_progress，然后开始真正处理。这里说一个大部分教程不会提的细节：如果批次卡在 validating 状态特别久迟迟不动，八成不是排队排到你，而是文件里某一行 JSON 格式有问题——最常见的是 prompt 文本里含有未转义的引号或换行符，导致那一行不是合法 JSON，但服务端给的提示通常很笼统（类似 Invalid file format），不会精确告诉你是第几行错了。我的做法是提交前先在本地用 json.loads() 把 JSONL 文件逐行过一遍，任何一行抛异常就打印行号，这一步能帮你省掉后面干等几十分钟才发现是格式问题的时间。

轮询间隔也有讲究：上面代码写的是 60 秒一次，如果批次量很小（比如几百条），处理可能几分钟就完事，60 秒轮一次不算浪费；但如果是几万条的大批次，处理动辄要几个小时，这时候把轮询间隔缩短到 60 秒反而没意义，纯粹在打接口位浪费一次调用配额，我一般会做成指数退避——刚开始 1 分钟查一次，超过 10 分钟没完成就拉长到 5 分钟一次，超过 1 小时就拉长到 15 分钟一次，同样能及时拿到结果，接口调用次数能省下一大截。

第三步：下载并解析结果

output = client.files.content(batch.output_file_id)
for line in output.text.splitlines():
    result = json.loads(line)
    print(result["custom_id"], result["response"]["body"]["choices"][0]["message"]["content"])

拿到结果后先别急着往数据库里写，务必先扫一遍每一行有没有 error 字段——批次里失败的请求不会让整个批次失败，而是混在结果文件里以错误对象的形式返回，格式大致是 {"custom_id": "req-003", "error": {"code": "...", "message": "..."}}。如果你的解析代码假设每一行都有 response 字段，遇到这种错误行会直接 KeyError 崩掉，所以拿字段前一定要先判断：

output = client.files.content(batch.output_file_id)
failed_ids = []
for line in output.text.splitlines():
    result = json.loads(line)
    if result.get("error"):
        failed_ids.append(result["custom_id"])
        continue
    content = result["response"]["body"]["choices"][0]["message"]["content"]
    print(result["custom_id"], content)

if failed_ids:
    print(f"共 {len(failed_ids)} 条失败，需要重新提交：", failed_ids)

失败的这部分不用整个批次重跑，把 failed_ids 对应的原始请求挑出来单独打包成一个新的小 JSONL 再提交一次就行，成本上也不亏，反正只是失败的那一小撮。

大批量任务怎么分片

如果你的任务量特别大（比如百万级请求），单个 JSONL 文件会受到大小和请求条数的限制，这个上限各家不一样、也会调整，具体数值以官方文档为准，但思路是通用的：把大任务切成多个子批次分别提交。这里有两个容易翻车的地方值得提前打个预防针：

  • custom_id 要带分片前缀：如果每个分片都从 req-001 开始编号，多个分片各自看没问题，但等你把所有分片的结果汇总到一起处理时，不同分片里的 req-001 会互相冲突覆盖。稳妥的做法是编号里带上分片号，比如 shard-03-req-001，汇总时天然不会撞车。
  • 分片之间要留处理间隔再提交下一批：不是必须串行等上一个分片跑完才提交下一个，但如果你一次性把几十个分片全丢进去，很容易撞到账号级别的并发批次数量限制或者排队总量限制，报错信息通常是类似 too many pending batches 这种提示。稳妥做法是控制同时在跑的分片数（比如 5 个一组），跑完一组再提交下一组。

节省成本的实操建议

  • 尽量合并小任务：单个 JSONL 文件可包含数千条请求，避免频繁创建批次产生管理开销。批次数量本身也是有配额的，攒够一定量再提交一次，比零敲碎打提交几十个小批次要划算得多，管理成本（轮询、汇总）也低。
  • 用更小的模型：批处理场景对响应速度不敏感，可以大胆选用轻量模型（如 gpt-4o-mini），与折扣叠加后综合成本可降至旗舰实时调用的 10% 以下。判断能不能用轻量模型的标准很简单：先拿一小批数据（比如 200 条）分别用旗舰模型和轻量模型跑一遍，人工抽查对比结果质量，如果轻量模型的错误率能接受，就没理由多花几倍的钱。
  • 设置合理的 max_tokens：批处理任务通常输出格式固定，严格限制输出长度避免浪费。这一条经常被忽略：如果任务是”输出好评/差评”这种定长分类结果，max_tokens 设成 20 就足够，设成默认值或者不设，遇到模型偶尔”话痨”多输出几句解释文字，几十万条请求累积下来是一笔不小的额外开销。
  • 监控失败率：批次中部分请求可能失败，结果文件里有 error 字段，需要重试逻辑。如果发现失败率异常偏高（比如超过 5%），别急着无脑重试，先看错误码是什么类型——如果是内容审核相关的拒答，重试也没用，得先看是不是 prompt 本身有敏感内容；如果是模型侧的临时性错误，简单重试通常就能解决。

常见问题

Batch API 的最长等待时间是多少？
OpenAI 的 completion_window 最大为 24 小时，Anthropic 也类似。如果 24 小时内未处理完，批次会标记为 expired，需要重新提交。实际大多数批次远早于上限完成。

国内大模型厂商有 Batch API 吗？
截至 2026-06，阿里云百炼、字节跳动火山引擎等均在不同程度支持批量推理，但折扣力度和接口设计各有差异，建议查阅各家最新文档。

Batch API 支持流式输出吗？
不支持。Batch API 的本质是异步处理，结果以文件形式返回，天然不支持 streaming。

批次提交后可以取消吗？
可以。在批次进入处理前，可以调用取消接口；一旦开始处理则无法取消。

延伸阅读：