OpenAI的Batch API为批量文本处理提供了一条低成本路径:把多个请求写进一个JSONL文件,通过上传接口创建批处理任务,服务端会在后台排队执行,完成后再下载结果文件。同步调用Chat Completions接口时,每个请求都会实时返回并计费,而Batch API的单价只有同步的一半,这对有大量非实时数据要处理的团队来说,省下来的成本相当可观。接下来从请求文件准备、任务提交、结果获取到常见的坑,完整走一遍流程。

Batch API的适用场景与计费逻辑
同步接口的设计目标是低延迟,输入一个prompt,马上拿到completion。这种模式适合聊天机器人、实时助手等对响应时间敏感的产品。但很多工作并不需要毫秒级返回,比如给一万条商品描述生成摘要、对测试集跑模型评测、从语料中提取结构化字段,这类任务完全可以在后台慢慢执行,第二天再取结果也不影响进度。
Batch API正是面向这种异步场景。它的计费方式是按照每个batch中实际消耗的token数计算,但单价为标准同步接口的一半。以gpt-3.5-turbo为例,同步输入每百万token价格如果是x美元,Batch API就按0.5倍计费。这个半价策略覆盖了输入和输出token,并且batch任务不会占用常规的RPM/TPM速率限制,你可以一次性提交几千个请求而不用担心触发限流。
要注意的是,批处理任务有完成时间窗口,官方文档提到通常会在24小时内完成,实际大多数任务在几小时内即可取回结果。如果你的业务对时效性要求很高,比如用户等待几秒就必须出结果,那Batch API并不适合。
准备请求文件并创建批处理任务
使用Batch API的第一步是在本地构造一个JSONL文件,每一行对应一个独立的请求。每个请求对象需要包含custom_id字段作为唯一标识,以及标准的Chat Completions参数,例如model、messages。custom_id可以是你自己的业务主键,后面取结果时靠它把响应和原始数据对应起来。下面是一个请求文件示例,文件路径在Windows下可以保存为C:\batch\input.jsonl。
{"custom_id": "request-1", "method": "POST", "url": "/v1/chat/completions", "body": {"model": "gpt-3.5-turbo", "messages": [{"role": "system", "content": "You are a helpful assistant."}, {"role": "user", "content": "Summarize the product description."}], "temperature": 0.2}}
{"custom_id": "request-2", "method": "POST", "url": "/v1/chat/completions", "body": {"model": "gpt-3.5-turbo", "messages": [{"role": "user", "content": "Extract key features from this text."}]}}
准备好文件后,调用Files上传接口,把JSONL文件上传到OpenAI,获得一个file id。然后调用Batch创建接口,指定input_file_id和endpoint,以及completion_window。下面以Python为例,展示上传和创建批处理的完整代码。代码中需要替换你的API密钥,建议通过环境变量读取,不要硬编码在源码里。
import os
from openai import OpenAI
client = OpenAI(api_key=os.environ.get("OPENAI_API_KEY"))
# 上传输入文件
with open(r"C:\batch\input.jsonl", "rb") as f:
file_obj = client.files.create(file=f, purpose="batch")
print(f"上传成功,file id: {file_obj.id}")
# 创建批处理任务
batch = client.batches.create(
input_file_id=file_obj.id,
endpoint="/v1/chat/completions",
completion_window="24h"
)
print(f"批处理任务已创建,batch id: {batch.id}")
上传文件后可以在OpenAI控制台看到文件状态,或者在代码中轮询batch状态。状态流转通常是从validating到in_progress,再到finalizing,最后completed。如果输入文件格式错误,可能在validating阶段就失败,这时要检查每一行的JSON是否合法,以及body里是否缺少model等必填字段。
获取结果并解析输出
批处理任务完成之后,OpenAI会生成一个输出文件,包含每个请求的执行结果。你可以通过batch对象拿到output_file_id,再调用Files下载接口读取内容。输出文件同样是JSONL格式,每一行包含custom_id、response字段,如果请求失败,还会有error字段。这里的关键是用custom_id把结果和原始输入对应起来,避免顺序错乱。
import json
# 假设batch对象已经从客户端获取
batch = client.batches.retrieve(batch.id)
if batch.status == "completed":
output_file_id = batch.output_file_id
content = client.files.content(output_file_id).content
# content是bytes,按行解析
lines = content.decode("utf-8").splitlines()
for line in lines:
data = json.loads(line)
request_id = data["custom_id"]
if data.get("error"):
print(f"{request_id} 失败: {data['error']}")
else:
message = data["response"]["body"]["choices"][0]["message"]["content"]
print(f"{request_id} 结果: {message[:80]}...")
如果任务量很大,建议分批下载和解析,避免一次性加载到内存。输出文件的行顺序不一定与输入文件一致,所以不能依赖行号匹配。每个请求的token用量会记录在response的usage字段中,方便你做成本统计。
处理失败请求时,先根据error类型判断是模型拒绝、超时还是网络问题。对于可重试的错误,可以把失败的custom_id抽出来,重新构造一个JSONL文件再次提交,注意不要重复计费。
成本节省与常见问题规避
Batch API的50%成本节省来自异步调度带来的资源利用优化。对于大批量任务,这能直接变成可观的金额。但成本节省的前提是任务真正适合异步处理。如果一个批处理任务提交后超过24小时还没完成,应当检查文件规模和当前队列压力,必要时将大文件拆分成多个小批次提交,避免单批次过大影响完成时间。
另一个常见的坑是文件格式错误。JSONL文件要求每一行都是一个完整的JSON对象,行尾不能有多余逗号,编码必须是UTF-8。在Windows上生成文件时,建议使用支持Linux换行符的编辑器,或通过Python的json.dumps逐行写入。不要把整个列表序列化成一个JSON数组,那会导致整批任务验证失败。
此外,Batch API不保证请求按照输入顺序执行,所以设计custom_id时要确保每个请求彼此独立,没有依赖关系。如果某些请求之间有先后顺序要求,需要把它们拆成多个批处理任务,在前一个任务完成后再提交下一批。
总的来说,OpenAI Batch API为离线批量处理提供了一个简单但有效的成本优化方案。只要控制好文件格式、合理拆分任务、做好错误重试,就能用一半的价格完成大规模文本生成工作。
OpenAI Batch API异步批量处理成本节省修改时间:2026-10-06 02:39:51