4.5　设计高效的批处理策略

这一条考什么

这一条考你能不能根据工作负载的延迟要求，在同步 Messages API 和 Message Batches API 之间做出选择。你要记住 Batches API 的几个关键特性：费用便宜 50%、最长 24 小时处理窗口、没有延迟 SLA、单个请求内不支持多轮工具调用、用 custom_id 对应请求和结果。你还要会根据 SLA 算出提交频率，会只重提失败的请求，也知道大批量提交之前应该先用样本集把提示词调好。

核心概念

Message Batches API 的基本特性

• 成本：比同步调用便宜 50%，输入和输出 token 都打五折。
• 处理窗口：一个批次最长 24 小时内处理完，大部分批次会更快，但不保证任何延迟，没有 SLA。超过 24 小时还没处理的请求会以 expired 结束。
• 异步：提交后得到批次 ID，之后轮询 processing_status，等到 ended 再取结果。结果的顺序不一定和提交顺序相同。
• 每个请求相互独立：一个请求失败不影响其他请求。

batch = client.messages.batches.create(requests=[
    {"custom_id": "doc-0001",
     "params": {"model": "claude-sonnet-4-5", "max_tokens": 1024,
                "messages": [{"role": "user", "content": "..."}]}},
])

适合与不适合批处理的工作负载

判断标准只有一个：有没有人或流程在同步等这个结果？

• 适合：夜间报表、每周审计、夜间批量生成测试、历史文档回填抽取、离线评测。这些任务晚几个小时出结果都没问题，用批处理能省一半钱。
• 不适合：合并前检查（pre-merge check）这种阻塞流程。开发者在等 CI 结果，最长 24 小时、没有 SLA 的处理方式无法接受。另外实时客服、交互式 Agent 也不适合。

所以同一个团队往往两种都用：PR 阻塞检查走同步 API，夜间的全库安全审计走 Batches API。

批处理不支持请求内的多轮工具调用

一个批次请求就是一次 Messages 调用，不是一个 Agent 循环。如果模型返回了 stop_reason: "tool_use"，你没办法在批处理过程中执行工具、把 tool_result 塞回去让它接着跑。想做多轮，就只能等结果回来，再把 tool_result 拼进下一个批次请求，每一轮都可能要等上几个小时。

因此需要中途调用工具的 Agent 式任务（查数据库、读文件后再决定下一步）不适合用批处理。适合批处理的是"一问一答就能完成"的任务。如果用工具只是为了拿到结构化输出（见 4.3），那没问题，因为不需要把结果回传给模型。

custom_id：对应请求与结果

每个请求都必须带一个在本批次内唯一的 custom_id。结果按 custom_id 返回，每条结果的 result.type 是以下之一：

• succeeded：成功，里面有完整的 message；
• errored：请求出错，比如参数不合法、输入超出上下文窗口；
• canceled：批次被取消；
• expired：24 小时内没处理到。

custom_id 最好直接用业务主键（文档 ID、文件路径的哈希），这样失败时马上就知道是哪份文档出了问题。

根据 SLA 计算提交频率

如果业务要求"文档进来后 30 小时内必须出结果"，而批处理最坏要 24 小时，那留给攒批的时间最多是 30 − 24 = 6 小时。实际可以每 4 小时提交一次：文档最多等 4 小时被打包，再加最多 24 小时处理，最坏 28 小时，留出 2 小时余量给重提和下游处理。

公式：提交间隔 ≤ SLA − 最长处理时间 − 安全余量。

处理失败：只重提失败的那部分

假设 10,000 份文档里有 300 份因为超出上下文窗口而 errored：

• 正确：按 custom_id 找出这 300 份，修复原因（比如把超长文档切块，每块单独一个请求，custom_id 写成 doc-0042#part1），再只把这些放进新批次提交。
• 错误：整批 10,000 份重新提交；或者只调大 max_tokens（max_tokens 管的是输出上限，跟输入超长没关系）。

先用样本集调提示词

大批量提交前，先拿一个有代表性的小样本（比如 50 到 100 份，覆盖各种文档类型）用同步 API 或小批次迭代提示词，直到首次成功率和抽取质量都达标，再提交全量。否则一个提示词缺陷会让上万个请求全部返工，就算打了五折，重跑也很贵，还要再等一个 24 小时窗口。

