Anthropic Message Batches API 教程：用 TypeScript 批量调用 Claude 并降低成本

如果你有一批文本要交给 Claude 处理，但结果并不需要立刻拿到——比如把几万条用户评论分类、把一批文档摘要或翻译、给数据打标签、跑离线评测——那么逐条调用同步的 Messages API 是浪费。Anthropic 的 Message Batches API 就是为这类场景准备的：把请求打包成一个批次提交，之后异步取回结果，代价是等待时间变长，收益是单次调用的成本明显下降。本教程面向已经会用 TypeScript 调用 Claude 的开发者，从安装 SDK 开始，一步步把批量任务跑通。

需要先明确一点：批次接口不是同步接口的替代品。聊天、需要即时反馈的交互式应用、任何要求秒级返回的场景都不适合用它。判断标准很简单——这个结果晚一小时甚至晚一天拿到，业务上能不能接受。能接受，就适合走批次。

准备工作

账号与密钥

你需要一个 Anthropic 账号，并在控制台创建一个 API Key。密钥形如 sk-ant-...，只在创建时完整显示一次，请立刻保存到安全的地方。批次接口与普通 Messages API 共用同一个密钥，不需要额外申请权限。

把密钥写进项目根目录的 .env 文件，不要提交到版本库：

ANTHROPIC_API_KEY=sk-ant-...

运行环境

需要 Node.js 环境（建议使用当前仍在维护的 LTS 版本）以及一个包管理器。示例使用 pnpm，换成 npm 或 yarn 也可以，命令对应调整即可。TypeScript 项目建议开启 strict，SDK 的类型定义比较完整，能帮你在编译期发现字段拼写错误。

安装 SDK

pnpm add @anthropic-ai/sdk

SDK 会自动读取环境变量 ANTHROPIC_API_KEY，所以初始化客户端时可以不传参数；显式传入也可以，便于在不同环境切换密钥。

先想清楚三件事

• 每个请求的唯一标识：批次里的每个请求都要带一个 custom_id，结果返回时靠它对应回原始数据。用数据库主键、订单号这类稳定 ID 最省事。
• 单批次规模：一个批次最多容纳 10,000 个请求。超过这个数量就要拆成多个批次分别提交。
• 结果落盘方式：批次结果以 JSONL 形式流式返回，边读边写比全部读进内存再处理更稳妥。

操作步骤

第一步：构造批次请求

批次接口的请求体是一个 requests 数组，每个元素包含两部分：custom_id 和 params。其中 params 的结构与普通 Messages API 完全一致——model、max_tokens、system、messages 这些字段照常写。换句话说，你原本写好的单条调用参数，几乎可以原样搬进批次里。

import Anthropic from "@anthropic-ai/sdk";

const client = new Anthropic({
  apiKey: process.env.ANTHROPIC_API_KEY,
});

// 待分类的文本列表，实际项目中通常来自数据库或 CSV
const reviews = [
  { id: "review-001", text: "配送很慢，但商品本身还不错" },
  { id: "review-002", text: "客服态度非常耐心" },
  { id: "review-003", text: "质量很差，用了几次就坏了" },
  // ... 最多 10,000 条
];

const SYSTEM_PROMPT = `请把用户评论归类到以下类别之一。
类别：delivery / product_quality / support / price / other
只返回 JSON：{"category": "...", "sentiment": "positive|neutral|negative"}`;

const batch = await client.messages.batches.create({
  requests: reviews.map((review) => ({
    custom_id: review.id,
    params: {
      model: "claude-sonnet-4-6",
      max_tokens: 100,
      system: SYSTEM_PROMPT,
      messages: [{ role: "user", content: review.text }],
    },
  })),
});

console.log(`Batch created: ${batch.id}`);
console.log(`Status: ${batch.processing_status}`);

提交成功后返回的对象里，id 形如 msgbatch_...，processing_status 初始为 in_progress。这个 id 要保存下来，后续查询状态和拉取结果都要用。

