批量任务(Batch API)使用教程
批量任务就是:把几十、几百条请求写进一个文件,一次交给 Li API,它在后台慢慢帮你跑完,跑完你再下载结果。适合“同一个提示词,对一大批数据各跑一遍”的活,比如给语料打标签、批量摘要、评测集打分、批量翻译。
- 价格和平时一样:每条请求按正常价格扣费,没有折扣,也不加价。
- 不用守着:提交后可以关电脑,24 小时内跑完,一般几分钟到几小时。
- 不会被“高并发”报错打断:后台会自动排队、遇到上游繁忙自动等一会再试,你不用自己写重试。
- 结果保存 7 天:请在 7 天内下载,过期自动删除。
- 不适合聊天和写代码:Claude Code、Codex、Cherry Studio 这类要马上看到回复的,继续用平时的方式,不要用批量任务。
第一部分:准备输入文件
输入文件是一个 .jsonl 文件:一行就是一条请求,每行都是一段完整的 JSON。下面是 3 条请求的例子,可以直接复制保存为 input.jsonl:
{"custom_id": "task-1", "method": "POST", "url": "/v1/chat/completions", "body": {"model": "deepseek-v4-flash", "messages": [{"role": "user", "content": "用一句话介绍长城"}]}}
{"custom_id": "task-2", "method": "POST", "url": "/v1/chat/completions", "body": {"model": "deepseek-v4-flash", "messages": [{"role": "user", "content": "用一句话介绍故宫"}]}}
{"custom_id": "task-3", "method": "POST", "url": "/v1/chat/completions", "body": {"model": "deepseek-v4-flash", "messages": [{"role": "user", "content": "用一句话介绍兵马俑"}]}}
每一行要注意 4 点:
custom_id你自己起的编号,同一个文件里不能重复。结果里会带着它,方便你对上号。url同一个文件里所有行必须一样,而且要和第二部分提交时写的接口一致。body就是平时直接调接口时发的那段内容,model必须是你的密钥能用的模型(在控制台“模型广场”能看到)。- 不用写
stream写了也会被自动改成不流式。
支持的接口:/v1/chat/completions、/v1/responses、/v1/embeddings、/v1/completions、/v1/messages(Claude 格式)。
第二部分:用 Python 提交
先安装官方 SDK(只需一次):
pip install openai
然后新建一个 submit.py,把 你的密钥 换成控制台“令牌”页面里 sk- 开头的那串:
from openai import OpenAI
client = OpenAI(api_key="你的密钥", base_url="https://api.yulitongxing.com/v1")
# 1. 上传输入文件
f = client.files.create(file=open("input.jsonl", "rb"), purpose="batch")
# 2. 创建批量任务
batch = client.batches.create(
input_file_id=f.id,
endpoint="/v1/chat/completions",
completion_window="24h",
)
print("任务编号:", batch.id, "状态:", batch.status)
运行 python submit.py,记下打印出来的任务编号(batch_ 开头)。
第三部分:查看进度、下载结果
新建 check.py:
from openai import OpenAI
client = OpenAI(api_key="你的密钥", base_url="https://api.yulitongxing.com/v1")
batch = client.batches.retrieve("这里填任务编号")
print("状态:", batch.status, "进度:", batch.request_counts)
if batch.output_file_id:
open("output.jsonl", "w", encoding="utf-8").write(client.files.content(batch.output_file_id).text)
print("成功的结果已保存到 output.jsonl")
if batch.error_file_id:
open("errors.jsonl", "w", encoding="utf-8").write(client.files.content(batch.error_file_id).text)
print("失败的请求已保存到 errors.jsonl")
状态是什么意思:
in_progress:正在跑,隔一会再查。completed:跑完了,按上面的代码下载结果。failed:没跑起来,看batch.errors,里面会写第几行有什么问题。cancelled:你取消了,已经跑完的部分照样能下载。expired:24 小时没跑完,已经跑完的部分照样能下载,没跑的在 errors.jsonl 里。
结果文件也是一行一条,用 custom_id 对上号;回复内容在 response.body 里,和平时调接口返回的一模一样。
想中途取消:client.batches.cancel("任务编号")。
使用限制
- 单个输入文件最大 20MB、最多 2000 行。
- 同一个密钥同时最多 3 个任务在跑。
- 同一个密钥最多存 100MB 文件,旧文件可以用
client.files.delete("文件编号")删掉。 - 只有提交任务的那把密钥能看到这个任务和结果,换一把密钥就看不到了。
- 如果你的令牌设置了“IP 白名单”,请从白名单内的电脑提交。
常见问题
为什么跑得比我自己并发调用慢?
这是故意的。后台只用很少的并发慢慢跑,遇到上游繁忙就等一会再试,这样既不会把号池挤爆,也不会给你返回一堆“高并发”报错。换来的是稳,不是快;急用的少量请求请直接调接口。
提交后马上变成 failed,提示 model_not_found
那一行写的模型你的密钥用不了,或者模型名拼错了。到控制台“模型广场”核对名字,改好后重新上传、重新提交。
跑到一半整个任务变成 failed,提示 insufficient_user_quota
账户余额不够了。为了不给你刷出一大堆失败记录,余额不足时整个任务会立刻停下,剩下的行记为失败。充值后,把没成功的那些行重新做成一个文件再提交一次。
errors.jsonl 里有一行写着 upstream_timeout
那条请求等了 10 分钟还没回复,系统自动重试过一次仍然没有回复。这种情况下上游有可能已经处理并扣费了,只是回复没传回来。为了防止重复扣费,系统不会无限重试,请你自己决定要不要重新提交这一行。
批量任务比平时便宜吗?
不便宜,和平时价格一样。它解决的是“大批量请求不被高并发报错打断、不用自己写重试”的问题。
结果能保存多久?
7 天。7 天后输入文件和结果文件都会自动删除,请及时下载。
批量任务用我哪个密钥扣费?
提交任务用的那把密钥。任务里每一条请求都用这把密钥调用,扣费记录在控制台“使用日志”里能看到,和平时一样。