企业官网建设流程全解析

• 人工智能
• AI 技能
• AI 评测

导读

Message Batches API（POST /v1/messages/batches）允许开发者把大量 Messages API 请求打包成批、异步处理，并享受标准价格 50% 的折扣。本文以 skills/claude-api/typescript/claude-api/batches.md 为核心骨架，结合仓库内 Python / cURL / C# / PHP 的对应实现与成本优化文档，系统讲解在@anthropic-ai/sdk（TypeScript）中如何创建批次、轮询状态、读取结果、取消批次，以及如何用提示词缓存进一步压低批处理成本。读完本文，你将能编写一套完整可运行的批量文本分类、批量摘要等离线任务流水线。

一、什么是 Message Batches API：异步处理与半价计费

Batches API 的核心设计是把同步的messages.create请求转为异步队列任务：客户端一次性提交一批请求（每个请求就是一个完整的 Messages 参数集），服务端异步排队执行，全部完成后统一提供结果下载。这带来两个直接收益：

• 成本减半：批次内所有 token 用量（输入 + 输出）按标准价格 50% 计费；
• 吞吐解耦：适合"没人等着实时返回"的离线负载，如夜间批量分类、批量摘要、大规模评测集打分、历史数据清洗。

与普通 Messages API 相比，批处理不是简单的接口换名——它引入了配额、生命周期与结果类型等一整套不同的约束，下面逐一展开。

关键事实（Key Facts）

根据原文档，使用前必须先记住这五条硬性约束：

此外，所有 Messages API 特性在批处理中均可用：vision（图片输入）、工具调用（tool use）、提示词缓存（prompt caching）等，无需降级。仓库在 skills/claude-api/shared/cost-optimization.md 中将批处理定位为"免费赢项（free win）"之一：它不降低输出质量，只是把"没人等待"的那部分流量挪到半价档。

二、环境准备：安装 SDK 与初始化客户端

批处理调用与普通消息调用共用同一个Anthropic客户端，因此初始化方式完全一致。仓库的 typescript/claude-api/README.md 给出了安装与初始化标准做法：

npm install @anthropic-ai/sdk

import Anthropic from "@anthropic-ai/sdk"; // 推荐：从环境变量解析凭据（ANTHROPIC_API_KEY / ANTHROPIC_AUTH_TOKEN / `ant auth login` 配置），不要硬编码 key const client = new Anthropic(); // 仅在必须注入指定 key 时使用显式传参 // const client = new Anthropic({ apiKey: "your-api-key" });

ESM 注意：在 ES 模块的.ts文件中，__dirname/__filename是 undefined，直接使用会抛ReferenceError。需要读取本地文件（例如后续要打包进批次的图片 base64）时，cwd 相对读取请传裸相对路径（如fs.readFileSync("./sample.png")），脚本相对路径则基于import.meta.url推导。

三、创建批次（Create a Batch）

批次创建通过client.messages.batches.create()完成，入参是一个requests数组。每个请求元素包含两大部分：

• custom_id：自定义字符串标识，用于在结果中反查对应请求（类似"批次内的 request_id"）；
• params：一个完整的非流式 Messages 请求参数对象（model、max_tokens、messages、system、tools等），与messages.create的参数同构。

const messageBatch = await client.messages.batches.create({ requests: [ { custom_id: "request-1", params: { model: "claude-opus-5", max_tokens: 16000, messages: [ { role: "user", content: "Summarize climate change impacts" }, ], }, }, { custom_id: "request-2", params: { model: "claude-opus-5", max_tokens: 16000, messages: [ { role: "user", content: "Explain quantum computing basics" }, ], }, }, ], }); console.log(`Batch ID: ${messageBatch.id}`); console.log(`Status: ${messageBatch.processing_status}`);

创建成功后返回的批次对象关键字段：

• id：批次唯一 ID，后续 retrieve / results / cancel 都要用到；
• processing_status：当前处理状态（见下节状态流转）。

设计建议：custom_id应携带业务语义（如classify-0、analysis-42），而不是无意义序号，这样解析结果时无需额外映射表。仓库的 python/claude-api/batches.md 中端到端示例即采用classify-{i}这类模式。

原始 HTTP 形态（cURL 视角）

如需理解底层协议或脱离 SDK 调试，可以参照 curl/examples.md 中的请求头规范（Content-Type: application/json、x-api-key、anthropic-version: 2023-06-01）。批次端点与普通消息端点不同，SDK 的batches命名空间正是封装了POST /v1/messages/batches及其子路径，这与 csharp/claude-api/batches.md（client.Messages.Batches.Create）和 php/claude-api/batches.md（$client->messages->batches->create）的命名空间一一对应，从源码结构可以推断三套 SDK 遵循同一套 REST 资源模型。

四、轮询完成状态（Poll for Completion）

批处理是异步的，创建后需要轮询client.messages.batches.retrieve(batchId)直到终态。原文档给出的轮询模式如下：

