Anthropic SDK Batch 批处理

第 7 篇用提示缓存把重复前缀的输入打到一折，但每一分新内容仍按全价计费，而且每个请求都得当场等回复。工程里还有一类任务根本没人等：夜里跑几万条文档摘要、给全站历史评论打情感标签、把一批评测集整体重跑——量大、单条独立、今晚出结果还是明早出结果无所谓。第 7 篇结尾预告的半价通道就是为它们准备的：Message Batches API 把请求打包提交，服务端排进队列异步消化，所有 token 一律五折。

本文是《Anthropic SDK 从入门到实战》第 8 篇。全文按"适用判断 → 请求构造 → 提交跟踪 → 结果解析 → 成本账 → 结构化组合 → 实战 → 总结"组织，示例延续第 1 篇装好的环境与 2.3.0 版 SDK。

1、什么任务适合批处理

批处理的交换条件一句话讲完：用等待时间换五折。服务端把整批请求排进队列按自己的节奏消化，不承诺任何一条何时完成，只承诺一个总窗口。判断该不该进批，只看三个问题，决策流如下：

解读：两个判断都落到"时间敏感"上。只要有人盯着屏幕等，或者下游流水线几小时内就要消费结果，就老老实实走同步 create()；反过来，夜间批跑、次日交付的任务，进批等于白拿五折。典型分布如下：

• 适合：全量文档摘要、知识库预构建、评论与工单打标、评测集批量跑分
• 不适合：在线对话、Agent 实时工具循环、用户正在等待的任何链路
• 边界：强依赖上一步输出的流水线别进批——等待会逐级放大

规模上还有一个硬边界：每批最多 100,000 条请求或 256 MB，先到为准，更大的量拆成多批提交。工程形态上，批处理天然适配定时任务：一个脚本睡前提交、另一个脚本晨起取结果，中间不需要任何常驻进程盯着。限制与计费的权威口径见官方批处理文档。

2、构造批处理请求列表

一个批就是一组"编了号的 Messages 请求"。每条请求两个字段：custom_id 是你起的编号，params 就是单条 Messages API 的请求体。构造示例如下：

import anthropic
from anthropic.types.message_create_params import (
    MessageCreateParamsNonStreaming)
from anthropic.types.messages.batch_create_params import Request

client = anthropic.Anthropic()

requests = [
    Request(
        custom_id="doc-001",                    # 批内唯一：对账的钥匙
        params=MessageCreateParamsNonStreaming(
            model="claude-sonnet-5-5",
            max_tokens=1024,
            messages=[{"role": "user",
                       "content": "用三句话总结这份年报的要点"}],
        ),
    ),
    Request(
        custom_id="doc-002",
        params=MessageCreateParamsNonStreaming(
            model="claude-sonnet-5-5",
            max_tokens=1024,
            messages=[{"role": "user",
                       "content": "用三句话总结这份招股书的要点"}],
        ),
    ),
]

解读：params 的字段与非流式 create() 完全一致——model、max_tokens、system、messages、tools、cache_control、output_config 全部照搬。换句话说，第 1 篇以来写过的每一个请求体，原样塞进 params 就是合法的批请求，Request 与 MessageCreateParamsNonStreaming 只是 SDK 的静态类型包装，本质就是 custom_id 加 params 两个键的字典。custom_id 在批内必须唯一，它是把结果对回输入的唯一钥匙——因为结果的返回顺序是任意的，第 4 节会兑现这句话。字段全集见官方 Messages API 参考。

3、提交与状态轮询

列表构造好，一次 create() 整批提交，然后轮询等它跑完。完整代码如下：

import time

batch = client.messages.batches.create(requests=requests)
print(batch.id, batch.processing_status)   # msgbatch_ 开头，in_progress

while True:
    batch = client.messages.batches.retrieve(batch.id)
    if batch.processing_status == "ended":
        break
    c = batch.request_counts               # 三项计数实时更新
    print(f"处理中 {c.processing} 成功 {c.succeeded} 失败 {c.errored}")
    time.sleep(60)                         # 小批几分钟即完，别轮询太密

print(f"结束：成功 {batch.request_counts.succeeded} "
      f"失败 {batch.request_counts.errored}")

解读：create() 立刻返回，不等任何一条执行；retrieve() 拉取最新状态，request_counts 里的三项计数让进度可见。processing_status 的全部取值如下：

时间预期要有数：多数批次在 1 小时内完成，上限 24 小时。24 小时窗口内没轮到的请求会以 expired 收场，需要重新提交；已完成的批次结果保留 29 天，不必连夜取回。中途反悔可以调 client.messages.batches.cancel(batch.id)，状态先变 canceling，未开始的请求不再执行。日常运维则靠 client.messages.batches.list()，自动翻页列出历史批次，对账与排查都从它入手。

