OpenRouter Batch API 批量推理实战：半价成本与工程化避坑指南

1. 批量推理这件事，为什么值得单独聊

做AI应用开发的朋友大概率都遇到过这种场景：白天用户请求稀稀拉拉，晚上跑数据清洗、内容打标、离线摘要的时候，几万条文本要过一遍大模型。这时候你会发现两件事——第一，钱烧得比想象中快；第二，接口的并发限制卡得人难受。OpenRouter 推出 Batch API 这件事，本质上就是冲着这两个痛点来的：把不要求实时返回的推理任务打包提交，换取接近半价的成本，同时避开在线接口的速率限制。

我自己手上有几个项目长期跑批量任务，比如给历史文章做结构化抽取、给商品评论做情感分类、给客服对话做质量打分。这些任务的共同点是：结果不急着要，但量大、重复性高、对成本极度敏感。Batch API 这种模式其实在行业里不算新鲜，但 OpenRouter 把它做成了一个统一入口，能横跨多家模型供应商，这就有点意思了。你不用为每个模型单独对接一套批处理流程，一个 key、一套格式，就能把任务分发到不同模型上。

这篇文章适合谁看？如果你正在用 OpenRouter 做在线推理，想进一步压成本；或者你手头有大量离线任务，正在纠结用哪家 API 更划算；再或者你只是听说过 Batch API 但没实际跑过，想搞清楚它和普通调用的区别、坑在哪里——那这篇内容应该能帮你省下不少试错时间。我会从设计思路、核心机制、实操流程到踩坑记录，完整讲一遍。

2. Batch API 的整体设计与思路拆解

2.1 为什么批量能便宜：从资源调度说起

要理解半价这件事，得先明白在线推理和批量推理在服务端的成本结构差异。在线接口的核心约束是延迟——用户发一个请求，服务端必须在几百毫秒到几秒内返回。为了满足这个约束，供应商必须预留大量算力，保证高峰期也不排队。这些预留的算力在低谷期就是浪费的。

批量推理反过来：你告诉服务端"我不急，你什么时候有空什么时候算"，服务端就可以把这些任务塞进低谷期的空闲算力里，甚至可以把多个请求合并成一个大 batch 一起前向计算，GPU 利用率能拉高好几倍。单位 token 的边际成本降下来了，供应商自然愿意让出一部分利润给你。这就是半价的底层逻辑，不是什么营销补贴，而是实打实的资源错峰。

OpenRouter 作为聚合层，它的角色是把你的批量任务路由到背后各家供应商的批处理通道。这里有个关键点：不同供应商对批处理的定义不一样，有的支持真正的离线队列，有的只是把并发限制放宽。OpenRouter 做了一层抽象，让你用统一的接口提交，但底层的执行策略还是取决于你选的模型。

2.2 统一入口的价值：一个 key 打通多家模型

我早期做批量任务的时候，最烦的就是每换一个模型就要重写一遍提交逻辑。A 家的批处理用 JSONL 上传文件，B 家要用 SDK 建 job，C 家干脆只给你一个异步接口自己轮询。代码里全是适配层，维护成本极高。

OpenRouter 的 Batch API 把这一层抹平了。你提交的请求体格式和普通 chat completions 基本一致，只是多了一个批量的外壳。返回的是一个 job id，你拿着这个 id 去轮询状态，完成后拉取结果。这套模式对开发者很友好，因为学习成本几乎为零——你已经会调 OpenRouter 的普通接口了，批量接口就是换个 endpoint 的事。

提示：统一入口不等于统一行为。不同模型对批量任务的最大条数、单条 token 上限、超时时间可能不同，提交前最好查一下目标模型的限制，别一股脑塞十万条进去。

2.3 什么任务适合走批量，什么任务千万别走

这是我最想强调的一点。Batch API 不是万能的，用错场景反而添乱。

适合批量的任务有几个特征：结果可以延迟交付（分钟级到小时级都能接受）、任务之间相互独立、单条请求的输入输出规模可控。典型的就是数据标注、内容审核预筛、离线翻译、批量摘要、embedding 生成这类。

不适合的任务也很明确：任何需要实时反馈的交互，比如聊天机器人、在线搜索补全、实时推荐。这些场景用户等着结果，你走批量通道等于让用户干等，体验直接崩掉。另外，有严格顺序依赖的任务也不适合，比如多轮对话的后续轮次依赖前一轮输出，批处理没法保证执行顺序。

我一般会用一个简单的判断标准：如果这个任务的延迟容忍度超过 5 分钟，且调用量在千次以上，就值得考虑批量；否则老老实实走在线接口。

3. 核心机制与关键参数解析

3.1 提交、轮询、拉取：三段式生命周期

Batch API 的交互模型是典型的异步三段式，理解这个生命周期对排查问题很关键。