let batch; while (true) { batch = await client.messages.batches.retrieve(messageBatch.id); if (batch.processing_status === "ended") break; console.log( `Status: ${batch.processing_status}, processing: ${batch.request_counts.processing}`, ); await new Promise((resolve) => setTimeout(resolve, 60_000)); } console.log("Batch complete!"); console.log(`Succeeded: ${batch.request_counts.succeeded}`); console.log(`Errored: ${batch.request_counts.errored}`);

关键字段说明：

• processing_status：批次状态。常见取值包括processing（处理中）、ended（已结束）以及取消路径上的canceling/ 已取消等；
• request_counts：计数统计对象，包含processing（处理中数量）、succeeded（成功数量）、errored（失败数量）等子字段，可在轮询过程中实时观察进度。

轮询间隔建议：多数批次 1 小时内完成，原文档默认每 60 秒轮询一次；对紧急度低的超大批次可放宽间隔以降低 API 调用次数。Python 版端到端示例（python/claude-api/batches.md）对小型批次用 10 秒间隔，可作对照参考。

五、读取结果（Retrieve Results）

批次进入ended后，通过client.messages.batches.results(batchId)以**异步迭代器（async iterable）**形式逐条读取结果。每条结果包含custom_id与result对象，result.type是一个判别联合（discriminated union），必须在switch中按类型分支处理：

for await (const result of await client.messages.batches.results( messageBatch.id, )) { switch (result.result.type) { case "succeeded": console.log( `[${result.custom_id}] ${result.result.message.content[0].text.slice(0, 100)}`, ); break; case "errored": if (result.result.error.type === "invalid_request") { console.log(`[${result.custom_id}] Validation error - fix and retry`); } else { console.log(`[${result.custom_id}] Server error - safe to retry`); } break; case "expired": console.log(`[${result.custom_id}] Expired - resubmit`); break; } }

结果类型与重试语义

原文档的 TypeScript 版覆盖succeeded / errored / expired三类；python/claude-api/batches.md 额外展示了canceled分支，说明底层结果类型集合中还存在"已取消"这一状态，TS 项目中如处理取消流程建议一并判断。

六、取消批次（Cancel a Batch）

当批次仍在处理中、但你决定不再需要它时（例如上游数据出错），可以调用取消接口：

const cancelled = await client.messages.batches.cancel(messageBatch.id); console.log(`Status: ${cancelled.processing_status}`); // "canceling"

取消是异步生效的：返回的processing_status为canceling（取消中），随后通过 retrieve 轮询会看到它过渡到已取消终态。已ended的批次无法再取消。

七、端到端实战：批量情感分类流水线

把以上四个步骤串起来，即构成一个完整的批处理流水线。下面是一个基于原文档模式、并参考仓库 Python 端到端示例（python/claude-api/batches.md）整合出的 TypeScript 完整示例——对多条评论文本做情感分类：

