5.9 KiB
5.9 KiB
| name | description |
|---|---|
| cost-aware-llm-pipeline | LLM API 使用成本優化模式 — 根據任務複雜度進行模型路由、預算追蹤、重試邏輯以及提示詞快取 (Prompt Caching)。 |
具備成本意識的 LLM 流水線 (Cost-Aware LLM Pipeline)
在維持品質的同時控制 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. 不可變的成本追蹤 (Immutable Cost Tracking)
使用凍結資料類別 (Frozen dataclasses) 追蹤累計支出。每次 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. 精確的重試邏輯
僅在發生暫時性錯誤時重試。若發生身分驗證或錯誤請求 (Bad Request) 錯誤,則應立即失敗。
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)
快取長型的系統提示詞,避免在每次請求時重複發送。
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)
| 模型 | 輸入 (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) 的生產系統。