另外，批次里大量请求共用一段很长的 system prompt 时，可以配合 prompt caching。批处理可能持续超过 5 分钟，文档建议这种情况用 1 小时的缓存 TTL，否则缓存可能在后续请求被处理前就已经过期。

正确做法与反模式

• 正确：阻塞型的合并前检查用同步 API，夜间和每周分析用 Batches API。反模式：为了省 50% 把 PR 阻塞检查也改成批处理。
• 正确：需要多轮工具调用的 Agent 任务用同步 API 跑循环。反模式：期望在一个批次请求里完成"调工具 → 回传结果 → 继续推理"。
• 正确：用 custom_id 定位失败项，修复原因（比如切块）后只重提这些。反模式：整批重提，或者不改任何东西原样重提。
• 正确：根据 SLA 减去 24 小时再减去余量来定提交间隔。反模式：每天提交一次，却承诺 30 小时 SLA（最坏 24 + 24 = 48 小时）。
• 正确：先在样本集上调好提示词再提交全量。反模式：直接提交 10 万条，发现问题后再全部重跑。

常见陷阱

• "批处理便宜，所以 CI 阻塞检查也用它"：没有延迟 SLA，最长 24 小时。只要题干里有人在等、需要在合并前给出结果，就选同步 API。
• "批处理能保证 24 小时内返回，所以可以承诺 24 小时 SLA"：24 小时是处理上限，不是端到端保证，还得算上攒批等待时间。SLA 计算题要减去处理窗口。
• "在批次请求里让模型调用工具并继续推理"：不支持。批处理只适合单轮请求。
• "整批重提"或"换更大上下文的模型重跑全部"：浪费钱。正确选项是按 custom_id 只重提失败项，并针对失败原因做修复。
• "调大 max_tokens 解决输入超长"：max_tokens 限制的是输出长度，跟输入超出上下文窗口无关。输入超长应该切块。
• "按提交顺序对应结果"：结果顺序不保证，必须用 custom_id 对应。

实操练习

目标：提交一个小批次，按 custom_id 汇总结果，并只把失败项挑出来重提。

1. 准备 3 份短文档和 1 份故意设置非法参数的请求（比如 max_tokens 设为 0）来模拟失败。
2. 提交批次，轮询直到 processing_status == "ended"。
3. 遍历结果，按 result.type 分类。
4. 只把 errored 的 custom_id 修正后重新提交。

import time
import anthropic

client = anthropic.Anthropic()
docs = {"doc-001": "Q3 收入 120 万美元。", "doc-002": "退货率 4.2%。", "doc-003": "新增客户 37 家。"}

def req(cid, text, max_tokens=256):
    return {"custom_id": cid, "params": {
        "model": "claude-sonnet-4-5", "max_tokens": max_tokens,
        "messages": [{"role": "user", "content": f"用一句话总结：{text}"}]}}

requests = [req(k, v) for k, v in docs.items()]
requests.append(req("doc-004", "这条请求用来模拟失败", max_tokens=0))

batch = client.messages.batches.create(requests=requests)
while True:
    batch = client.messages.batches.retrieve(batch.id)
    print("status:", batch.processing_status, batch.request_counts)
    if batch.processing_status == "ended":
        break
    time.sleep(30)

failed = []
for entry in client.messages.batches.results(batch.id):
    if entry.result.type == "succeeded":
        print(entry.custom_id, entry.result.message.content[0].text)
    else:
        print(entry.custom_id, "FAILED:", entry.result.type)
        failed.append(entry.custom_id)

if failed:
    retry = [req(cid, "这条请求已修正", max_tokens=256) for cid in failed]
    print("只重提:", failed, client.messages.batches.create(requests=retry).id)

预期结果：request_counts 里有 3 条 succeeded、1 条 errored（如果参数校验在提交时就报错，整个创建调用会失败，那就换成用超长输入来模拟）；最后只有 doc-004 被重提。注意观察结果顺序不一定是 001 到 004，这正是要用 custom_id 对应的原因。

自测清单

• Message Batches API 在成本、处理窗口和 SLA 上分别有什么特点？
• 为什么合并前检查不应该用批处理？举出三个适合批处理的工作负载。
• 为什么批处理不能完成需要中途执行工具的多轮任务？
• 业务 SLA 是 30 小时，提交间隔最多能设多长？怎么算？
• 300 份文档因输入超长失败，正确的修复和重提步骤是什么？
• 为什么要先用样本集调提示词，再提交全量批次？