第一阶段是提交。你把一批请求组织成一个数组，每个元素包含一个自定义的custom_id和标准的请求体。custom_id是你自己定义的标识符，用来在结果里对应回原始请求。这个字段非常重要，因为批量返回的结果顺序不保证和提交顺序一致，你必须靠custom_id来匹配。

第二阶段是轮询。提交成功后你会拿到一个 batch job 的 id，然后定期查询这个 job 的状态。状态一般有几种：validating（校验中）、in_progress（处理中）、completed（完成）、failed（失败）、cancelled（取消）、expired（超时）。轮询频率别太高，我一般 30 秒到 1 分钟查一次，太频繁纯属浪费请求。

第三阶段是拉取结果。job 完成后，结果通常以文件或分页列表的形式提供。每条结果里带着custom_id、状态、以及模型返回的内容。失败的条目会单独标记，你可以只重跑失败的部分，不用整批重来。

3.2 custom_id 的设计技巧

custom_id看起来是个小细节，但设计不好会给你后面带来大麻烦。我的经验是：custom_id要满足唯一性、可追溯性、可解析性三个要求。

唯一性不用多说，重复的 id 会导致结果匹配混乱。可追溯性指的是你看到这个 id 能知道它对应哪条原始数据，比如用数据库主键或者业务编号。可解析性指的是 id 本身最好带一点结构信息，方便你后续做分组统计。

我常用的格式是{业务前缀}_{批次号}_{序号}，比如review_20240501_000123。这样一眼就能看出这条数据属于哪个业务、哪个批次、第几条。别用纯 UUID，虽然唯一但完全没法追溯，出问题的时候你会很痛苦。

3.3 成本计算：半价到底省多少

半价是个笼统的说法，实际省多少取决于模型和 token 结构。我拿一个真实项目算过账：一个内容摘要任务，输入平均 800 token，输出平均 200 token，总共 5 万条。

按在线价格，假设某模型输入 1 元/百万 token、输出 2 元/百万 token，那么总成本是 50000 × (800×1 + 200×2) / 1000000 = 50000 × 1200 / 1000000 = 60 元。走批量半价就是 30 元。单看一次不多，但这类任务往往是每周甚至每天跑，一年下来就是几千块的差距。

注意：半价通常只针对 token 费用，有些供应商对批量任务还会收额外的存储费或文件处理费，虽然金额很小，但算总账的时候别漏掉。

3.4 并发与限流：批量不等于无限

很多人以为走了批量通道就没有并发限制了，这是个误解。批量通道放宽的是在线接口那种严格的 RPM（每分钟请求数）限制，但供应商对单个 batch job 的总量、同时进行的 job 数量仍然有约束。

我遇到过的情况是：单个 job 最多 5 万条请求，同时最多 3 个 job 在跑。超过这个数就得排队或者分批提交。所以如果你的任务量特别大，比如上百万条，需要提前规划好分批策略，别指望一个 job 搞定。

4. 实操流程：从零跑通一个批量任务

4.1 环境准备与密钥配置

先把基础环境搭好。我用 Python 演示，因为生态最成熟。需要装requests或者直接用openai的 SDK（OpenRouter 兼容 OpenAI 的接口格式）。

pip install openai requests

密钥配置我强烈建议用环境变量，别硬编码在代码里。OpenRouter 的密钥在控制台生成，格式是一串以sk-or-开头的字符串。

export OPENROUTER_API_KEY="sk-or-你的密钥"

提示：密钥泄露是高频事故。如果你把代码传到公开仓库，务必先确认密钥没有跟着上去。我见过太多人因为这一条被刷爆额度。

4.2 构造批量请求体

批量请求的核心是把多条独立请求打包。每条请求包含custom_id和body，body里就是标准的 messages 结构。

import json def build_batch_item(custom_id, prompt, model="deepseek/deepseek-chat"): return { "custom_id": custom_id, "method": "POST", "url": "/v1/chat/completions", "body": { "model": model, "messages": [ {"role": "system", "content": "你是一个专业的内容摘要助手。"}, {"role": "user", "content": prompt} ], "max_tokens": 500, "temperature": 0.3 } } items = [] for idx, text in enumerate(my_texts): cid = f"summary_batch01_{idx:06d}" items.append(build_batch_item(cid, f"请为以下内容生成摘要：\n{text}")) with open("batch_input.jsonl", "w", encoding="utf-8") as f: for item in items: f.write(json.dumps(item, ensure_ascii=False) + "\n")

这里用 JSONL 格式，每行一个 JSON 对象。为什么用 JSONL 而不是一个大 JSON 数组？因为 JSONL 支持流式读取，几万条数据不会一次性占满内存，而且单行出错不影响其他行。

4.3 提交任务与轮询状态

提交任务后拿到 job id，然后写一个轮询循环。

