同样的请求，走同步 API 是一个价，走 Batch API 直接打五折。一次提交最多 10 万个请求，几小时内跑完。如果你每个月底都在为批量处理文档付全价，这篇文章能帮你省一半。

一、先说结论：不是所有请求都需要"秒回"

我在香港做 FinTech，每个月都有这么几类活：给几千条客户反馈做情绪分类、给一批合同做条款提取、给历史数据做批量清洗标注。

最开始我图省事，直接循环调用同步 API——结果就是慢，而且贵。几千条数据，每条都按全价付 token，月底账单看着肉疼。

直到我把这些活切到 Claude Message Batches API：把上千个请求打包成一个 batch 提交，输入和输出 token 一律半价，让 Anthropic 在后台异步处理。跑批任务不需要秒回，我只需要吞吐量和更低的账单。

这篇文章我把 Batch API 的原理、省钱数学、完整代码、踩坑一次性讲清楚。核心结论先放这：只要你的请求不需要实时响应，切到 Batch API 就能省 50%，和缓存叠加能省到 77%。

收藏提示①：文末的封装函数可以直接复用，套到你自己的批量任务里就能看到账单降下来。

二、环境信息

项	版本
Python	3.13.12
anthropic SDK	最新版
模型	claude-sonnet-4-6（示例用）
价格	Anthropic 官方文档（会变，动手前以官方定价页为准）

三、Batch API 是什么：用延迟换半价

同步调用（client.messages.create）是"我发请求，几秒内拿结果"。Batch API 是"我打包一批请求，几小时后统一拿结果"。

维度	同步 API	Batch API
响应	秒级	几小时（上限 24 小时）
价格	全价	输入输出都半价
单次规模	一个请求	最多 10 万个请求
结果顺序	按调用顺序	无序，靠 custom_id 匹配

本质上，Batch API 就是用延迟换成本——Anthropic 可以在资源空闲时跑你的任务，所以给你打折。对任何"不需要用户在屏幕前等结果"的批量任务，这笔交易都很划算。

四、省钱数学：半价起步，叠加缓存更狠

Batch API 单独用，输入输出 token 都是半价。如果和上篇讲的 Prompt Caching 叠加，还能再砍一刀。

用真实数字算一笔账（Sonnet 4.6，5 万请求/月，5,000 token 系统提示 + 1,000 token 输入 + 300 token 输出）：

无任何优化：
  (5000+1000) × $3/M × 50000 + 300 × $15/M × 50000 = $1,125/月

Batch + 缓存叠加：
  系统提示走缓存  5000 × $0.30/M × 50000 = $75
  输入走 Batch    1000 × $1.50/M × 50000 = $75
  输出走 Batch     300 × $7.50/M × 50000 = $112.50
  合计 = $262.50/月 —— 比基线省 77%

也就是说，Batch API 负责"半价"，Prompt Caching 负责"缓存读取 90% off"，两者相乘，批量任务的成本能从 $1,125 压到 $262。这就是上篇和这篇加起来想讲的完整答案。

收藏提示②：Batch 半价 + 缓存 90% off 可以叠加，不是二选一。批量任务两个都开，省得最多。

上图：同样的 5 万请求/月，无优化 $1,125 / Batch $562.5 / Batch+缓存 $262.5。两个折扣相乘，批量账单直接砍掉七成以上。

五、完整实战：一个可复用的批量封装

先看 Batch API 的完整工作流程：

上图：核心步骤是"构建 requests（带 custom_id）→ 提交 batch → 轮询状态 → 检查 counts → 按 custom_id 匹配结果"。custom_id 是命脉，下面我会重点讲。

下面是我实际在用的封装，把"提交 → 轮询 → 匹配"整个流程包起来：

下面是我实际在用的封装，把"提交 → 轮询 → 匹配结果"整个流程包起来：

import time
import anthropic
from anthropic.types.message_create_params import MessageCreateParamsNonStreaming
from anthropic.types.messages.batch_create_params import Request

client = anthropic.Anthropic()