提醒：五折买的是"最迟 24 小时"，不是"24 小时内任意时刻"。对截止时间敏感的任务，要么给批次留足余量，要么别进批。

4、解析批处理结果

ended 之后取结果。results() 返回一个迭代器，逐行吐出每条请求的结果，解析骨架如下：

results, retry, invalid = {}, [], []
for r in client.messages.batches.results(batch.id):
    match r.result.type:
        case "succeeded":
            msg = r.result.message          # 就是一个 Message 对象
            results[r.custom_id] = next(
                (b.text for b in msg.content if b.type == "text"), "")
        case "errored":
            if r.result.error.type == "invalid_request":
                invalid.append(r.custom_id)  # 请求本身错：改参数重交
            else:
                retry.append(r.custom_id)    # 服务端抖动：可原样重试
        case "canceled" | "expired":
            retry.append(r.custom_id)        # 整批取消或超时：重交

解读：succeeded 分支里的 r.result.message 与同步调用的响应对象同构，第 1 篇的遍历判型照用。errored 要分两类：invalid_request 说明这条请求本身非法（比如漏了 max_tokens），改好参数再交；其余是服务端瞬时错误，原样重试即可。SDK 之下，HTTP 层的原始形态是两个分离的 jsonl 文件——成功与失败各一个，每行一个 JSON 对象，样例如下：

{"custom_id":"review-0007","result":{"type":"succeeded","message":{"id":"msg_01...","content":[...]}}}
{"custom_id":"review-0042","result":{"type":"errored","error":{"type":"invalid_request","message":"max_tokens: Field required"}}}

解读：results() 迭代器做的事就是把这两个文件拉下来合并成一条流。四种 result.type 的含义与处理汇总如下：

核心纪律一条：按 custom_id 对账，不按位置——第 1 万条可能比第 1 条先回来，收集端永远用字典以 custom_id 为键，处理完再按需排序。另一条好消息是单条失败不拖垮整批：499 条成功、1 条 invalid_request 是常态，失败名单攒起来单独重交一个小批就行。

5、五折账与缓存叠加

五折覆盖全部计费项——普通输入、输出、缓存写入、缓存读取，无一例外。以两个常用档位为例，同步与 Batch 的单价对照如下（每百万 token，2026 年 10 月官方标准价）：

解读：表中后两行正是与第 7 篇提示缓存的乘法叠加——缓存价格先按倍数算出，再整体打五折。Sonnet 的缓存写入本来是 1.25 倍（2.50 美元），进批后 0.625 倍（1.25 美元）；缓存读取本来 0.1 倍（0.20 美元），进批后 0.05 倍（0.10 美元），只剩原价的二十分之一。批内共享同一段大 system 的任务——比如 500 条评论共用同一套分类规则——先付一次便宜的写入、后续全是深度折扣的读取，"重复前缀"与"不急任务"两个折扣都吃到。价格以官方模型概览为准，叠加细则见官方提示缓存文档。

输出 token 同样五折，长文摘要、逐条改写这类输出大户进批收益更明显——折扣作用在整张账单上，输入输出两头都砍半。

代价只有一样：时间。小批常见几分钟到一小时，大批排队更久，且批处理与快速模式互斥——要快就别进批，要省就别求快。

6、批量结构化输出

批量抽取是批处理的主场，但它与第 3 篇的结构化输出怎么组合，需要先讲清一个核实结论：client.messages.parse() 是 SDK 在客户端做的三层包装——把 Pydantic 模型转成 schema、随请求下发、响应回来后反序列化；Batch 的 requests 只认原始 Messages 参数，没有 parse 入口。替代姿势分两步：schema 随 params 原样下发，反序列化在收结果时自己做。第一步的写法如下：

SENTIMENT_SCHEMA = {
    "type": "object",
    "properties": {
        "sentiment": {"type": "string",
                      "enum": ["positive", "negative", "neutral"]},
        "confidence": {"type": "number"},
    },
    "required": ["sentiment", "confidence"],
    "additionalProperties": False,
}

params = MessageCreateParamsNonStreaming(
    model="claude-haiku-5-5",
    max_tokens=512,
    output_config={"format": {"type": "json_schema",
                              "schema": SENTIMENT_SCHEMA}},
    messages=[{"role": "user", "content": "这条评论是好评还是差评……"}],
)

解读：output_config.format 就是第 3 篇讲过的 API 层正式参数，在批里原样进 params，服务端同样保证首个 text 块是合法 JSON——形状这层保证一分不少。少掉的只是客户端那层便捷包装：Pydantic 校验挪到收集结果的环节自己做，第 7 节的实战把这一步串进去。另有一个批处理送的自由度：params 逐条独立，同一批里混用不同模型、不同 schema 都合法——预算敏感的任务可以让简单条目走 Haiku、疑难条目走 Sonnet，一次提交分层消化。特性细节见官方结构化输出文档。

