更新于

Anthropic Batch API 国内使用指南

AnthropicClaudeBatch API技术深度

Anthropic 的 Message Batches API（批量消息接口）允许开发者一次性提交数千条请求，异步处理后统一取回结果，单价仅为标准接口的 50%。对于需要大批量处理文档、标注、评估或离线推理的场景，这是降本增效的利器。

然而国内开发者访问 Anthropic 原生接口存在网络障碍，本文从原理到实战，完整介绍 Batch API 的使用方式，以及如何通过国内 API 中转稳定使用。

一、Batch API 的核心原理

Message Batches API 是 Anthropic 为非实时、大批量场景设计的异步接口。与标准 /messages 接口的区别如下：

对比维度标准 Messages APIMessage Batches API
请求方式同步，即时返回异步，提交后轮询
单价标准定价约 50% 折扣
最大批量1 条单批 10,000 条
超时限制无（SSE 流式）批次 24 小时内完成
适用场景在线问答、实时应用数据标注、批量评估、离线任务

Batch API 的处理流程分三步：

  1. 创建批次：POST 一个包含多条请求的 JSON，得到 batch_id
  2. 轮询状态：GET 该 batch_id 的状态，直到 processing_status 变为 ended
  3. 下载结果：GET 结果文件，按 custom_id 对应回每条原始请求

二、请求格式详解

批次请求的主体是一个 requests 数组，每条请求包含：

POST /v1/messages/batches
{
  "requests": [
    {
      "custom_id": "req-001",
      "params": {
        "model": "claude-sonnet-4-6",
        "max_tokens": 1024,
        "messages": [
          {"role": "user", "content": "请用一句话总结：人工智能的本质是什么？"}
        ]
      }
    },
    {
      "custom_id": "req-002",
      "params": {
        "model": "claude-sonnet-4-6",
        "max_tokens": 1024,
        "messages": [
          {"role": "user", "content": "请将以下文本翻译成英文：今天天气很好。"}
        ]
      }
    }
  ]
}

custom_id 是你自定义的唯一标识符，结果中会原样返回，方便你将结果映射回原始数据。每条 params 与标准 Messages API 的请求体完全相同，支持 system、tools、temperature 等所有参数。

三、Python 完整调用示例

以下是用 anthropic SDK 进行批量处理的完整示例，适配国内中转：

import anthropic
import time
import json

# 使用国内中转时替换 base_url
client = anthropic.Anthropic(
    api_key="your-relay-api-key",
    base_url="https://api.yotradeapi.com",  # 国内中转地址
)

# 准备批量请求
def build_batch_requests(texts: list[str]) -> list[dict]:
    return [
        {
            "custom_id": f"item-{i}",
            "params": {
                "model": "claude-sonnet-4-6",
                "max_tokens": 512,
                "messages": [
                    {"role": "user", "content": f"请给以下文本打情感标签（正面/负面/中性）：\n{text}"}
                ],
            },
        }
        for i, text in enumerate(texts)
    ]

# 创建批次
texts = [
    "这款产品质量太差了，完全不值这个价格",
    "服务非常好，客服响应及时，下次还会购买",
    "产品一般，价格也合理，没有特别的感受",
]

batch = client.messages.batches.create(
    requests=build_batch_requests(texts)
)
print(f"批次已创建：{batch.id}，状态：{batch.processing_status}")

# 轮询等待完成
def wait_for_batch(client, batch_id: str, poll_interval: int = 30):
    while True:
        batch = client.messages.batches.retrieve(batch_id)
        print(f"状态：{batch.processing_status}，"
              f"成功：{batch.request_counts.succeeded}，"
              f"失败：{batch.request_counts.errored}")
        if batch.processing_status == "ended":
            return batch
        time.sleep(poll_interval)

completed_batch = wait_for_batch(client, batch.id)

# 处理结果
results = {}
for result in client.messages.batches.results(completed_batch.id):
    custom_id = result.custom_id
    if result.result.type == "succeeded":
        content = result.result.message.content[0].text
        results[custom_id] = content
    else:
        results[custom_id] = f"ERROR: {result.result.error.type}"

