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 建 jobC 家干脆只给你一个异步接口自己轮询。代码里全是适配层维护成本极高。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的 SDKOpenRouter 兼容 OpenAI 的接口格式。pip install openai requests密钥配置我强烈建议用环境变量别硬编码在代码里。OpenRouter 的密钥在控制台生成格式是一串以sk-or-开头的字符串。export OPENROUTER_API_KEYsk-or-你的密钥提示密钥泄露是高频事故。如果你把代码传到公开仓库务必先确认密钥没有跟着上去。我见过太多人因为这一条被刷爆额度。4.2 构造批量请求体批量请求的核心是把多条独立请求打包。每条请求包含custom_id和bodybody里就是标准的 messages 结构。import json def build_batch_item(custom_id, prompt, modeldeepseek/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 fsummary_batch01_{idx:06d} items.append(build_batch_item(cid, f请为以下内容生成摘要\n{text})) with open(batch_input.jsonl, w, encodingutf-8) as f: for item in items: f.write(json.dumps(item, ensure_asciiFalse) \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: fBearer {API_KEY}, Content-Type: application/json } # 提交 with open(batch_input.jsonl, rb) as f: resp requests.post( f{BASE_URL}/batches, headersheaders, 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}, headersheaders) 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, headersheaders ) 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, encodingutf-8) as f: for item in failed_items: f.write(json.dumps(item, ensure_asciiFalse) \n) print(f已生成重试文件共 {len(failed_items)} 条)这套重试逻辑我封装成了一个函数跑批量任务的时候直接调用省心很多。5. 常见问题与排查技巧实录5.1 提交就报错先查格式再查权限批量任务提交失败八成是格式问题。JSONL 文件最常见的坑是最后一行没有换行符、某一行 JSON 语法错误、字段名拼错。我建议提交前先用脚本校验一遍每一行能不能正常解析。def validate_jsonl(path): with open(path, r, encodingutf-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 常见问题速查表问题现象可能原因排查方向提交报 400JSONL 格式错误逐行校验 JSON 语法提交报 401密钥无效或过期重新生成密钥提交报 402余额不足充值后重试卡在 validating文件过大或格式校验慢拆分文件减少单批条数卡在 in_progress排队或模型负载高等待或换模型结果匹配错乱custom_id 重复或含特殊字符规范 id 命名规则失败率高单条超限或触发过滤检查输入长度和内容成本没降量小或重试多做成本测算控制重试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 调用。这样离线部分成本压到最低在线部分延迟也小。另一个方向是做多模型对比。同一批数据分别提交给几个不同的模型跑完之后对比输出质量用来做模型选型的依据。批量接口让这种对比实验的成本变得可接受以前要花几百块的实验现在几十块就能跑。还有一个思路是把批量任务做成定时调度。比如每天凌晨自动把前一天的新数据打包提交早上上班的时候结果已经躺在数据库里了。这种睡后处理的模式特别适合内容类和数据类项目。批量推理这个能力本质上是在成本和时间之间做了一次交换。你愿意多等一会儿就能少花一半的钱。对于量大、不急、重复性高的任务这笔账怎么算都划算。真正要花心思的地方是把流程做稳、把错误处理好、把成本算清楚。这几点做到了批量接口就是你的省钱利器。