import time import requests API_KEY = os.environ["OPENROUTER_API_KEY"] BASE_URL = "https://openrouter.ai/api/v1" headers = { "Authorization": f"Bearer {API_KEY}", "Content-Type": "application/json" } # 提交 with open("batch_input.jsonl", "rb") as f: resp = requests.post( f"{BASE_URL}/batches", headers=headers, files={"file": ("batch_input.jsonl", f, "application/jsonl")} ) job = resp.json() job_id = job["id"] print(f"任务已提交，job_id={job_id}") # 轮询 while True: status_resp = requests.get(f"{BASE_URL}/batches/{job_id}", headers=headers) status = status_resp.json() state = status["status"] print(f"当前状态：{state}") if state in ("completed", "failed", "cancelled", "expired"): break time.sleep(30)

轮询间隔我设的 30 秒。实测下来，小批量任务（几千条）通常几分钟就完成，大批量可能要几十分钟甚至更久。别把间隔设得太短，没意义还增加无谓的请求。

4.4 拉取结果与错误处理

任务完成后拉取结果文件，逐行解析，用custom_id匹配回原始数据。

result_resp = requests.get( f"{BASE_URL}/batches/{job_id}/results", headers=headers ) success_count = 0 fail_count = 0 results_map = {} for line in result_resp.text.strip().split("\n"): record = json.loads(line) cid = record["custom_id"] if record.get("error"): fail_count += 1 results_map[cid] = {"status": "failed", "error": record["error"]} else: success_count += 1 content = record["response"]["body"]["choices"][0]["message"]["content"] results_map[cid] = {"status": "success", "content": content} print(f"成功 {success_count} 条，失败 {fail_count} 条")

失败条目一定要单独收集起来，分析失败原因。常见的失败原因包括：单条请求超 token 上限、内容触发了安全过滤、模型临时不可用。前两种需要你修改数据后重跑，第三种直接重试就行。

4.5 失败重试的批处理策略

失败重试不要整批重跑，那样既浪费钱又浪费时间。我的做法是把失败条目单独抽出来，组成一个新的小批次重新提交。如果某个条目连续失败三次，就标记为人工介入，别再自动重试了。

failed_items = [item for item in items if results_map[item["custom_id"]]["status"] == "failed"] if failed_items: with open("retry_input.jsonl", "w", encoding="utf-8") as f: for item in failed_items: f.write(json.dumps(item, ensure_ascii=False) + "\n") print(f"已生成重试文件，共 {len(failed_items)} 条")

这套重试逻辑我封装成了一个函数，跑批量任务的时候直接调用，省心很多。

5. 常见问题与排查技巧实录

5.1 提交就报错：先查格式再查权限

批量任务提交失败，八成是格式问题。JSONL 文件最常见的坑是：最后一行没有换行符、某一行 JSON 语法错误、字段名拼错。我建议提交前先用脚本校验一遍每一行能不能正常解析。

def validate_jsonl(path): with open(path, "r", encoding="utf-8") as f: for i, line in enumerate(f, 1): line = line.strip() if not line: continue try: json.loads(line) except json.JSONDecodeError as e: print(f"第 {i} 行格式错误：{e}") return False return True

如果格式没问题还是报错，那就是权限或额度问题。检查密钥是否有效、账户余额是否充足。OpenRouter 的余额不足会直接拒绝提交，报错信息有时候不够明确，容易让人误以为是格式问题。

5.2 任务卡在 in_progress 不动

这种情况我遇到过几次，原因各不相同。最常见的是任务量太大，供应商那边排队。其次是某个模型临时负载高，批处理通道被降级。还有一种情况是单条请求的max_tokens设得太大，导致整体处理时间拉长。

排查思路：先看任务提交了多久，如果超过预期时间的两倍，可以考虑取消重提。如果反复卡住，换一个模型试试，可能是特定供应商的问题。另外，把max_tokens调到一个合理值，别动不动就设几千，输出长度直接影响处理时间。

5.3 结果对不上号：custom_id 的坑

结果匹配错乱，几乎都是custom_id的问题。要么是重复了，要么是提交时被截断，要么是解析时没做去空格处理。我踩过一次坑：custom_id里带了空格，结果返回的时候空格被规范化了，导致匹配失败。从那以后我规定custom_id只能用字母、数字和下划线。

还有一个隐蔽的坑：如果你的原始数据里有重复内容，而你又用内容哈希做custom_id，那重复内容会生成相同的 id，结果就乱了。所以custom_id一定要包含一个全局唯一的序号。

5.4 成本没降下来：检查这几个地方

有人跑完批量发现没省多少钱，通常是这几个原因。第一，任务量太小，批量的固定开销摊薄不了。第二，失败重试次数太多，重试的请求可能按在线价格计费。第三，选的模型本身不支持批量折扣，或者折扣比例低于预期。第四，输入输出 token 结构不理想，比如输入极短输出极长，而折扣主要打在输入侧。