# 打印结果
for i, text in enumerate(texts):
    key = f"item-{i}"
    print(f"\n原文：{text}")
    print(f"标签：{results.get(key, 'N/A')}")

四、国内接入：中转兼容性说明

Anthropic 官方 API 对中国大陆网络不友好，稳定调用需要借助 API 中转服务。好消息是，支持 Anthropic 协议的中转服务 对 Batch API 的支持路径与标准接口完全一致，只需：

  1. 将 client = anthropic.Anthropic(base_url="https://api.yotradeapi.com") 中的 base_url 指向中转地址
  2. 使用中转提供的 API Key 替换原有 Key
  3. 其余代码完全不变

如果中转服务仅支持 OpenAI 兼容协议（/v1/chat/completions），则不支持 Batch API，因为 Batch API 是 Anthropic 私有协议。选用中转服务时需确认其是否原生转发 /v1/messages/batches 端点。

五、错误处理与重试策略

批次级别和请求级别有两种错误需要区分处理：

# 检查整体批次状态
if completed_batch.processing_status != "ended":
    raise RuntimeError(f"批次未正常结束：{completed_batch.processing_status}")

# 检查单条请求结果
failed_ids = []
for result in client.messages.batches.results(completed_batch.id):
    if result.result.type == "errored":
        error = result.result.error
        print(f"请求 {result.custom_id} 失败：{error.type} - {error.message}")
        failed_ids.append(result.custom_id)

# 对失败的请求可以单独重提一个新批次
if failed_ids:
    print(f"需要重试 {len(failed_ids)} 条请求")

常见错误类型：

错误类型含义处理建议
invalid_request请求参数格式错误检查 params 结构
overloaded_error模型过载（概率较低）对该条重试
api_error服务端内部错误重试整批或单条

六、大规模任务的分批策略

单批上限是 10,000 条，但实际运营中建议每批控制在 1,000–3,000 条：

  • 更快完成（大批次完成时间更不可预测）
  • 更容易追踪失败并局部重试
  • 降低单批失败时的损失

分批处理的参考代码：

def chunked(lst, size):
    for i in range(0, len(lst), size):
        yield lst[i:i + size]

all_texts = [...]  # 假设有 10000 条
batch_size = 2000

batch_ids = []
for chunk in chunked(all_texts, batch_size):
    batch = client.messages.batches.create(
        requests=build_batch_requests(chunk)
    )
    batch_ids.append(batch.id)
    print(f"提交批次 {batch.id}，共 {len(chunk)} 条")

# 并发轮询所有批次
import concurrent.futures

def poll_and_collect(batch_id):
    completed = wait_for_batch(client, batch_id)
    return list(client.messages.batches.results(completed.id))

with concurrent.futures.ThreadPoolExecutor(max_workers=5) as executor:
    all_results = list(executor.map(poll_and_collect, batch_ids))

七、适用场景与成本估算

典型适合 Batch API 的场景：

  • 大规模文档摘要（每天处理数千篇文章）
  • 训练数据标注（正负样本分类、质量评分）
  • 离线评估流水线（LLM-as-Judge 批量打分）
  • 批量翻译（产品描述、用户评论多语言化）
  • 代码批量审查（扫描整个仓库的文件）

成本估算示例（仅供参考）：

假设使用 claude-sonnet-4-6，平均每条请求消耗 500 input tokens + 200 output tokens，处理 10,000 条：

  • 标准 API：10,000 × (500 × $3/M + 200 × $15/M) ≈ $45
  • Batch API：同上 × 50% ≈ $22.5

10,000 条请求节省约 $22.5。对于每日数十万条的生产任务，节省金额将相当可观。

八、相关阅读

需要稳定调用 Anthropic Batch API 或其他 Claude 接口，YoTradeApi 提供原生 Anthropic 协议中转，批量推理同样享受半价优惠。

相关阅读