Message Batches API

Message Batches API 是 Anthropic 提供的異步 API，讓你一次提交大量請求，系統在後台處理完後讓你下載結果，而不是每個請求即時等待回應。

工作流程：

1. 把多個 API 請求打包成一個 JSONL 文件，每行是一個獨立的請求（JSON 格式，包含 custom_id、model、messages 等）
2. 通過 API 提交這個 JSONL 文件，獲得一個 batch_id
3. 定期輪詢這個 batch 的狀態（in_progress / ended）
4. 狀態變為 ended 後，下載結果文件（JSONL 格式，每行是一個請求的結果）
5. 用 custom_id 對應每個結果和你的原始請求

費用優勢：標準 Messages API 的 50%。沒有最低請求數量要求，一個 batch 可以只有一個請求（雖然這樣做顯然沒有意義）。

時間承諾：Anthropic 保證在 24 小時內完成，但通常更快（幾分鐘到幾小時，取決於當前的系統負載和 batch 大小）。如果你有 SLA 要求（如「必須在 2 小時內完成」），用標準即時 API 更可靠。

支援的模型：目前 Batch API 支援主流的 Claude 模型，包括 Fable 5、Opus 4.8、Sonnet 4.6、Haiku 4.5 等。

Batch API 和 Prompt Caching 怎麼搭配使用，實際費用能省多少？

這是費用優化空間最大的組合：

Batch API 的省錢邏輯：把標準 API 費用直接砍半。原本 $15/M 的 Sonnet 4.6 輸出，用 Batch API 變成 $7.5/M。

Prompt Caching 的省錢邏輯：對固定的 System Prompt（超過 1,024 Token），快取命中時輸入費用降低 90%。假設你的 System Prompt 是 3,000 Token，每次呼叫的合約文本是 5,000 Token：

• 不用快取：輸入計費是 8,000 Token
• 用快取（命中率 85%）：實際輸入計費 ≈ 3,000 × 10% × 85% + 3,000 × 1 × 15% + 5,000 = 255 + 450 + 5,000 = 5,705 Token

疊加效果計算（Sonnet 4.6 為例）：

標準即時 API：$3/M 輸入 + $15/M 輸出 Batch API：$1.5/M 輸入 + $7.5/M 輸出 Batch + Prompt Caching（System Prompt 部分省 90%）：有效輸入費用進一步大幅降低

假設每次請求：System Prompt 3,000 Token + 文件 5,000 Token + 輸出 1,000 Token：

• 標準 API：(8,000 × $3 + 1,000 × $15) / 1,000,000 = $0.039/次
• Batch + 快取（85% 命中）：約 $0.012/次，省了 69%
• 按月 10,000 次計算：標準 API $390/月 → Batch + 快取 $120/月，每月省 $270。

Batch API 有哪些使用限制和注意事項？什麼情況下不適合用？

技術限制：

每個 JSONL 文件最多支援 100,000 個請求或 256MB，超過的需要拆成多個 batch。Batch 的結果文件只保留一段時間（目前約 29 天），需要在保留期內下載，否則可能遺失。如果 batch 裡某些請求失敗，成功的請求仍然計費，失敗的不計費——需要對失敗的請求做重新提交的機制。

特殊功能的支援：部分 beta 功能在 Batch API 裡需要特殊處理。例如 300K 輸出 Token 的擴展輸出功能（Opus 4.8、Opus 4.7、Opus 4.6 和 Sonnet 4.6 支援），需要在 batch 請求裡加入 output-300k-2026-03-24 beta 標頭。

絕對不適合 Batch API 的場景：

任何需要用戶在畫面前等待的互動——聊天介面、即時問答、用戶觸發的任何操作，用標準 API。需要確定 SLA 的任務（如「必須在 10 分鐘內完成」），Batch API 的 24 小時 SLA 完全無法滿足。需要根據前一步的結果決定下一步的串行工作流，Batch API 的異步特性讓它不適合。

最適合 Batch API 的場景特徵：任務和任務之間完全獨立（每個文件的分析和其他文件無關）；任務結果不需要在觸發後幾分鐘內就能看到；任務量大（幾十到幾萬個請求）。

實際上怎麼用 Python 實作 Batch API 的完整流程？

一個完整的 Python 實作範例：

import anthropic
import json
import time

client = anthropic.Anthropic()

# 步驟一：準備 batch 請求（通常從文件或資料庫讀取）
requests = []
for i, document in enumerate(documents_to_analyze):
    requests.append({
        "custom_id": f"doc-{i}",
        "params": {
            "model": "claude-sonnet-4-6",
            "max_tokens": 1024,
            "system": [
                {
                    "type": "text",
                    "text": "你是一個合約分析助手...",
                    "cache_control": {"type": "ephemeral"}  # 啟用 Prompt Caching
                }
            ],
            "messages": [{"role": "user", "content": document}]
        }
    })

# 步驟二：提交 batch
batch = client.messages.batches.create(requests=requests)
batch_id = batch.id
print(f"Batch {batch_id} submitted, {len(requests)} requests")

# 步驟三：輪詢狀態（生產環境建議用 webhook 或排程而不是忙等）
while True:
    status = client.messages.batches.retrieve(batch_id)
    if status.processing_status == "ended":
        break
    print(f"Status: {status.processing_status}, "
          f"succeeded: {status.request_counts.succeeded}")
    time.sleep(60)  # 每分鐘查一次

# 步驟四：下載並處理結果
results = {}
for result in client.messages.batches.results(batch_id):
    if result.result.type == "succeeded":
        results[result.custom_id] = result.result.message.content[0].text
    else:
        print(f"Request {result.custom_id} failed: {result.result.error}")

print(f"Done. {len(results)} successful results.")

注意事項：client.messages.batches.results() 是串流式的，適合結果文件很大的情況，不需要一次把所有結果載入記憶體。生產環境裡，用 webhook 通知（而不是輪詢）讓你的系統在 batch 完成後被通知，而不是一直查詢。