7、实战：评论情感分类

零件齐了，做完整的一单：500 条商品评论批量情感分类，Haiku 轻量档，结构化输出，失败条目单独回收。先定义业务模型并读入数据：

import json, time
from pydantic import BaseModel, ValidationError

class Sentiment(BaseModel):
    sentiment: str        # positive、negative、neutral
    confidence: float     # 0 到 1

reviews = [json.loads(line) for line
           in open("reviews.jsonl", encoding="utf-8")]  # 500 条
print(len(reviews), "条评论待分类")

解读：schema 沿用第 6 节的 SENTIMENT_SCHEMA，字段与模型严格对齐——Batch 走原始参数，schema 要手写这一份，不像 parse() 能从模型类自动生成。接着构造 500 条请求：

SYSTEM = "你是电商评论情感分类器，输出情感极性与置信度。"

requests = [
    Request(
        custom_id=f"review-{i:04d}",       # 补零编号，对账排障两便
        params=MessageCreateParamsNonStreaming(
            model="claude-haiku-5-5",
            max_tokens=512,
            system=SYSTEM,                 # 500 条共用同一段规则
            output_config={"effort": "low",        # 分类不需要深思考
                           "format": {"type": "json_schema",
                                      "schema": SENTIMENT_SCHEMA}},
            messages=[{"role": "user", "content": r["text"]}],
        ),
    )
    for i, r in enumerate(reviews)
]

解读：output_config 里 effort 与 format 并排——Haiku 默认带思考且思考计入 max_tokens，分类这种轻任务压到 low 档，既省输出又快。提交、等待、收集与校验一气呵成：

batch = client.messages.batches.create(requests=requests)
print(f"批 ID {batch.id}")

while True:
    if client.messages.batches.retrieve(batch.id).processing_status == "ended":
        break
    time.sleep(60)

parsed, failed = {}, []
for r in client.messages.batches.results(batch.id):
    if r.result.type != "succeeded":
        failed.append(r.custom_id)        # 请求失败：进重交名单
        continue
    raw = next(b.text for b in r.result.message.content
               if b.type == "text")
    try:
        parsed[r.custom_id] = Sentiment.model_validate_json(raw)
    except ValidationError:
        failed.append(r.custom_id)        # 业务校验不过：同样回收

pos = sum(1 for v in parsed.values() if v.sentiment == "positive")
print(f"成功 {len(parsed)}，失败 {len(failed)}，好评率 {pos / len(parsed):.1%}")

一次典型运行的输出如下：

500 条评论待分类
批 ID msgbatch_01ABC...
成功 497，失败 3，好评率 68.2%

解读：3 条失败翻一下各自的 error.type，invalid_request 的改参数、其余的原样重交一个小批——单条失败不拖整批的后腿，这正是批处理的容错形态。算账：每条输入约 300 token、输出约 60 token，500 条同步口径约 0.030 美元，Batch 实付约 0.015 美元，半价自动兑现；量放大到每天几万条，一个月就是实打实的对半砍。若再给共享的 SYSTEM 加上 cache_control 断点，按第 5 节的叠加口径，读取部分还能再降一半。

8、总结

本篇把 Message Batches 的 2026 年 10 月口径讲全：不急、量大、可等的任务进批，全部 token 五折；每批上限 100,000 条或 256 MB，多数 1 小时内完成、上限 24 小时、结果保留 29 天；requests 就是编号的 Messages 请求（custom_id 加 params，后者与 create() 参数完全一致），轮询 processing_status 到 ended，results() 顺序任意、按 custom_id 对账，errored 分请求错与瞬时错两类，expired 重交；五折与提示缓存乘法叠加，Sonnet 批内读取低至 0.10 美元；结构化输出把 output_config 写进 params，收集端自己 model_validate。一句话带走：有人在线等才同步，其余进批五折。

留三个练习：其一，把第 3 篇发票提取器的 20 张发票搬进 Batch，params 里带上 record_invoice 工具定义，对照同步版的耗时与账单；其二，给第 7 节的 SYSTEM 加 cache_control 断点重跑，观察批内首写与后续读取的 usage 分布，验证第 5 节的叠加账；其三，故意删掉某一条的 max_tokens 再交一批，验证它进失败名单而其余 499 条照常成功。对于手头任何"今晚跑、明早看"的任务，建议今天就把它改成 Batch 提交——同一套请求体换个入口，账单直接对半。下一篇讲模型选型与成本核算：把四档模型放进同一张决策表，前 8 篇攒下的价格、缓存与批处理三本账在那里合成一张总账。