---
name: cost-aware-llm-pipeline
description: LLM API 使用成本優化模式 — 根據任務複雜度進行模型路由、預算追蹤、重試邏輯以及提示詞快取 (Prompt Caching)。
---

# 具備成本意識的 LLM 流水線 (Cost-Aware LLM Pipeline)

在維持品質的同時控制 LLM API 成本的模式。將模型路由、預算追蹤、重試邏輯與提示詞快取整合為一個可組合的流水線。

## 何時啟用

- 建置呼叫 LLM API (Claude, GPT 等) 的應用程式。
- 處理複雜度各異的批次項目。
- 需要將 API 支出控制在預算範圍內。
- 在不犧牲複雜任務品質的前提下優化成本。

## 核心概念

### 1. 依任務複雜度進行模型路由

針對簡單任務自動選擇較便宜的模型，將昂貴的模型保留給複雜任務。

```python
MODEL_SONNET = "claude-sonnet-4-6"
MODEL_HAIKU = "claude-haiku-4-5-20251001"

_SONNET_TEXT_THRESHOLD = 10_000  # 字元數門檻
_SONNET_ITEM_THRESHOLD = 30     # 項目數門檻

def select_model(
    text_length: int,
    item_count: int,
    force_model: str | None = None,
) -> str:
    """根據任務複雜度選擇模型。"""
    if force_model is not None:
        return force_model
    if text_length >= _SONNET_TEXT_THRESHOLD or item_count >= _SONNET_ITEM_THRESHOLD:
        return MODEL_SONNET  # 複雜任務
    return MODEL_HAIKU  # 簡單任務 (便宜 3-4 倍)
```

### 2. 不可變的成本追蹤 (Immutable Cost Tracking)

使用凍結資料類別 (Frozen dataclasses) 追蹤累計支出。每次 API 呼叫都會回傳一個新的追蹤器 — 絕對不要修改現有狀態。

```python
from dataclasses import dataclass

@dataclass(frozen=True, slots=True)
class CostRecord:
    model: str
    input_tokens: int
    output_tokens: int
    cost_usd: float

@dataclass(frozen=True, slots=True)
class CostTracker:
    budget_limit: float = 1.00
    records: tuple[CostRecord, ...] = ()

    def add(self, record: CostRecord) -> "CostTracker":
        """回傳包含新記錄的新追蹤器 (絕不修改自身狀態)。"""
        return CostTracker(
            budget_limit=self.budget_limit,
            records=(*self.records, record),
        )

    @property
    def total_cost(self) -> float:
        return sum(r.cost_usd for r in self.records)

    @property
    def over_budget(self) -> bool:
        return self.total_cost > self.budget_limit
```

### 3. 精確的重試邏輯

僅在發生暫時性錯誤時重試。若發生身分驗證或錯誤請求 (Bad Request) 錯誤，則應立即失敗。

```python
from anthropic import (
    APIConnectionError,
    InternalServerError,
    RateLimitError,
)

_RETRYABLE_ERRORS = (APIConnectionError, RateLimitError, InternalServerError)
_MAX_RETRIES = 3

def call_with_retry(func, *, max_retries: int = _MAX_RETRIES):
    """僅針對暫時性錯誤進行重試，其餘錯誤則立即報錯。"""
    for attempt in range(max_retries):
        try:
            return func()
        except _RETRYABLE_ERRORS:
            if attempt == max_retries - 1:
                raise
            time.sleep(2 ** attempt)  # 指數退避 (Exponential backoff)
    # 身分驗證錯誤 (AuthenticationError)、錯誤請求 (BadRequestError) 等 → 立即抛出異常
```

### 4. 提示詞快取 (Prompt Caching)

快取長型的系統提示詞，避免在每次請求時重複發送。

```python
messages = [
    {
        "role": "user",
        "content": [
            {
                "type": "text",
                "text": system_prompt,
                "cache_control": {"type": "ephemeral"},  # 快取此部分
            },
            {
                "type": "text",
                "text": user_input,  # 變動部分
            },
        ],
    }
]
```

## 組合運作

將上述四種技術整合至單一流水線函式中：

```python
def process(text: str, config: Config, tracker: CostTracker) -> tuple[Result, CostTracker]:
    # 1. 模型路由
    model = select_model(len(text), estimated_items, config.force_model)

    # 2. 檢查預算
    if tracker.over_budget:
        raise BudgetExceededError(tracker.total_cost, tracker.budget_limit)

    # 3. 呼叫並採用重試與快取機制
    response = call_with_retry(lambda: client.messages.create(
        model=model,
        messages=build_cached_messages(system_prompt, text),
    ))

    # 4. 追蹤成本 (不可變狀態)
    record = CostRecord(model=model, input_tokens=..., output_tokens=..., cost_usd=...)
    tracker = tracker.add(record)

    return parse_result(response), tracker
```

## 價格參考 (2025-2026)

| 模型 | 輸入 (USD/每百萬 tokens) | 輸出 (USD/每百萬 tokens) | 相對成本 |
|-------|---------------------|----------------------|---------------|
| Haiku 4.5 | $0.80 | $4.00 | 1x |
| Sonnet 4.6 | $3.00 | $15.00 | ~4x |
| Opus 4.5 | $15.00 | $75.00 | ~19x |

## 最佳實踐

- **從最便宜的模型開始**，僅在達到複雜度門檻時才路由至昂貴的模型。
- 在處理批次任務前 **設定明確的預算限制** — 寧可提早失敗也不要超額支出。
- **記錄模型選擇的決策**，以便您可以根據實際數據調整門檻。
- 針對超過 1024 tokens 的系統提示詞 **使用提示詞快取** — 可節省成本並降低延遲。
- **絕對不要在身分驗證或驗證錯誤時重試** — 僅針對暫時性故障（網路、速率限制、伺服器錯誤）執行重試。

## 應避免的反模式

- 不論複雜度如何，對所有請求都使用最昂貴的模型。
- 在發生所有錯誤時都重試（會在永久性失敗上浪費預算）。
- 修改成本追蹤狀態（會導致偵錯與稽核變得困難）。
- 在程式碼庫中硬編碼模型名稱（應使用常數或配置檔）。
- 忽視重複性系統提示詞的快取機制。

## 何時使用

- 任何呼叫 Claude、OpenAI 或類似 LLM API 的應用程式。
- 成本會迅速累積的批次處理流水線。
- 需要智慧路由的多模型架構。
- 需要預算防護欄 (Guardrails) 的生產系統。