agentic-awesome-skills 中的 Claude Message Batches API（Python）：异步批量消息处理实战指南

AI 技能AI 插件【免费下载链接】agentic-awesome-skillsAAS Core is the local, agent-first control plane for complete catalog discovery, agent-owned selection, stack validation, and planning, backed by 2,445 agentic skills. Includes CLI, local MCP, catalog, plugins, and Workbench.项目地址https://gitcode.com/gh_mirrors/an/agentic-awesome-skills点击查看免费下载导读本文以plugins/agentic-awesome-skills-claude插件中claude-apiskill 的 batches.md 为核心骨架系统讲解如何用 Python 官方 SDK 调用 Claude Message Batches APIPOST /v1/messages/batches——以标准价格50% 的成本异步批量处理海量 Messages API 请求。读完本文你将掌握批次创建、状态轮询、结果分类读取、批次取消以及与 Prompt Caching 组合的降本方案并能直接复用文末的端到端代码跑通一个真实批处理任务。一、Batches API 是什么异步批处理的价值与边界Batches API 是 Messages API 的异步形态你把成百上千条独立的请求一次性提交由服务端排队处理处理完成后统一拉取结果。它不改变请求语义而是改变交付方式——从同步等待单条响应变为异步提交、事后批量收割。本仓库claude-apiskill 的 SKILL.md 明确给出了它的适用场景Batch processing (non-latency-sensitive)即非延迟敏感的离线批处理。典型的场景包括大规模文本分类、情感标注如文末示例的商品评论分类批量摘要、批量翻译、批量实体抽取离线数据清洗与结构化抽取需要跨大量文档复用同一上下文的分析任务。Key Facts关键事实全部来自关联文档约束/能力数值单批次最大请求数100,000 个请求单批次最大体积256 MB完成时间大多数批次 1 小时内完成最长 24 小时结果保留期创建后 29 天内可获取成本所有 token 用量均按标准价50%计费能力范围支持全部 Messages API 特性vision 视觉、tools 工具调用、prompt caching 缓存等两点需要特别提醒第一批处理天然有延迟——若你的业务需要秒级响应请走同步的client.messages.create()第二50% 折扣是针对批处理通道的整体计费策略不因单条请求大小而变化因此请求越大、批量越大节省越明显。二、环境准备与客户端初始化在编写批处理代码前先按 Python claude-api README 完成环境准备pip install anthropic客户端初始化有三种方式import anthropic # 方式一默认读取环境变量 ANTHROPIC_API_KEY client anthropic.Anthropic() # 方式二显式传入 API key client anthropic.Anthropic(api_keyyour-api-key) # 方式三异步客户端配合 async/await 使用 async_client anthropic.AsyncAnthropic()关联文档中的示例全部使用方式一即通过ANTHROPIC_API_KEY环境变量注入密钥。不要把 API key 硬编码进代码——error-codes.md 将API key in code列为 401 错误的典型诱因密钥泄露。模型 ID 的选择批次中的每条请求都要声明model。仓库的 shared/models.md 强调只能使用表中列出的精确模型 ID绝不猜测或拼接。当前推荐模型如下仓库缓存日期 2026-02-17来源 SKILL.md模型模型 ID使用此值上下文窗口输入 $/1M tokens输出 $/1M tokensClaude Opus 4.6claude-opus-4-6200K1M beta$5.00$25.00Claude Sonnet 4.6claude-sonnet-4-6200K1M beta$3.00$15.00Claude Haiku 4.5claude-haiku-4-5200K$1.00$5.00价格数据为仓库缓存值仅作成本估算参考实际价格请以官方实时数据为准仓库 live-sources.md 提供了实时定价文档的 WebFetch 地址。注意 50% 折扣同样适用于上述单价——以 Haiku 4.5 跑批量分类为例输入成本从 $1.00/1M 降至 $0.50/1M。三、创建批次核心 API 与请求结构Batches API 的 Python SDK 入口是client.messages.batches.create()。关联文档给出的最小可运行示例import anthropic from anthropic.types.message_create_params import MessageCreateParamsNonStreaming from anthropic.types.messages.batch_create_params import Request client anthropic.Anthropic() message_batch client.messages.batches.create( requests[ Request( custom_idrequest-1, paramsMessageCreateParamsNonStreaming( modelclaude-opus-4-6, max_tokens1024, messages[{role: user, content: Summarize climate change impacts}] ) ), Request( custom_idrequest-2, paramsMessageCreateParamsNonStreaming( modelclaude-opus-4-6, max_tokens1024, messages[{role: user, content: Explain quantum computing basics}] ) ), ] ) print(fBatch ID: {message_batch.id}) print(fStatus: {message_batch.processing_status})请求结构拆解Request与custom_id每个Request由两部分组成custom_id必填客户端自定义的唯一标识符用于在结果中关联哪条请求对应哪个结果。建议采用可读、可排序的命名如request-1、classify-0因为结果返回时并不保证顺序custom_id是你还原业务数据的唯一锚点。params一个完整的MessageCreateParamsNonStreaming——与同步messages.create()的参数完全一致支持model、max_tokens、messages、system、tools、cache_control等全部 Messages API 参数。三条重要规则custom_id在同一批次内必须唯一否则结果无法区分模型 ID 必须是精确值如claude-opus-4-6拼错会以 404/invalid_request错误落回该条请求的结果中详见 error-codes.md创建成功的响应包含batch.id与processing_status此时通常为in_progress后续轮询、取结果、取消都要用到batch.id。从仓库 SKILL.md 的默认约定看除非用户另有指定模型默认使用claude-opus-4-6而文末端到端示例用claude-haiku-4-5跑低成本分类体现了按任务选模型的工程取舍。四、轮询批次完成状态processing_status 与 request_counts批次是异步的创建后需要轮询直到终态。关联文档的标准轮询模式import time while True: batch client.messages.batches.retrieve(message_batch.id) if batch.processing_status ended: break print(fStatus: {batch.processing_status}, processing: {batch.request_counts.processing}) time.sleep(60) print(Batch complete!) print(fSucceeded: {batch.request_counts.succeeded}) print(fErrored: {batch.request_counts.errored})字段语义processing_status批次状态机。常见取值包括in_progress处理中、ended结束可获取结果、canceling取消中详见第六节。当且仅当状态为ended时才应去拉取结果。request_counts一个计数对象包含processing仍在处理、succeeded成功、errored出错等字段用于进度感知。配合print日志可以在长耗时批次中持续观察进度。轮询间隔建议关联文档使用time.sleep(60)60 秒间隔。这是合理的默认值——大多数批次 1 小时内完成秒级轮询只会白白消耗 API 配额。对于小型批次如几十条请求可以按端到端示例那样缩短到time.sleep(10)加快反馈对于上万条请求的大批次建议保持 60 秒或更长的间隔。设计考量为什么有 24 小时上限批处理本质是排队 分片执行服务端在资源空闲时优先处理因此官方给出多数 1 小时内完成、最长 24 小时的保证。这意味着依赖批次结果的下游任务要容忍最长 24 小时的延迟边界若批次在 24 小时内未能完成部分请求可能进入expired状态需要在结果读取阶段单独处理见下节。五、读取结果按 custom_id 收割并分类处理批次结束后用client.messages.batches.results(batch.id)逐条取出结果。关联文档使用了 Python 3.10 的match/case结构模式匹配文档明确提示Python 3.10 以下请改用if/elif链for result in client.messages.batches.results(message_batch.id): match result.result.type: case succeeded: print(f[{result.custom_id}] {result.result.message.content[0].text[:100]}) case errored: if result.result.error.type invalid_request: print(f[{result.custom_id}] Validation error - fix request and retry) else: print(f[{result.custom_id}] Server error - safe to retry) case canceled: print(f[{result.custom_id}] Canceled) case expired: print(f[{result.custom_id}] Expired - resubmit)四种结果类型的处理策略结果类型含义推荐处理succeeded请求成功result.result.message为完整 Message 对象提取content文本按custom_id归位errored请求失败看result.result.error.typeinvalid_request是请求本身有问题如参数非法需修复后重建请求其他类型多为服务端错误可安全重试canceled批次被取消部分请求未执行记录即可按业务决定是否重建expired批次超时24 小时边界或结果过期重新提交resubmit提取文本的细节result.result.message.content是 content block 列表。同步请求中首个 block 通常是文本因此示例用content[0].text取文本。但从源码使用惯例看参见 README 对 thinking block 的处理更稳健的写法是遍历并筛选type text的 block若请求启用了 thinkingcontent[0]可能是 thinking block 而非文本。六、取消批次cancel 的语义与边界cancelled client.messages.batches.cancel(message_batch.id) print(fStatus: {cancelled.processing_status}) # canceling取消调用后状态会进入canceling已处理完成的请求结果仍可取回未处理的请求会以canceled类型落在结果流中。取消不是瞬间完成需要配合轮询确认最终状态。注意取消通常只对尚未执行或仍在排队的请求生效如果批次已进入快速执行阶段取消可能需要时间生效因此应在业务上把取消视为异步操作。七、批处理 × Prompt Caching让大批量请求共享同一上下文批量任务最典型的成本杀手是每条请求都重复发送同一份大文档。解决方案是把共享内容放进system块并标记cache_control让所有请求复用同一份缓存上下文。关联文档给出了完整模式shared_system [ {type: text, text: You are a literary analyst.}, { type: text, text: large_document_text, # Shared across all requests cache_control: {type: ephemeral} } ] message_batch client.messages.batches.create( requests[ Request( custom_idfanalysis-{i}, paramsMessageCreateParamsNonStreaming( modelclaude-opus-4-6, max_tokens1024, systemshared_system, messages[{role: user, content: question}] ) ) for i, question in enumerate(questions) ] )机制与收益system数组中的cache_control: {type: ephemeral}标记该内容块为可缓存默认 TTL 5 分钟可显式指定ttl: 1h等见 README 的 Prompt Caching 节批处理内大量请求共享同一份系统上下文时首次请求全价写入缓存后续请求命中缓存缓存部分成本可降约 90%在此基础上再叠加批处理自身的 50% 折扣效果叠加——这是仓库文档中成本优化的核心组合拳。阅读建议本仓库 SKILL.md 的阅读指引将batches.md与README.md捆绑使用原因正在于此——批处理几乎总是与 prompt caching、错误处理、模型选择配合使用而不是孤立调用。八、完整端到端示例评论情感分类关联文档最后给出的完整示例从准备请求到收割结果一气呵成建议作为你的脚手架代码import anthropic import time from anthropic.types.message_create_params import MessageCreateParamsNonStreaming from anthropic.types.messages.batch_create_params import Request client anthropic.Anthropic() # 1. Prepare requests items_to_classify [ The product quality is excellent!, Terrible customer service, never again., Its okay, nothing special., ] requests [ Request( custom_idfclassify-{i}, paramsMessageCreateParamsNonStreaming( modelclaude-haiku-4-5, max_tokens50, messages[{ role: user, content: fClassify as positive/negative/neutral (one word): {text} }] ) ) for i, text in enumerate(items_to_classify) ] # 2. Create batch batch client.messages.batches.create(requestsrequests) print(fCreated batch: {batch.id}) # 3. Wait for completion while True: batch client.messages.batches.retrieve(batch.id) if batch.processing_status ended: break time.sleep(10) # 4. Collect results results {} for result in client.messages.batches.results(batch.id): if result.result.type succeeded: results[result.custom_id] result.result.message.content[0].text for custom_id, classification in sorted(results.items()): print(f{custom_id}: {classification})这段代码展示了批处理的四个标准阶段准备请求用列表推导批量构造Requestcustom_id与业务数据一一对应classify-0→ 第 0 条评论创建批次一次create提交全部请求等待完成小批次用 10 秒轮询状态为ended时退出收割结果遍历结果流按custom_id存入字典最后排序输出——输出顺序与提交顺序一致便于人工核对。选用claude-haiku-4-5的原因可从 SKILL.md 模型表 读出Haiku 4.5 是最快、最具成本效益的模型单字分类这种简单任务用它 50% 批处理折扣成本最优。九、工程化加固错误处理与重试策略批处理虽为异步但创建、轮询、取结果这三类调用本身仍是同步 HTTP 请求可能抛异常。结合 README 的错误处理节 与 error-codes.md 的异常映射表推荐用 SDK 的类型化异常处理import anthropic try: batch client.messages.batches.create(requestsrequests) except anthropic.BadRequestError as e: print(fBad request: {e.message}) # 400请求结构非法 except anthropic.AuthenticationError: print(Invalid API key) # 401 except anthropic.PermissionDeniedError: print(API key lacks required permissions) # 403 except anthropic.RateLimitError as e: retry_after int(e.response.headers.get(retry-after, 60)) print(fRate limited. Retry after {retry_after}s.) # 429 except anthropic.APIStatusError as e: if e.status_code 500: print(fServer error ({e.status_code}). Retry later.) # 5xx else: print(fAPI error: {e.message}) except anthropic.APIConnectionError: print(Network error. Check internet connection.)错误码与异常类映射来自 error-codes.mdHTTP 状态码错误类型是否可重试常见原因400invalid_request_error否请求格式/参数非法401authentication_error否API key 无效或缺失403permission_error否key 无权限404not_found_error否端点或模型 ID 错误413request_too_large否请求超限单条请求过大429rate_limit_error是请求/Token 超限500api_error是Anthropic 服务问题529overloaded_error是API 过载SDK 自带重试无需重复造轮子README 的 Retry 节 明确指出Anthropic SDK 已对 429 与 5xx 自动指数退避重试默认max_retries2。仅当需要自定义重试行为如更多次数、更长退避时才自行实现且实现时只重试 429/5xx4xx 客户端错误直接抛出。批处理特有的错误场景批次内的请求错误不抛异常它们以errored结果类型出现在结果流中需要按第五节的方法分类处理——这是与同步 API 最大的心智差异413单批次上限 256 MB超限会拒绝创建请在提交前控制请求总大小截断历史、压缩图片或分片提交。十、跨语言一致性同一语义多语言 SDK本仓库的claude-apiskill 同时提供 Python 与 TypeScript 两个完整版本TypeScript 版 batches.md 与本文档共享完全相同的 Key Facts 与流程骨架。对比可见流程PythonTypeScript创建client.messages.batches.create()client.messages.batches.create()轮询client.messages.batches.retrieve(id)client.messages.batches.retrieve(id)取结果client.messages.batches.results(id)迭代器for await ... batches.results(id)异步迭代器取消client.messages.batches.cancel(id)client.messages.batches.cancel(id)结果分类match/case3.10switch/caseSDK 方法命名完全对齐错误分类逻辑invalid_request区分可修复/可重试也保持一致。这意味着如果你在 TypeScript/Node.js 技术栈中需要同样的批处理能力直接对照 TypeScript 版本文档 即可概念与本节各流程一一对应。十一、常见问题速查现象原因处理批次状态迟迟不结束大批次排队执行耐心等待24 小时为上限期间用request_counts.processing观察进度部分结果报invalid_request该请求参数非法模型 ID 错误、messages 结构错误等修复该请求后重新提交参考 error-codes.md 的 400 排查清单结果流中出现expired批次超时或结果过期创建后 29 天重新提交长期任务注意在 29 天内取走结果想节省大文档重复传输成本未使用缓存用第七节的cache_control方案共享系统上下文match/case语法报错Python 3.10改用if/elif链逐类型判断创建批次报 401/403API key 问题检查ANTHROPIC_API_KEY环境变量与 key 权限小结Batches API 是 Claude 生态中降本 吞吐的关键通道单批次最高 10 万请求、256 MB全部 token 半价计费且完整保留 vision、tool use、prompt caching 等 Messages API 能力。结合本仓库claude-apiskill 的文档体系你可以在 batches.md 与 README.md 之间按需跳转——前者覆盖批处理全流程代码后者补齐客户端初始化、错误处理、成本优化与多轮对话等配套能力需要最新模型与定价时参考 shared/live-sources.md 中记录的官方实时文档地址。把本文的端到端示例作为起点将items_to_classify替换为你的真实数据、custom_id替换为你的业务主键即可在生产环境中落地一套半价的批量推理流水线。赞分享AI 技能AI 插件【免费下载链接】agentic-awesome-skillsAAS Core is the local, agent-first control plane for complete catalog discovery, agent-owned selection, stack validation, and planning, backed by 2,445 agentic skills. Includes CLI, local MCP, catalog, plugins, and Workbench.项目地址https://gitcode.com/gh_mirrors/an/agentic-awesome-skills点击查看免费下载相关推荐Claude 批量处理如何用 Message Batches API 异步跑大规模请求并降低一半成本Claude 批量处理如何用 Message Batches API 异步跑大规模请求并降低一半成本 当你手上有大量不需要实时响应的 Claude Messa示例工程Claude API Python 开发实战基于 agentic-awesome-skills 仓库构建 Messages API 应用Claude API Python 开发实战基于 agentic awesome skills 仓库构建 Messages API 应用 本篇技术指南以 clAI 技能AI 插件marimo 循环依赖 Lint 规则 MB003原理、报错解读与修复实战marimo 循环依赖 Lint 规则 MB003原理、报错解读与修复实战 导读 本篇文章聚焦 marimo 内置的静态检查规则 MB003: cycle dAI 技能AI 插件上一篇Matting Anything实用技巧如何用语言提示词精准控制alpha matte生成下一篇AVA 并发控制与 --concurrency 参数校验从快照测试到源码实现创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考