关于模型选择：批次接口支持 Claude 全系模型，具体可用型号以官网当前信息为准。对成本敏感、任务又相对简单的场景（分类、打标签），可以选更轻量的型号；需要更强推理能力的场景再换大模型。模型名称属于易变信息，写进代码前建议先核对官方文档。

第二步：轮询等待批次结束

批次是异步执行的，提交之后需要定期查询状态，直到 processing_status 变成 ended。返回对象里的 request_counts 会给出 succeeded、errored、processing 等计数，用来观察进度。

async function waitForBatch(batchId: string): Promise<Anthropic.MessageBatch> {
  const POLL_INTERVAL_MS = 30_000; // 30 秒查一次

  while (true) {
    const batch = await client.messages.batches.retrieve(batchId);
    console.log(
      `[${new Date().toISOString()}] status: ${batch.processing_status} ` +
        `(succeeded: ${batch.request_counts.succeeded}, ` +
        `errored: ${batch.request_counts.errored}, ` +
        `processing: ${batch.request_counts.processing})`
    );

    if (batch.processing_status === "ended") {
      return batch;
    }

    await new Promise((resolve) => setTimeout(resolve, POLL_INTERVAL_MS));
  }
}

const completedBatch = await waitForBatch(batch.id);

轮询间隔没有硬性规定，30 秒到几分钟都合理。批次通常在 1 小时内完成，最长不超过 24 小时。如果是在服务端长期运行的任务，把轮询逻辑交给定时任务调度器（例如 Vercel Cron、Cloud Scheduler 之类的计划任务服务）比在一个进程里死循环更省资源，也更容易在进程重启后恢复。

第三步：读取结果

client.messages.batches.results() 返回的是一个 AsyncIterable，逐条产出结果条目。因为是流式读取，即使批次里有 10,000 个请求，内存占用也不会随规模线性增长。

每个条目包含 custom_id 和 result，而 result.type 有三种可能：

下面这段代码把三种情况都覆盖了，并且对模型返回的文本做了 JSON 解析保护——模型偶尔会带上多余文字，直接 JSON.parse 会抛异常：

type ClassificationResult = {
  category: string;
  sentiment: "positive" | "neutral" | "negative";
};

const results: Array<{
  id: string;
  data: ClassificationResult | null;
  error?: string;
}> = [];

for await (const entry of await client.messages.batches.results(completedBatch.id)) {
  if (entry.result.type === "succeeded") {
    const raw = entry.result.message.content[0];
    if (raw.type !== "text") {
      results.push({ id: entry.custom_id, data: null, error: "unexpected content type" });
      continue;
    }
    try {
      const parsed: ClassificationResult = JSON.parse(raw.text);
      results.push({ id: entry.custom_id, data: parsed });
    } catch {
      results.push({
        id: entry.custom_id,
        data: null,
        error: `parse failed: ${raw.text}`,
      });
    }
  } else if (entry.result.type === "errored") {
    results.push({
      id: entry.custom_id,
      data: null,
      error: entry.result.error.type,
    });
  } else {
    // expired：批次未在 24 小时内处理完
    results.push({ id: entry.custom_id, data: null, error: "expired" });
  }
}

const succeeded = results.filter((r) => r.data !== null);
const failed = results.filter((r) => r.data === null);
console.log(`Done: ${succeeded.length} succeeded, ${failed.length} failed`);

注意结果条目的顺序不保证与提交顺序一致，所以一定要靠 custom_id 做映射，不要依赖数组下标。

第四步：叠加 Prompt Caching 进一步压缩成本

如果每个请求都带同一段很长的 system prompt，那么这段内容会在每个请求里重复计费。Prompt Caching 允许把这段前缀缓存起来，命中缓存时输入 token 的费用大幅下降。在批次场景里，把 system 从字符串改成数组，并在文本块上加 cache_control 即可：

const batch = await client.messages.batches.create({
  requests: reviews.map((review) => ({
    custom_id: review.id,
    params: {
      model: "claude-sonnet-4-6",
      max_tokens: 100,
      system: [
        {
          type: "text",
          text: SYSTEM_PROMPT,
          cache_control: { type: "ephemeral" }, // 显式声明缓存
        },
      ],
      messages: [{ role: "user", content: review.text }],
    },
  })),
});

