Claude API 批量处理完全指南：Message Batches API 大规模数据处理实战

import anthropic
import json
import time

client = anthropic.Anthropic()

# === 1. 准备批量请求 ===
def create_batch_requests(texts: list[str], task: str) -> list:
    requests = []
    for i, text in enumerate(texts):
        requests.append({
            "custom_id": f"request-{i}",  # 用于匹配结果
            "params": {
                "model": "claude-haiku-3-5",  # 批量任务推荐 Haiku（更便宜）
                "max_tokens": 200,
                "messages": [{
                    "role": "user",
                    "content": f"{task}\n\n{text}"
                }]
            }
        })
    return requests

# === 2. 提交批次 ===
def submit_batch(requests: list):
    batch = client.beta.messages.batches.create(requests=requests)
    print(f"批次已提交: {batch.id}")
    print(f"状态: {batch.processing_status}")
    print(f"请求数: {batch.request_counts.processing}")
    return batch.id

# === 3. 轮询进度 ===
def wait_for_batch(batch_id: str, poll_interval: int = 30):
    while True:
        batch = client.beta.messages.batches.retrieve(batch_id)
        counts = batch.request_counts
        total = counts.processing + counts.succeeded + counts.errored
        done = counts.succeeded + counts.errored
        print(f"进度: {done}/{total} | 成功: {counts.succeeded} | 失败: {counts.errored}")

        if batch.processing_status == "ended":
            print(f"批次完成！结果文件: {batch.results_url}")
            return batch
        
        time.sleep(poll_interval)

# === 4. 获取结果 ===
def get_results(batch_id: str) -> dict:
    results = {}
    for result in client.beta.messages.batches.results(batch_id):
        if result.result.type == "succeeded":
            results[result.custom_id] = result.result.message.content[0].text
        else:
            results[result.custom_id] = f"ERROR: {result.result.error.type}"
    return results

# === 完整流程 ===
texts = [
    "Artificial intelligence is transforming software development...",
    "The latest Claude model shows significant improvements...",
    # ... 更多文本
]

requests = create_batch_requests(texts, "请将以下英文翻译成中文：")
batch_id = submit_batch(requests)
batch = wait_for_batch(batch_id)
results = get_results(batch_id)

for custom_id, translation in results.items():
    print(f"{custom_id}: {translation[:50]}...")

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

const client = new Anthropic();

async function batchClassify(items: string[]) {
  // 提交批次
  const batch = await client.beta.messages.batches.create({
    requests: items.map((item, i) => ({
      custom_id: `item-${i}`,
      params: {
        model: "claude-haiku-3-5",
        max_tokens: 50,
        messages: [{
          role: "user" as const,
          content: `将以下文本分类为：正面/负面/中性。只输出类别词。\n\n${item}`
        }]
      }
    }))
  });

  console.log(`批次 ID: ${batch.id}`);

  // 等待完成
  let current = batch;
  while (current.processing_status !== "ended") {
    await new Promise(r => setTimeout(r, 30000)); // 等 30 秒
    current = await client.beta.messages.batches.retrieve(batch.id);
    const { succeeded, errored, processing } = current.request_counts;
    console.log(`进度: 成功 ${succeeded}, 失败 ${errored}, 处理中 ${processing}`);
  }

  // 收集结果
  const results: Record<string, string> = {};
  for await (const result of await client.beta.messages.batches.results(batch.id)) {
    if (result.result.type === "succeeded") {
      results[result.custom_id] = result.result.message.content[0].type === "text"
        ? result.result.message.content[0].text : "";
    }
  }
  return results;
}

import os, glob

def batch_summarize_documents(docs_dir: str):
    # 读取所有 .txt 文件
    files = glob.glob(f"{docs_dir}/*.txt")
    texts = {}
    for path in files:
        with open(path) as f:
            texts[os.path.basename(path)] = f.read()[:4000]  # 限制长度
    
    # 构建批量请求
    requests = [{
        "custom_id": filename,
        "params": {
            "model": "claude-haiku-3-5",
            "max_tokens": 300,
            "messages": [{
                "role": "user",
                "content": f"用 3 句话概括以下文档的核心内容：\n\n{content}"
            }]
        }
    } for filename, content in texts.items()]
    
    # 提交（每批最多 10000 条）
    batch = client.beta.messages.batches.create(requests=requests[:10000])
    print(f"提交 {len(requests)} 篇文档，批次 ID: {batch.id}")
    return batch.id