Prompt Caching 與 Batch:成本砍半

Step 1. 啟用 Prompt Caching,觀察 cache write 與 read

在需要快取的區塊最後加上 "cache_control": {"type": "ephemeral"}。以下範例用一份模擬的長合約文件:

import anthropic

client = anthropic.Anthropic()

# 模擬一份超長文件(真實情境可能是讀取 PDF 轉成的文字)
LONG_DOCUMENT = "這是合約條款全文..." + ("內容重複填充..." * 400)

def ask_about_contract(question: str):
    response = client.messages.create(
        model="claude-sonnet-4-6",
        max_tokens=1024,
        system=[
            {
                "type": "text",
                "text": "你是一位合約審閱助理,請根據使用者問題分析以下合約內容。",
            },
            {
                "type": "text",
                "text": LONG_DOCUMENT,
                "cache_control": {"type": "ephemeral"},  # 快取斷點放在這裡
            },
        ],
        messages=[{"role": "user", "content": question}],
    )
    u = response.usage
    print(f"問題: {question[:20]}...")
    print(f"  cache_write: {u.cache_creation_input_tokens}")
    print(f"  cache_read:  {u.cache_read_input_tokens}")
    print(f"  input:       {u.input_tokens}")
    print(f"  output:      {u.output_tokens}")
    return response.content[0].text

# 第一次:cache write(寫入快取,稍貴)
ask_about_contract("這份合約的違約條款在哪裡?")

# 第二次:cache read(命中快取,省 90%)
ask_about_contract("交貨期限是多少天?")

# 第三次:仍然 cache read
ask_about_contract("付款方式是什麼?")

執行後你會看到:第一次 cache_creation_input_tokens > 0、cache_read_input_tokens == 0。第二次起則反過來,cache_read_input_tokens > 0,代表省錢生效了。只要每次問問題的間隔在 5 分鐘以內,快取就會持續命中。

Step 2. 同樣功能的 JavaScript 版本

import Anthropic from "@anthropic-ai/sdk";

const client = new Anthropic();
const LONG_DOCUMENT = "合約條款全文...".padEnd(30000, "內容");

async function askAboutContract(question) {
  const response = await client.messages.create({
    model: "claude-sonnet-4-6",
    max_tokens: 1024,
    system: [
      {
        type: "text",
        text: "你是一位合約審閱助理,請根據使用者問題分析以下合約內容。",
      },
      {
        type: "text",
        text: LONG_DOCUMENT,
        cache_control: { type: "ephemeral" },
      },
    ],
    messages: [{ role: "user", content: question }],
  });

  const u = response.usage;
  console.log(`write: ${u.cache_creation_input_tokens} | read: ${u.cache_read_input_tokens}`);
  return response.content[0].text;
}

await askAboutContract("違約條款在哪裡?");   // cache write
await askAboutContract("交貨期限多少天?");   // cache read ✓

Step 3. 使用 1 小時 TTL(適合慢速批次場景)

如果你的任務間隔超過 5 分鐘,例如批次任務在後台排隊等待 30 分鐘才開始處理,5 分鐘 TTL 的快取可能在批次跑完前就過期了。這時改用 1 小時 TTL:

{
    "type": "text",
    "text": LONG_DOCUMENT,
    "cache_control": {"type": "ephemeral", "ttl": "1h"},  # 1 小時 TTL
}

使用 1 小時 TTL 時,cache write 費用是 Sonnet 4.6 的 $6.00/MTok(基準的 2 倍)。只要在 1 小時內讀超過 2 次就回本。批次任務的每份文件通常對應多個分析問題,很容易超過這個門檻。

Step 4. 送出 Batch 任務

假設你要把 1,000 筆客服評論做情緒分類,不需要即時結果:

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

client = anthropic.Anthropic()

feedbacks = [
    {"id": "fb_001", "text": "商品很快到,包裝也很好,非常滿意!"},
    {"id": "fb_002", "text": "客服沒有回應,等了三天都沒消沒息。"},
    {"id": "fb_003", "text": "品質普通,跟預期差一點點。"},
    # ... 更多筆,實際情況可能有幾千筆
]

SYSTEM_PROMPT = (
    "你是客服分析師。請將以下回饋分類為「正面」、「負面」或「中性」,"
    "只回傳一個詞,不要加其他說明。"
)