缓存有最小长度门槛：system prompt 需要达到一定 token 数（约 1,024 token）才会生效。达到门槛后，整个批次通常只发生一次缓存写入，后续请求都命中缓存。批次本身的折扣与缓存折扣可以叠加，因此长 system prompt 的大批量任务，成本压缩空间相当可观。具体的折扣比例、缓存有效期与最小 token 数属于会调整的信息，请以官网当前信息为准。

一个完整示例

把前面的步骤串起来，下面是一个可以直接运行的最小脚本。它接收一组 { id, text }，提交批次、等待完成、收集结果，最后返回 custom_id 到模型输出的映射。

import Anthropic from "@anthropic-ai/sdk";

const client = new Anthropic();

async function runClassificationBatch(
  texts: { id: string; text: string }[]
) {
  // 1. 创建批次
  const batch = await client.messages.batches.create({
    requests: texts.map(({ id, text }) => ({
      custom_id: id,
      params: {
        model: "claude-haiku-4-5-20251001", // 成本优先时可选轻量型号
        max_tokens: 100,
        messages: [
          {
            role: "user",
            content: `Classify: ${text}\nJSON only: {"category":"..."}`,
          },
        ],
      },
    })),
  });

  console.log(`Batch ${batch.id} submitted (${texts.length} requests)`);

  // 2. 等待完成
  let current = batch;
  while (current.processing_status !== "ended") {
    await new Promise((r) => setTimeout(r, 30_000));
    current = await client.messages.batches.retrieve(batch.id);
  }

  // 3. 读取结果
  const output: Record<string, string> = {};
  for await (const entry of await client.messages.batches.results(batch.id)) {
    if (entry.result.type === "succeeded") {
      const text = entry.result.message.content[0];
      if (text.type === "text") {
        output[entry.custom_id] = text.text;
      }
    }
  }

  return output;
}

调用方式：

const result = await runClassificationBatch([
  { id: "review-001", text: "配送很慢，但商品本身还不错" },
  { id: "review-002", text: "客服态度非常耐心" },
]);

console.log(result);

这个脚本刻意省略了错误处理与结果落盘，目的是让主流程清晰。生产环境里建议补上三点：把 errored 与 expired 的 custom_id 收集起来单独重试；把结果写入数据库或 JSONL 文件而不是只放内存；把轮询交给计划任务，避免进程长时间挂起。

适用场景与不适用场景

批次接口适合的任务类型比较集中：

• 文档的批量分类、摘要、翻译
• 数据增强，例如地址规范化、自动打标签
• 夜间批处理任务，例如生成评估报告
• 用 LLM-as-Judge 做离线评测，也可以接进 CI 流程

反过来，聊天机器人、需要实时 API 响应的交互式产品、任何用户在前台等待结果的场景，都不适合用批次。判断依据始终是那一条：这个结果能不能等。

注意事项

• 单批次上限 10,000 个请求。超出就要拆分，拆分时注意每个批次的 custom_id 仍然要全局唯一，方便后续合并结果。
• 完成时间不固定。通常 1 小时内返回，最长可能到 24 小时。超过 24 小时仍未处理的请求会以 expired 状态返回，需要重新提交。
• 结果顺序不保证。必须用 custom_id 做映射，不要假设返回顺序与提交顺序一致。
• 模型输出不一定是合法 JSON。即使 prompt 里明确要求只返回 JSON，也要用 try/catch 包住解析逻辑，并把解析失败的原文记录下来，便于排查。
• 批次状态是异步的。提交成功只代表请求被接受，不代表处理完成，务必轮询到 ended 再取结果。
• 折扣比例、可用模型、缓存门槛都会变动。本文提到的折扣、token 门槛、模型名称等信息请以官网当前信息为准，不要把它们硬编码成业务假设。
• 密钥不要写进代码。使用环境变量或密钥管理服务，并注意批次结果里可能包含原始业务数据，落盘时要考虑访问控制。