企业官网建设流程全解析

• AI 技能
• AI 插件

导读

本文以plugins/agentic-awesome-skills-claude插件中claude-apiskill 的 batches.md 为核心骨架，系统讲解如何用 Python 官方 SDK 调用 Claude Message Batches API（POST /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（关键事实，全部来自关联文档）

两点需要特别提醒：第一，批处理天然有延迟——若你的业务需要秒级响应，请走同步的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_key="your-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）：

价格数据为仓库缓存值，仅作成本估算参考；实际价格请以官方实时数据为准（仓库 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_id="request-1", params=MessageCreateParamsNonStreaming( model="claude-opus-4-6", max_tokens=1024, messages=[{"role": "user", "content": "Summarize climate change impacts"}] ) ), Request( custom_id="request-2", params=MessageCreateParamsNonStreaming( model="claude-opus-4-6", max_tokens=1024, messages=[{"role": "user", "content": "Explain quantum computing basics"}] ) ), ] ) print(f"Batch ID: {message_batch.id}") print(f"Status: {message_batch.processing_status}")

请求结构拆解：Request与custom_id

每个Request由两部分组成：

1. custom_id（必填）：客户端自定义的唯一标识符，用于在结果中关联"哪条请求对应哪个结果"。建议采用可读、可排序的命名（如request-1、classify-0），因为结果返回时并不保证顺序，custom_id是你还原业务数据的唯一锚点。
2. 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(f"Status: {batch.processing_status}, processing: {batch.request_counts.processing}") time.sleep(60) print("Batch complete!") print(f"Succeeded: {batch.request_counts.succeeded}") print(f"Errored: {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")

四种结果类型的处理策略

提取文本的细节

result.result.message.content是 content block 列表。同步请求中首个 block 通常是文本，因此示例用content[0].text取文本。但从源码使用惯例看（参见 README 对 thinking block 的处理），更稳健的写法是遍历并筛选type == "text"的 block；若请求启用了 thinking，content[0]可能是 thinking block 而非文本。

六、取消批次：cancel 的语义与边界

cancelled = client.messages.batches.cancel(message_batch.id) print(f"Status: {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_id=f"analysis-{i}", params=MessageCreateParamsNonStreaming( model="claude-opus-4-6", max_tokens=1024, system=shared_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.", "It's okay, nothing special.", ] requests = [ Request( custom_id=f"classify-{i}", params=MessageCreateParamsNonStreaming( model="claude-haiku-4-5", max_tokens=50, messages=[{ "role": "user", "content": f"Classify as positive/negative/neutral (one word): {text}" }] ) ) for i, text in enumerate(items_to_classify) ] # 2. Create batch batch = client.messages.batches.create(requests=requests) print(f"Created 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}")

这段代码展示了批处理的四个标准阶段：

1. 准备请求：用列表推导批量构造Request，custom_id与业务数据一一对应（classify-0→ 第 0 条评论）；
2. 创建批次：一次create提交全部请求；
3. 等待完成：小批次用 10 秒轮询，状态为ended时退出；
4. 收割结果：遍历结果流，按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(requests=requests) except anthropic.BadRequestError as e: print(f"Bad 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(f"Rate limited. Retry after {retry_after}s.") # 429 except anthropic.APIStatusError as e: if e.status_code >= 500: print(f"Server error ({e.status_code}). Retry later.") # 5xx else: print(f"API error: {e.message}") except anthropic.APIConnectionError: print("Network error. Check internet connection.")

错误码与异常类映射（来自 error-codes.md）：

SDK 自带重试，无需重复造轮子

README 的 Retry 节 明确指出：Anthropic SDK 已对 429 与 5xx 自动指数退避重试（默认max_retries=2）。仅当需要自定义重试行为（如更多次数、更长退避）时才自行实现，且实现时只重试 429/5xx，4xx 客户端错误直接抛出。

批处理特有的错误场景

• 批次内的请求错误不抛异常：它们以errored结果类型出现在结果流中，需要按第五节的方法分类处理——这是与同步 API 最大的心智差异；
• 413：单批次上限 256 MB，超限会拒绝创建；请在提交前控制请求总大小（截断历史、压缩图片或分片提交）。

十、跨语言一致性：同一语义，多语言 SDK

本仓库的claude-apiskill 同时提供 Python 与 TypeScript 两个完整版本，TypeScript 版 batches.md 与本文档共享完全相同的 Key Facts 与流程骨架。对比可见：

SDK 方法命名完全对齐，错误分类逻辑（invalid_request区分可修复/可重试）也保持一致。这意味着：如果你在 TypeScript/Node.js 技术栈中需要同样的批处理能力，直接对照 TypeScript 版本文档 即可，概念与本节各流程一一对应。

十一、常见问题速查

小结

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 插件