[ PROMPT_NODE_23834 ]
batches
[ SKILL_DOCUMENTATION ]
# Message Batches API — Python
Batches API (`POST /v1/messages/batches`) 以标准价格的 50% 异步处理 Messages API 请求。
## 关键事实
- 每个批次最多 100,000 个请求或 256 MB
- 大多数批次在 1 小时内完成；最长 24 小时
- 结果在创建后 29 天内可用
- 所有 Token 使用量享 50% 费用减免
- 支持所有 Messages API 功能（视觉、工具、缓存等）
---
## 创建批次
python
import anthropic
from anthropic.types.message_create_params import MessageCreateParamsNonStreaming
from anthropic.types.messages.batch_create_params import Request
client = anthropic.Anthropic()
message_batch = client.messages.batches.create(
requests=[
Request(
custom_id="request-1",
params=MessageCreateParamsNonStreaming(
model="claude-opus-4-7",
max_tokens=16000,
messages=[{"role": "user", "content": "Summarize climate change impacts"}]
)
),
Request(
custom_id="request-2",
params=MessageCreateParamsNonStreaming(
model="claude-opus-4-7",
max_tokens=16000,
messages=[{"role": "user", "content": "Explain quantum computing basics"}]
)
),
]
)
print(f"Batch ID: {message_batch.id}")
print(f"Status: {message_batch.processing_status}")
---
## 轮询完成状态
python
import time
while True:
batch = client.messages.batches.retrieve(message_batch.id)
if batch.processing_status == "ended":
break
print(f"Status: {batch.processing_status}, processing: {batch.request_counts.processing}")
time.sleep(60)
print("Batch complete!")
print(f"Succeeded: {batch.request_counts.succeeded}")
print(f"Errored: {batch.request_counts.errored}")
---
## 获取结果
> **注意：** 以下示例使用 `match/case` 语法，需要 Python 3.10+。对于早期版本，请改用 `if/elif` 链。
python
for result in client.messages.batches.results(message_batch.id):
match result.result.type:
case "succeeded":
msg = result.result.message
text = next((b.text for b in msg.content if b.type == "text"), "")
print(f"[{result.custom_id}] {text[:100]}")
case "errored":
if result.result.error.type == "invalid_request":
print(f"[{result.custom_id}] Validation error - fix request and retry")
else:
粤公网安备44030002003366号