import Anthropic from "@anthropic-ai/sdk"; const client = new Anthropic(); async function classifyInBatch(items: string[]): Promise<Map<string, string>> { // 1. 构造请求：每个 item 生成一个带语义 custom_id 的请求 const requests = items.map((text, i) => ({ custom_id: `classify-${i}`, params: { model: "claude-opus-5", max_tokens: 50, messages: [ { role: "user", content: `Classify as positive/negative/neutral (one word): ${text}`, }, ], }, })); // 2. 创建批次 const batch = await client.messages.batches.create({ requests }); console.log(`Created batch: ${batch.id}`); // 3. 轮询直至结束 while (true) { const current = await client.messages.batches.retrieve(batch.id); if (current.processing_status === "ended") break; console.log( `Status: ${current.processing_status}, processing: ${current.request_counts.processing}`, ); await new Promise((resolve) => setTimeout(resolve, 10_000)); } // 4. 收集结果（按 custom_id 组织，便于排序输出） const results = new Map<string, string>(); for await (const result of await client.messages.batches.results(batch.id)) { if (result.result.type === "succeeded") { const text = (result.result.message.content[0] as Anthropic.TextBlock).text; results.set(result.custom_id, text); } } return results; } // 用法：results.get("classify-0") 即第一条文本的分类结果

此处的content[0]直接断言为TextBlock，实际生产代码建议先用block.type === "text"收窄（仓库 TypeScript 文档在"基本消息请求"一节反复强调这一点，避免 TypeScript 类型报错）。

八、进阶：批处理 × 提示词缓存组合

批处理 5 折优惠可与提示词缓存（prompt caching）叠加，是离线任务压成本最有效的组合。核心技巧：把所有请求共享的稳定前缀（如同一个大型背景文档、统一角色设定）放进system并用cache_control标记，让批次内的多个请求共享同一份缓存。仓库 python/claude-api/batches.md 给出了共享 system 的构建方式（TS 版结构完全相同，只是类型书写不同）：

const sharedSystem: Anthropic.TextBlockParam[] = [ { type: "text", text: "You are a literary analyst." }, { type: "text", text: largeDocumentText, // 所有请求共享的上下文 cache_control: { type: "ephemeral" }, }, ]; const messageBatch = await client.messages.batches.create({ requests: questions.map((question, i) => ({ custom_id: `analysis-${i}`, params: { model: "claude-opus-5", max_tokens: 16000, system: sharedSystem, messages: [{ role: "user", content: question }], }, })), });

缓存相关验证字段（在普通消息响应中同样适用）：usage.cache_creation_input_tokens（写入缓存的 token，约 1.25 倍成本）、usage.cache_read_input_tokens（命中缓存的 token，约 0.1 倍成本）、usage.input_tokens（未命中部分，全价）。完整的放置模式与"静默失效项"排查清单见 skills/claude-api/shared/prompt-caching.md。

并发批次中的缓存是 best-effort：skills/claude-api/shared/cost-optimization.md 明确指出，并发批处理场景下的缓存命中并不保证，排查缓存命中率异常时不要把并发批次带来的 miss 当作"缓存被破坏"的故障。

九、设计约束与成本优化视角

从仓库成本优化文档（skills/claude-api/shared/cost-optimization.md）可以提炼出批处理场景的两条重要设计事实：

1. 批内请求是单发的（single-shot），无中间工具循环。批次执行期间不会像交互式对话那样自动执行"模型调用工具 → 回填结果 → 再调用"的循环。如果你的任务包含工具循环，有两种处理路径：

   • 保持交互式实时调用（放弃半价）；
   • 预先拉取工具所需输入、把工具循环扁平化为一个可批处理的单请求。文档中的实践案例显示，扁平化后的批处理运行成本约为原交互配置的一半，但这是架构决策而非参数调整——它改变了模型推理方式，文中示例的通过率保持不如交互式稳定，需结合评测权衡。

2. 批处理上限就是成本上限：成本优化文档将批处理定位为"标准档流量中无人等待部分"的 5 折天花板，建议先用service_tier分组观测哪些流量已在批次档、哪些可以迁移，再据此规划迁移范围。

此外还有一处易被忽略的成本细节：上下文编辑（context editing）会重写缓存对话，是省缓存钱的反面操作。批处理任务里若每个请求都从同一共享上下文出发且频繁清空旧内容，缓存收益会被抵消；共享前缀应保持字节级稳定，任何动态内容（如Date.now()、UUID）都要放在user消息而非共享前缀中。

十、跨语言对照：一套 REST 模型，五套 SDK 绑定

Message Batches API 的资源模型在各语言 SDK 中高度一致，便于团队多语言协作时对齐：

其中 Python 版还额外展示了列出批次（client.messages.batches.list(limit=20)）能力：迭代list()返回值会自动跨页翻完所有批次；如需手动控制分页，可用first_page.has_next_page()/get_next_page()/last_id游标。TS SDK 的batches.list命名空间与之对应，可按同样模式使用。

十一、注意事项清单（Checklist）

收尾前，把本文涉及的易错点集中成清单，方便直接对照落地：

1. 配额先行：单批 ≤ 100,000 请求且 ≤ 256 MB，超限会创建失败；
2. 不要同步等待：批次最长 24 小时，客户端必须设计为"提交 → 轮询 → 取结果"的异步流水线（任务队列 / cron / worker 均可）；
3. 结果会过期：29 天保留期，务必在期限内消费并落盘，expired结果需重新提交；
4. 结果按 custom_id 归位：读取结果时用custom_id做映射，不要在succeeded之外的类型上访问message字段；
5. 区分两类错误：errored中invalid_request需修参数重提，服务端错误可直接重试；
6. 取消是异步的：cancel()返回canceling，需轮询确认进入取消终态；
7. 内容块先收窄再取文本：content[0].text在 TS 类型上不是安全访问，务必按block.type判别后再读取；
8. 缓存前缀要稳定：共享 system 加cache_control，动态内容全部下沉到user消息；
9. 工具循环不适用：批次内请求是单发的，涉及工具链的任务要么预拉取输入扁平化，要么保持实时调用；
10. 模型与版本以现状为准：文中示例模型claude-opus-5取自本仓库文档；具体模型可用性、beta 头等信息请以 skills/claude-api/shared/models.md 及 SDK 发行说明为准。

小结

Message Batches API 是 Claude 平台上"低成本、高吞吐"离线工作负载的标准解法：用 50% 的价格换取最长 24 小时的异步排队，配合提示词缓存可在批内进一步共享上下文成本。TypeScript 侧只需掌握create → retrieve → results → cancel四个方法即可搭建完整流水线，而仓库内 Python / C# / PHP / cURL 的同构实现，为多语言团队提供了统一的可对照参考。

• 人工智能
• AI 技能
• AI 评测