我的建议是：跑批量之前先用一小批数据做成本测算，确认折扣确实生效再放量。别一上来就几万条，跑完才发现不划算。

5.5 常见问题速查表

6. 批量任务工程化的几个经验

6.1 把批量流程封装成可复用的管道

跑通一次批量不难，难的是把它变成稳定可复用的流程。我的做法是封装一个BatchPipeline类，把构造请求、提交、轮询、拉取、重试这几个环节串起来，对外只暴露一个run(items)方法。这样每次有新任务，我只需要准备数据，剩下的交给管道。

管道里我会加几个关键设计：状态持久化（把 job_id 和中间状态存到本地文件或数据库，防止程序崩溃后任务丢失）、断点续跑（重启后能从上次的状态继续）、日志记录（每一步都打日志，方便回溯）。这些在一次性脚本里可以省，但生产环境里一个都不能少。

6.2 分批策略：别把鸡蛋放一个篮子

单批条数不是越多越好。批太大，一旦失败整批重来，损失大；批太小，提交和轮询的固定开销占比高。我的经验值是单批 5000 到 20000 条之间，具体看单条的平均 token 量。如果单条很长，就取小值；单条很短，可以取大值。

另外，我会把不同优先级的任务分开批次。比如紧急的走小批次快速出结果，不紧急的走大批次慢慢跑。混在一起会导致紧急任务被拖慢。

6.3 监控与告警：别等跑完才发现问题

批量任务跑起来之后，人不可能一直盯着。我会加一个简单的监控：每隔一段时间检查 job 状态，如果失败率超过阈值（比如 10%），就发告警。告警渠道用邮件或者即时通讯工具的机器人，别搞太复杂。

监控指标我关注三个：完成进度、失败率、平均处理时长。进度用来判断还要等多久，失败率用来判断要不要干预，处理时长用来发现异常（比如突然变慢可能是供应商出问题了）。

6.4 数据预处理：批量任务的质量源头

批量任务的结果质量，很大程度上取决于输入数据的质量。我见过太多人把原始数据直接扔进去，结果模型输出一堆垃圾。预处理要做的事包括：清理乱码和特殊字符、截断超长文本、统一格式、过滤空内容。

截断这一步特别重要。单条请求超过模型的上下文上限会直接失败，与其让它失败再重试，不如提前截断。截断策略我一般用"保留头部 + 保留尾部 + 中间省略"，因为很多文本的关键信息在开头和结尾。

7. 我踩过的几个真实坑

第一个坑是密钥管理。早期我把密钥写在代码里，结果代码同步到团队仓库的时候忘了排除，虽然发现得早没造成损失，但吓出一身冷汗。从那以后所有密钥一律走环境变量，代码里只留读取逻辑。

第二个坑是custom_id用了中文。当时觉得中文可读性好，结果某些环节编码处理不一致，导致匹配失败。现在我的custom_id只用 ASCII 字符，可读性靠日志里的映射表来补。

第三个坑是没做成本测算就放量。有一次跑一个翻译任务，我以为批量能省一半，结果那个模型的批量折扣只有三成，加上重试的开销，实际只省了两成。后来我养成了习惯：任何批量任务先跑 100 条测成本，确认折扣符合预期再放量。

第四个坑是轮询太频繁。我一开始设的 5 秒轮询一次，结果一个跑了半小时的任务，光轮询就发了几百个请求。虽然轮询请求本身不贵，但没必要。现在统一 30 秒起步，长任务用指数退避。

8. 批量推理还能怎么扩展

Batch API 跑通之后，能玩的花样其实不少。我最近在尝试的一个方向是把批量任务和向量数据库结合起来：先用批量接口给海量文档生成 embedding，存进向量库，然后在线查询的时候只走一次 embedding 调用。这样离线部分成本压到最低，在线部分延迟也小。

另一个方向是做多模型对比。同一批数据分别提交给几个不同的模型，跑完之后对比输出质量，用来做模型选型的依据。批量接口让这种对比实验的成本变得可接受，以前要花几百块的实验，现在几十块就能跑。

还有一个思路是把批量任务做成定时调度。比如每天凌晨自动把前一天的新数据打包提交，早上上班的时候结果已经躺在数据库里了。这种"睡后处理"的模式特别适合内容类和数据类项目。

批量推理这个能力，本质上是在成本和时间之间做了一次交换。你愿意多等一会儿，就能少花一半的钱。对于量大、不急、重复性高的任务，这笔账怎么算都划算。真正要花心思的地方，是把流程做稳、把错误处理好、把成本算清楚。这几点做到了，批量接口就是你的省钱利器。