# 建立批次 request 清單
requests = [
    Request(
        custom_id=fb["id"],  # 一定要設,取結果時靠它對應
        params=MessageCreateParamsNonStreaming(
            model="claude-haiku-4-5",  # 分類用 Haiku 就夠,更省錢
            max_tokens=10,
            system=SYSTEM_PROMPT,
            messages=[{"role": "user", "content": fb["text"]}],
        ),
    )
    for fb in feedbacks
]

# 送出批次(立刻拿到 batch_id,但結果還沒好)
batch = client.messages.batches.create(requests=requests)
print(f"Batch ID: {batch.id}")
print(f"狀態: {batch.processing_status}")  # in_progress
print(f"請求數量: {batch.request_counts.processing}")

# 重要:把 batch.id 存起來,程式重啟後還能查詢
with open("batch_id.txt", "w") as f:
    f.write(batch.id)

注意:custom_id 是你自己取的識別碼。批次結果的回傳順序不保證和送入時一樣,一定要靠 custom_id 對應,不能靠位置。

Step 5. 輪詢狀態並取回結果

Batch 不是即時的,送出後要等。下面示範簡單的輪詢迴圈,實際生產環境建議改用獨立的 cron job,避免程式 crash 導致結果遺失:

import time

batch_id = open("batch_id.txt").read().strip()

# 持續輪詢直到完成
while True:
    status = client.messages.batches.retrieve(batch_id)
    counts = status.request_counts
    print(
        f"狀態: {status.processing_status} | "
        f"完成: {counts.succeeded} | "
        f"錯誤: {counts.errored} | "
        f"處理中: {counts.processing}"
    )
    if status.processing_status == "ended":
        break
    time.sleep(30)  # 每 30 秒查一次

# 取回每筆結果
results = {}
for entry in client.messages.batches.results(batch_id):
    if entry.result.type == "succeeded":
        label = entry.result.message.content[0].text.strip()
        results[entry.custom_id] = label
    else:
        # 記錄失敗的 request,稍後可以重跑
        results[entry.custom_id] = f"ERROR:{entry.result.error.type}"

for fb_id, label in results.items():
    print(f"{fb_id}: {label}")

Step 6. Batch API JavaScript 版本

import Anthropic from "@anthropic-ai/sdk";
import fs from "fs";

const client = new Anthropic();

async function runFeedbackBatch(feedbacks) {
  const requests = feedbacks.map((fb) => ({
    custom_id: fb.id,
    params: {
      model: "claude-haiku-4-5",
      max_tokens: 10,
      system: "你是客服分析師。請將以下回饋分類為「正面」、「負面」或「中性」,只回傳一個詞。",
      messages: [{ role: "user", content: fb.text }],
    },
  }));

  const batch = await client.messages.batches.create({ requests });
  console.log("Batch ID:", batch.id);
  fs.writeFileSync("batch_id.txt", batch.id);

  // 輪詢
  let current = batch;
  while (current.processing_status !== "ended") {
    await new Promise((r) => setTimeout(r, 30000));
    current = await client.messages.batches.retrieve(batch.id);
    const c = current.request_counts;
    console.log(`完成: ${c.succeeded} | 錯誤: ${c.errored} | 處理中: ${c.processing}`);
  }

  // 取結果
  const results = {};
  for await (const entry of await client.messages.batches.results(batch.id)) {
    results[entry.custom_id] =
      entry.result.type === "succeeded"
        ? entry.result.message.content[0].text.trim()
        : `ERROR:${entry.result.error.type}`;
  }
  return results;
}

Step 7. 終極組合:Batch + Prompt Caching

如果批次裡每筆 request 都包含同一份長文件,把 cache_control 標上去——Batch API 和 Prompt Caching 完全相容:

REFERENCE_DOC = "產品規格書全文..." * 100  # 模擬 20,000 token 文件

requests = [
    Request(
        custom_id=f"q_{i}",
        params=MessageCreateParamsNonStreaming(
            model="claude-sonnet-4-6",
            max_tokens=256,
            system=[
                {
                    "type": "text",
                    "text": "你是產品評測助理,根據以下規格書回答問題。",
                },
                {
                    "type": "text",
                    "text": REFERENCE_DOC,
                    # Batch 任務用 1h TTL,避免排隊等待期間快取過期
                    "cache_control": {"type": "ephemeral", "ttl": "1h"},
                },
            ],
            messages=[{"role": "user", "content": question}],
        ),
    )
    for i, question in enumerate(questions)
]

batch = client.messages.batches.create(requests=requests)

批次跑完後,查每筆結果的 usage.cache_read_input_tokens——如果大部分都大於 0,代表組合策略生效了。