def build_requests(items: list[dict]) -> list[Request]:
    """把待处理的数据包装成 batch 请求，custom_id 用于结果匹配"""
    requests = []
    for item in items:
        requests.append(Request(
            custom_id=item["id"],              # 关键：用你源数据的唯一 ID
            params=MessageCreateParamsNonStreaming(
                model="claude-sonnet-4-6",
                max_tokens=512,
                system="你是分类助手，只输出一个标签，不要解释。",
                messages=[{"role": "user", "content": item["text"]}],
            ),
        ))
    return requests

def run_batch(requests: list[Request], poll_seconds: int = 60) -> list[dict]:
    """提交 batch，轮询直到结束，按 custom_id 匹配返回结果"""
    batch = client.messages.batches.create(requests=requests)
    print(f"Batch ID: {batch.id}")

    # 轮询状态，直到 ended
    while True:
        batch = client.messages.batches.retrieve(batch.id)
        if batch.processing_status == "ended":
            break
        print(f"状态: {batch.processing_status}，{poll_seconds} 秒后再查")
        time.sleep(poll_seconds)

    # 检查有多少成功/失败
    print(f"请求计数: {batch.request_counts}")

    # 解析结果（JSONL，顺序不保证，靠 custom_id 匹配）
    results = []
    for result in client.messages.batches.results(batch.id):
        if result.result.type == "succeeded":
            results.append({
                "id": result.custom_id,
                "output": result.result.message.content[0].text,
            })
        else:
            results.append({
                "id": result.custom_id,
                "error": result.result.error.type,
            })
    return results

# 用法：批量分类客户反馈
items = [
    {"id": "fb-001", "text": "扣款重复了，麻烦退款"},
    {"id": "fb-002", "text": "App 上传头像就闪退"},
    # ... 几千条
]
requests = build_requests(items)
results = run_batch(requests)
print(f"处理完成 {len(results)} 条")

这套代码最核心的一点是 custom_id。Batch 的结果返回顺序和你提交的顺序不保证一致，所以每个请求都要带一个能从源数据反查的唯一 ID（数据库主键、文件名、工单号），拿到结果后再按 ID 拼回去。

六、踩坑记录：这四个坑，别踩

1. custom_id 一定要唯一且可反查：结果是无序返回的，如果 custom_id 只是 req_0、req_1 这种自增序号，一旦中间有请求失败，你根本对不上是哪条。用源数据的真实 ID。
2. 结束别只看状态，要看 request_counts：batch 状态变成 ended 不代表全成功。request_counts 里会列出 succeeded / errored / canceled / expired 的数量，先核对这个，再假设结果干净。
3. 24 小时是硬上限：单个 batch 最多处理 24 小时。如果任务量大到可能超时，拆成多个 batch，别硬塞一个。
4. 结果要按 custom_id 重新拼回顺序：Batch 返回的是 JSONL（每行一个结果），顺序随机。拿到结果后一定要用 custom_id 做一次 map，还原成你源数据的顺序，否则下游分析会乱。

七、什么时候用 Batch，什么时候用流式

决策就一个核心问题：用户是不是正在屏幕前等这个结果？

场景	选择	原因
对话式交互、用户实时等	流式	首字延迟（TTFT）最重要
批量分类、夜间跑批、离线评测	Batch	不需要秒回，省 50%
结果几分钟内就要	同步	Batch 至少几小时
结果能等几小时	Batch	半价划算

一句话：交互走流式，批处理走 Batch。把这两类请求分开，比一刀切省得多。

八、写在最后

Batch API 说白了就是一句大白话：“我不急，你打折给我跑”。技术上没有任何门槛，就是换个 endpoint、加个 custom_id、写个轮询循环。

但据我观察，很多人明知道批量任务不急着要结果，还是习惯性地循环调同步 API——因为"省事"。结果就是每个月为"秒回"这种你根本不需要的特性，多付了一倍的钱。

配合上一篇的 Prompt Caching，这两篇合起来就是 Claude API 省成本的两板斧：缓存省重复输入的 90%，Batch 省全部输入的 50%。批量任务两个都开，账单能砍掉七成以上。

需要说明的是，价格和 24 小时上限这些是版本敏感信息，会随 Anthropic 调整，本文以写作时的官方文档为准，动手前先查最新定价。

收藏提示③：如果这篇帮你省了钱，收藏 + 点赞，下次做批量任务时直接回来抄代码。