Claude API怎么批量处理任务？custom_id、异步状态与结果对齐

多份独立资料需要离线处理时，Claude Message Batches可异步执行一组Messages请求。它不是把所有材料塞进一个巨大聊天，也不是实时问答接口。本文用两条人工分类练习说明提交、取回和对齐，不实际创建付费批次。

第一步：先明确独立性与编号

每条任务应自带所需上下文，不依赖另一条先返回的答案。为每个输入设置稳定且唯一的custom_id，保存编号到原资料的映射；不能只用数组位置对齐，因为结果顺序不保证等于提交顺序。

先用小样本核对提示、模型与输出格式，再批量扩大。需要开发者API权限，消费端会员不等同于API额度；不同平台支持情况按官方文档核对，不默认所有托管入口同样支持。

第二步：提交可追溯的小批次

下面使用两条无敏感练习文字，逐条保留params。不要提交重复编号，尤其重试时先确认旧任务是否已成功，避免重复处理。实际代码应在受控环境读取凭据，不写入文件。

import anthropic
client = anthropic.Anthropic()
requests = []
for key, text in [("demo-a", "希望增加案例"), ("demo-b", "讲解速度太快")]:
    requests.append({"custom_id": key, "params": {
        "model": "claude-opus-5-5", "max_tokens": 256,
        "messages": [{"role": "user", "content": "将这条练习反馈概括为一个主题：" + text}]
    }})
batch = client.messages.batches.create(requests=requests)
print(batch.id)

第三步：保存批次ID并查询状态

提交得到ID不代表处理完成。保存真实batch.id，随后用batches.retrieve查询processing_status，设置有限查询间隔和超时，不用每次查询都创建新批次。离线任务应允许合理等待，不能承诺固定几分钟必然结束。

官方说明批处理可能持续较长时间，未在24小时内完成的请求可能过期。不要把到达期限简单解释成全部成功，也不要将等待期间没有输出视为没有任务。

status = client.messages.batches.retrieve(batch.id)
print(status.processing_status)
# 仅在批次结束且结果可用后取结果
# for row in client.messages.batches.results(batch.id):
#     print(row.custom_id, row.result.type)

第四步：逐条判断结果

结果按custom_id回填，分别检查succeeded、errored、canceled和expired等状态。只有成功项才读取对应message内容；批次结束只是整体状态，不能用它覆盖所有子任务。结果缺号或重复编号应停下排查。

应用应保留原输入、提示版本、模型、批次ID、逐条状态和输出位置。网络中断后先重新获取既有结果，不直接重新提交全量。输出内容还要进行任务验收，成功状态不等于分类一定准确。

第五步：只补跑必要失败项

检查失败原因：参数问题先修正，取消与过期按业务需要决定是否重试，不把永久错误放进无界重试。补跑建立新批次并保留与原custom_id的关系，避免失去审计链。

练习中两条分别应与案例需求、讲解节奏相关，人工核对主题是否有原文支持。真实大批任务还要抽查成功项与未分类项。验收应包含提交数、成功数、各失败类别、未对齐数和补跑记录，不能只展示一个批次完成截图。

官方资料

资料核对日期：2026年11月14日。

• Anthropic：Batch processing