cost-aware-llm-pipeline

成本感知 LLM 管線

在維持品質的同時控制 LLM API 成本的多種模式。將模型路由、預算追蹤、重試邏輯與提示快取組合成可組合的管線。

啟用時機

• 建構呼叫 LLM API（Claude、GPT 等）的應用程式
• 處理複雜度不一的批次項目
• 需要將 API 花費控制在預算內
• 在不犧牲複雜任務品質的前提下最佳化成本

核心概念

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

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

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. 不可變成本追蹤

使用凍結資料類別追蹤累計花費。每次 API 呼叫回傳新的追蹤器 — 絕不改變狀態。

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. 狹義重試邏輯

僅在暫時性錯誤時重試。認證錯誤或錯誤請求則快速失敗。

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)  # 指數退避
    # AuthenticationError, BadRequestError 等 → 立即拋出

4. 提示快取

快取較長的系統提示，避免每次請求都重新傳送。

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

組合使用

將四種技術結合在單一管線函式中：

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）

模型	輸入（$/1M tokens）	輸出（$/1M 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 的應用程式
• 成本快速累積的批次處理管線
• 需要智慧路由的多模型架構
• 需要預算護欄的生產系統