我们做 Claude API 开发时都会遇到同一个问题单条请求处理少量文本很轻松可一旦要跑几千条甚至几万条文本比如批量翻译商品描述、给用户评论打标签、对日志做安全巡检一条一条同步调用 API 不仅慢而且成本容易被忽视。网上关于 Claude 同步调用的教程很多但专门讲 Batch Processing批处理的中文系统教程却不多。这篇文章会从一个完整可运行的批量文本摘要项目入手带你理解 Claude 批处理的核心工作流掌握输入文件构造、任务提交、状态轮询、结果拉取全链路并附上高频报错与工程建议适合刚接触 Claude API 的初学者也适合正在做成本优化和数据批处理的后端开发。1. 背景与核心概念为什么 AI 调用需要批处理1.1 从同步调用说起先看一个最常见的调用方式你有一个 Python 脚本循环读取文本列表每次调用 Claude 接口生成摘要拿到结果后存起来。代码如下# 文件路径sync_demo.py from anthropic import Anthropic client Anthropic() texts [第一段文本, 第二段文本, 第三段文本] results [] for text in texts: message client.messages.create( modelMODEL_NAME_HERE, max_tokens1024, messages[{role: user, content: f请为下面内容生成摘要{text}}], ) results.append(message.content[0].text) print(results)这个写法单看没有任何问题但放到生产环境就会暴露出三个短板延迟叠加每条请求都要经过完整网络往返如果单次耗时 3 秒1000 条任务就是 3000 秒再算上重试整体时间不可控。并发难控制加线程池可以提速但并发太高容易触发限流需要自己实现退避重试逻辑代码复杂度明显上升。成本不划算每条请求都是相同计费没有批量优惠任务量上来之后 API 开销会占很大比例。同步调用适合交互式场景比如聊天机器人、实时翻译插件用户必须在几秒内得到结果。但是对离线数据处理来说同步方案不是最优解。1.2 批处理到底解决什么问题Claude 的 Batch Processing 是一个异步批量接口。你可以把大量独立请求打包成一个批处理任务提交给服务端服务端在后台统一调度处理处理完成后你再一次性拉取结果。这个模式带来的好处非常直接吞吐量高不需要自己控制并发把请求交给服务端调度再多的请求也只需要提交一次。成本优势批处理任务通常有更优惠的计费策略具体折扣以官方实时政策为准但对文本量大的团队来说选择批处理往往能显著摊薄单条成本。断点友好批处理是异步任务有明确状态中间失败了可以查询、重试不需要从头再来。逻辑清晰提交、轮询、下载三个阶段相对独立代码结构比线程池方案简单很多。1.3 典型适用场景哪些场景适合用批处理从工程经验来看满足“离线、大量、非实时”三个特征的任务都可以纳入考虑舆情评论分类一次性给几万条评论打情感标签。商品文案生成批量生成商品标题和卖点描述。文档摘要抽取对历史工单、合同、论文做重点提炼。命名实体识别从大量文本中抽取出人名、地名、机构名。数据集清洗与增强为机器学习构建训练集做改写、翻译、纠错。模型评估用一批评测问题跑多个模型然后对比输出。这些任务都不要求秒级响应用户只需要知道“任务提交成功稍后回来取结果”这就和批处理的设计目标完全一致。1.4 不适合批处理的场景批处理并不万能。如果业务要求实时应答比如客服机器人、IDE 代码补全、聊天对话那必须使用同步接口。另外批处理也不适合任务之间有依赖关系的场景。比如后一个请求需要前一个请求的输出作为输入这类串联任务需要靠编排引擎来驱动批处理只适合互相独立的并行请求。2. 环境准备与版本说明2.1 前置条件在开始之前你需要准备好以下环境一个可用的 Claude API Key并且账号已开通对应模型访问权限。Python 3.9 及以上版本推荐使用 3.10 或 3.11。能正常访问 Anthropic API 的网络环境。安装了anthropicPython SDK。这里要说明一点不同时期 SDK 版本差异较大批处理相关接口在不同版本中的命名可能不同甚至有部分接口处于 beta 阶段。下面所有代码都以“常见 SDK 写法”演示你实际运行时如果遇到AttributeError或接口不存在第一优先是检查 SDK 版本然后查阅官方 Python SDK 文档。2.2 安装 SDK创建虚拟环境并安装依赖python -m venv .venv source .venv/bin/activate # Windows 下使用 .venv\Scripts\activate pip install --upgrade anthropic验证安装是否成功python -c import anthropic; print(anthropic.__version__)如果能够正常输出版本号说明安装成功。2.3 项目结构规划为了让教程更贴近真实工程我们这里先规划好项目目录claude-batch-demo/ ├── input.jsonl # 批处理输入文件 ├── submit_batch.py # 提交批处理任务 ├── check_status.py # 查询任务状态 ├── download_results.py # 拉取并解析结果 ├── results/ # 结果存放目录 └── README.md # 说明文档实际项目里可能还需要config.py统一管理 API Key 和模型名这里我们先用最直接的方式方便你把注意力聚焦到批处理本身。2.4 环境变量配置不要把 API Key 硬编码在代码里。强烈建议使用环境变量export ANTHROPIC_API_KEYyour-api-key-herePython 代码中通过os.environ.get(ANTHROPIC_API_KEY)读取即可SDK 会自动读取这个环境变量。3. 批处理核心原理解析3.1 工作流程总览批处理不像同步接口那样“发起请求立即拿响应”它更像一个任务队列。完整流程分为三步提交输入把多条请求组合成一个 JSONL 文件调用批处理接口提交。轮询状态服务端后台开始调度处理我们定时查询任务状态。拉取结果任务完成后根据结果文件地址批量下载再按业务逻辑解析。一句话概括提交时打包运行时异步完成时批量拉取。3.2 JSONL 输入格式批处理任务的核心输入是 JSONL 文件后缀通常为.jsonl。JSONL 严格每行一个 JSON 对象行与行之间不能有空行也不能出现分隔符。每一行里包含两个关键字段custom_id你自己定义的请求标识。它在整个批处理任务里必须唯一后续下载结果时就是靠这个 ID 来对应输入输出。建议使用字母、数字、连字符、下划线避免中文或空格。params正常调用 messages.create 时的参数对象包含model、max_tokens、messages等也可以带上temperature、top_p等采样参数。一个合法的输入文件示例{custom_id: task-001, params: {model: MODEL_NAME_HERE, max_tokens: 512, messages: [{role: user, content: 请为下面的内容生成摘要人工智能技术正在改变制造业的生产组织方式。}]}} {custom_id: task-002, params: {model: MODEL_NAME_HERE, max_tokens: 512, messages: [{role: user, content: 请为下面的内容生成摘要企业数字化转型需要数据治理作为基础支撑。}]}}注意model字段需要替换为你账号实际可用的模型名称不同账号、不同时期可用的模型代号不一样这里不要照抄。3.3 任务状态生命周期批处理任务并不是提交后立即完成。它通常会经历以下几个阶段创建中刚提交系统还在校验输入文件。处理中服务端已接收任务正在排队和分批执行。已完成所有请求处理完毕可以拉取结果。失败输入文件格式严重错误或任务级异常。已取消主动撤销任务或超过处理时效未完成。整个处理窗口通常是数小时到 24 小时不等不要拿同步接口的时延预期来套批处理。这也是为什么批处理不适合实时场景。3.4 为什么批处理能降低成本批量任务在服务端可以复用批级调度资源所以单条请求的单位计费通常比同步调用更低。这是平台设计上的常见取舍用时间和灵活性换取成本。如果你的任务完全不急批处理是更经济的选择。3.5 批处理与同步调用的取舍这里做一个简单对比维度同步调用批处理响应延迟秒级分钟到小时级适用于实时交互是否请求量级适合少量适合大批量成本通常较高通常有优惠并发控制自己实现服务端调度代码复杂度简单中等含轮询实际项目中两者不是二选一往往共存。实时入口用同步离线跑批用批处理分工很明确。4. 完整实战案例批量文本摘要这一节我们完成一个可运行的端到端项目。假设你有 100 条文本需要做摘要用批处理批量完成。4.1 第一步构造输入 JSONL 文件先写一个生成脚本。为了演示方便这里假设数据源是 CSV 文件生产环境通常也是从数据库或消息队列中抽取数据。# 文件路径prepare_input.py import csv import json def build_jsonl(csv_path, jsonl_path, model, max_tokens512): with open(csv_path, r, encodingutf-8) as f: reader csv.DictReader(f) with open(jsonl_path, w, encodingutf-8) as out: for idx, row in enumerate(reader, start1): content row[content] request_obj { custom_id: fsummary-{idx:04d}, params: { model: model, max_tokens: max_tokens, messages: [ {role: user, content: f请为下面的内容生成摘要要求简洁准确{content}} ], }, } out.write(json.dumps(request_obj, ensure_asciiFalse) \n) print(f已生成 {jsonl_path}) if __name__ __main__: build_jsonl(input.csv, input.jsonl, modelMODEL_NAME_HERE)这段脚本做的事是遍历 CSV 每一行将文本字段拆成一条批处理请求写入 JSONL 文件。ensure_asciiFalse很重要它会保证中文内容以可读方式写入文件避免被转成\uXXXX形式。如果你没有 CSV 文件也可以在命令行直接准备一个很简单的input.jsonl手动编写几行即可不必拘泥于脚本。4.2 第二步提交批处理任务拿到输入文件之后调用批处理接口提交。核心代码# 文件路径submit_batch.py import os from anthropic import Anthropic client Anthropic() def create_batch(jsonl_path): with open(jsonl_path, rb) as f: batch client.beta.messages.batches.create( requests[ {custom_id: batch-1, params: {}} ], filef, ) return batch if __name__ __main__: batch create_batch(input.jsonl) print(批处理任务 ID:, batch.id) print(任务状态:, batch.processing_status)这里需要说明新版 SDK 对文件上传型批处理的支持方式可能变化有些版本直接支持传入文件对象有些版本需要先构造 requests 数组。上面示例是“通过文件对象上传”的思路如果报错请检查当前 SDK 对batches.create的签名。提交成功后脚本会打印批处理任务 ID例如batch_01XXXXXXXXXX记得保存它。后续查询状态、下载结果都要用到这个 ID。项目里建议把 task ID 写入本地 txt 文件方便下一个脚本复用with open(batch_id.txt, w) as f: f.write(batch.id)4.3 第三步轮询任务状态任务提交后不会立刻完成我们需要定时查询。实现一个简单的轮询脚本# 文件路径check_status.py import time import os from anthropic import Anthropic client Anthropic() def check_batch(batch_id): batch client.beta.messages.batches.retrieve(batch_id) print(状态:, batch.processing_status) print(请求总数:, batch.request_counts.total) print(已完成:, batch.request_counts.succeeded) print(失败数:, batch.request_counts.failed) return batch.processing_status if __name__ __main__: batch_id open(batch_id.txt).read().strip() while True: status check_batch(batch_id) if status in [ended, canceled]: break time.sleep(30)轮询间隔不建议太短。批处理任务通常运行几分钟到几小时设置 30 到 60 秒一次已经足够。间隔太短只会浪费请求配额并不会加快任务执行。4.4 第四步下载并解析结果任务状态变为完成后从结果文件拉取内容并解析# 文件路径download_results.py import json from anthropic import Anthropic client Anthropic() def download_results(batch_id, output_dirresults): import os os.makedirs(output_dir, exist_okTrue) result_file os.path.join(output_dir, result.jsonl) with open(result_file, w, encodingutf-8) as out: for result in client.beta.messages.batches.results(batch_id): out.write(json.dumps(result, ensure_asciiFalse) \n) print(结果已保存到, result_file) if __name__ __main__: batch_id open(batch_id.txt).read().strip() download_results(batch_id)下载完成后结果文件同样是 JSONL 格式每一行对应一个请求。每个结果对象里包含custom_id、result等字段result中又包含type、message、content等子字段。实际业务中我们需要从结果里提取最终文本# 文件路径parse_results.py import json def parse_result(line): obj json.loads(line) custom_id obj.get(custom_id) result obj.get(result, {}) if result.get(type) succeeded: content result.get(message, {}).get(content, []) text .join(block.get(text, ) for block in content if block.get(type) text) return custom_id, text else: return custom_id, fERROR: {result} if __name__ __main__: with open(results/result.jsonl, r, encodingutf-8) as f: for line in f: line line.strip() if not line: continue cid, text parse_result(line) print(cid, text[:100])4.5 预期输出说明整条流程跑完result.jsonl中每一行会输出一个结果对象。解析后你会发现每个custom_id都能和输入文件中的task-001或summary-0001一一对应。这就是设计custom_id的意义无论结果返回顺序如何你都能安全地把输出映射到业务数据。这里额外强调一个小细节结果顺序可能与输入顺序不一致。批处理是并行执行所以写代码时不要依赖行号对应必须通过custom_id关联。5. 常见问题与排查思路批处理在落地过程中会碰到不少报错这里整理高频问题清单。问题现象常见原因解决思路提交时报invalid jsonl格式错误JSONL 有空行、非 JSON 内容、字段缺失用 Pythonjson.loads逐行校验修正后再提交提示model 不存在或无权访问模型名填错或账号不可用确认当前模型代号换成账号可用的模型任务一直处于processing队列繁忙或输入量非常大耐心等待如果超 24 小时未完成联系官方支持渠道结果里出现大量custom_id解码错误custom_id包含非法字符只使用字母、数字、连字符、下划线并保证唯一下载结果时超时结果文件过大增加超时时间分段下载或用流式读取调用batches.create报AttributeErrorSDK 版本过旧或接口路径变更升级到最新 SDK查看当前版本官方文档拉取结果时出现 529 状态码服务端过载退避重试等几秒后再次请求5.1 输入文件校验小工具为了减少格式问题建议提交前写一个快速校验脚本# 文件路径validate_jsonl.py import json import sys def validate(path): with open(path, r, encodingutf-8) as f: for line_num, line in enumerate(f, start1): line line.strip() if not line: print(f第 {line_num} 行为空行请移除) continue try: obj json.loads(line) except json.JSONDecodeError as e: print(f第 {line_num} 行解析失败: {e}) continue if custom_id not in obj or params not in obj: print(f第 {line_num} 行缺少 custom_id 或 params) print(校验完成) if __name__ __main__: validate(sys.argv[1] if len(sys.argv) 1 else input.jsonl)这个脚本虽然简单但在大批量提交前跑一遍能省去很多麻烦。5.2 排查优先级遇到问题时推荐按这个顺序排查检查输入 JSONL 是否合法。这是最容易被忽略的坑。检查模型名是否拼写正确并发一条同步请求确认模型可用。检查 API Key 是否有批处理权限或配额。检查 SDK 版本是否足够新。检查网络是否能稳定访问 Anthropic API。6. 最佳实践与工程建议6.1 输入数据与 custom_id 设计custom_id不只是字符串它是业务数据的关联键。实际项目中建议使用业务主键或复合键例如user_id:order_id的拼接形式但要先确认替换为符合字符要求的格式。更好的做法是维护一张映射表{ summary-0001: {user_id: 1001, order_id: A10001}, summary-0002: {user_id: 1002, order_id: A10002} }下载结果后通过custom_id反查映射表就能把模型输出写回数据库。6.2 任务拆分与文件大小控制不要试图一次提交几十万条请求。建议按业务维度把任务切分成多个批处理文件每个文件控制在合理规模每个批处理文件内请求数量适中避免文件过大导致上传和下载超时。每组任务单独记录batch_id方便单独重跑。按时间分区例如按天、按小时生成独立的批处理任务便于排查问题。6.3 状态轮询策略轮询不要太频繁建议使用指数退避前 5 分钟每 30 秒查一次。5 分钟后每 5 分钟查一次。超过 1 小时后每 15 分钟查一次。这样可以避免无效请求占用 API 配额。生产环境更推荐用任务表记录状态由调度系统统一驱动。6.4 结果落库与失败重跑下载结果后不要直接覆盖原文件建议按批次号和任务 ID 命名例如results/batch_20250211_1200_01.jsonl解析结果时将succeeded和failed分开记录。失败请求可以汇总到retry.jsonl稍后重新提交。6.5 安全与合规处理涉密或敏感数据时要特别谨慎不要把真实手机号、身份证号直接放进 JSONL建议先做脱敏。设置严格的 API Key 权限最小化访问范围。生产环境对工具链做访问控制禁止非授权人员提交、取消、下载批处理任务。对结果数据设置合理的保存周期过期清理。6.6 成本控制批处理虽然单位成本更低但任务量巨大时总开销仍然不小。建议按天统计请求数量和 token 消耗。对超出预期的任务量做风控告警。明确模型选择策略简单任务用小模型复杂任务才上大模型。在params中合理控制max_tokens避免输出过长造成浪费。6.7 日志与监控线上环境要保留轨迹建议至少记录任务提交时间、提交人、输入文件路径。批量任务 ID、请求总数。完成时间、成功数、失败数。失败原因归类。这些信息是后续排查问题的基础。7. 总结与学习路线通过这篇文章你已经掌握了 Claude 批处理的完整链路从 JSONL 输入文件构造、批处理任务提交、状态轮询、结果下载到按custom_id解析输出。你也知道了批处理适合什么场景、不适合什么场景以及如何优化成本和排查报错。下一步建议你做三件事第一把示例代码跑通用自己的数据集替换文本摘要需求改成分类、翻译、实体抽取加深对流程的熟悉。第二学习如何把批处理和消息队列或任务调度框架整合实现离线数据平台级的自动触发、重试、告警。第三研究一下结果文件中的 token 使用统计结合你的账单理解批处理任务的实际成本和优惠幅度。如果你在跑批处理时遇到奇奇怪怪的报错欢迎把错误信息和任务状态发在评论区。读代码是学习亲手跑通一条批处理流水线才是真正的掌握。