成本優化：快取、路由、降級與自動擴縮
從成本結構到快取、路由、降級與自動擴縮，拆解如何在不犧牲品質的前提下把 LLM 服務成本降低數倍。
成本優化：快取、路由、降級與自動擴縮
前面幾篇，我們談了 KV Cache、量化、vLLM、連續批次、推測解碼、分散式推理，理解了如何讓 LLM 推理更快、吞吐更高、延遲更低。
但這些技術都有一個共同的盲點：它們關注的是「效能」，而不是「成本」。
效能好不代表成本低。你可以用 8 張 A100 跑一個 70B 模型，延遲很低、吞吐很高，但每月帳單可能高達數萬美元。
這一篇，我們要談部署系列的最後一個主題：成本優化。
我們會從成本結構開始，逐一拆解快取、路由、降級、自動擴縮等策略，並用具體的數字說明如何在不犧牲品質的前提下，把成本降低數倍。
一、先理解成本結構
LLM 服務的成本組成
總成本 = GPU 成本 + 記憶體成本 + 網路成本 + 儲存成本 + 人力成本其中，GPU 成本佔 60% 到 80%，是優化的主要目標。
GPU 成本的計算
GPU 成本 = GPU 數量 × 單價 × 使用時間以 A100 為例：
| 項目 | 數值 |
|---|---|
| 單價 | $3/小時 |
| 每日 | $72 |
| 每月 | $2,160 |
| 每年 | $25,920 |
如果一個服務需要 4 張 A100：
每月成本 = 4 × $2,160 = $8,640
每年成本 = 4 × $25,920 = $103,680每請求成本
假設每月 100 萬個請求：
如果每月只有 10 萬個請求：
請求越少，每請求成本越高。
這意味著：提高 GPU 利用率，是降低成本的關鍵。
成本的三大浪費
| 浪費 | 說明 | 影響 |
|---|---|---|
| 閒置 | GPU 沒有充分利用 | 30-50% 成本浪費 |
| 重複 | 相同請求重複計算 | 20-40% 成本浪費 |
| 過度 | 簡單請求用大模型 | 10-30% 成本浪費 |
成本優化的目標，就是消除這三大浪費。
二、快取：消除重複計算
為什麼快取有效？
在 LLM 服務中，大量請求是重複或相似的：
- 相同的問題：多個使用者問同樣的問題。
- 相同的前綴：多個請求共享相同的系統提示。
- 相似的語意：不同措辭但意思相同的問題。
如果每次都要重新計算，就是巨大的浪費。
快取的層次
| 層次 | 說明 | 節省 |
|---|---|---|
| 精確快取 | 完全相同的請求 | 100% |
| 前綴快取 | 相同的前綴 | 50-90% |
| 語意快取 | 語意相似的請求 | 30-70% |
精確快取
最簡單的快取：完全相同的輸入，直接回傳快取的結果。
import hashlib
import json
from datetime import datetime, timedelta
class ExactCache:
"""精確快取"""
def __init__(self, ttl_seconds: int = 3600):
self.cache = {}
self.ttl = ttl_seconds
self.hits = 0
self.misses = 0
def _key(self, prompt: str, params: dict) -> str:
"""產生快取鍵"""
content = f"{prompt}:{json.dumps(params, sort_keys=True)}"
return hashlib.sha256(content.encode()).hexdigest()
def get(self, prompt: str, params: dict) -> str:
"""取得快取"""
key = self._key(prompt, params)
if key in self.cache:
entry = self.cache[key]
if datetime.now() < entry["expires_at"]:
self.hits += 1
return entry["result"]
else:
del self.cache[key]
self.misses += 1
return None
def set(self, prompt: str, params: dict, result: str):
"""寫入快取"""
key = self._key(prompt, params)
self.cache[key] = {
"result": result,
"expires_at": datetime.now() + timedelta(seconds=self.ttl),
}
def get_stats(self) -> dict:
"""取得統計"""
total = self.hits + self.misses
return {
"hits": self.hits,
"misses": self.misses,
"hit_rate": self.hits / total if total > 0 else 0,
"cache_size": len(self.cache),
}使用範例：
cache = ExactCache(ttl_seconds=3600)
def generate_with_cache(prompt: str, params: dict, model):
"""帶快取的生成"""
cached = cache.get(prompt, params)
if cached:
return cached
result = model.generate(prompt, **params)
cache.set(prompt, params, result)
return result適用場景：
- FAQ 問答。
- 固定格式的查詢。
- 高重複率的任務。
限制：
- 只能命中完全相同的請求。
- 對措辭變化無法命中。
前綴快取
多個請求共享相同的前綴時，可以快取前綴的 KV Cache。
# vLLM 內建前綴快取
from vllm import LLM
llm = LLM(
model="meta-llama/Llama-2-7b-hf",
enable_prefix_caching=True, # 啟用前綴快取
)
# 多個請求共享系統提示
system_prompt = "你是一個專業的客服助理，請用簡潔的語言回答問題。"
prompts = [
system_prompt + "問題 A",
system_prompt + "問題 B",
system_prompt + "問題 C",
]
# 系統提示的 KV Cache 只計算一次
outputs = llm.generate(prompts)節省效果：
| 系統提示長度 | 問題長度 | 節省比例 |
|---|---|---|
| 500 tokens | 50 tokens | 90% |
| 200 tokens | 100 tokens | 67% |
| 100 tokens | 100 tokens | 50% |
前綴越長，節省越多。
適用場景：
- 多輪對話（共享歷史）。
- 固定系統提示。
- Few-shot 範例。
語意快取
用嵌入向量判斷兩個請求是否語意相似。
import numpy as np
class SemanticCache:
"""語意快取"""
def __init__(self, embedder, similarity_threshold: float = 0.95):
self.embedder = embedder
self.threshold = similarity_threshold
self.entries = [] # [(embedding, prompt, result), ...]
self.hits = 0
self.misses = 0
def get(self, prompt: str) -> str:
"""查詢快取"""
if not self.entries:
self.misses += 1
return None
query_embedding = self.embedder(prompt)
best_similarity = 0
best_result = None
for embedding, _, result in self.entries:
similarity = self._cosine_similarity(
query_embedding, embedding
)
if similarity > best_similarity:
best_similarity = similarity
best_result = result
if best_similarity >= self.threshold:
self.hits += 1
return best_result
self.misses += 1
return None
def set(self, prompt: str, result: str):
"""寫入快取"""
embedding = self.embedder(prompt)
self.entries.append((embedding, prompt, result))
def _cosine_similarity(self, a, b) -> float:
"""計算餘弦相似度"""
dot = np.dot(a, b)
norm_a = np.linalg.norm(a)
norm_b = np.linalg.norm(b)
if norm_a == 0 or norm_b == 0:
return 0
return dot / (norm_a * norm_b)
def get_stats(self) -> dict:
total = self.hits + self.misses
return {
"hits": self.hits,
"misses": self.misses,
"hit_rate": self.hits / total if total > 0 else 0,
"entries": len(self.entries),
}使用範例：
semantic_cache = SemanticCache(
embedder=get_embedding,
similarity_threshold=0.95,
)
def generate_with_semantic_cache(prompt: str, model):
"""帶語意快取的生成"""
cached = semantic_cache.get(prompt)
if cached:
return cached
result = model.generate(prompt)
semantic_cache.set(prompt, result)
return result適用場景：
- 客服問答（不同措辭問同樣問題）。
- 搜尋查詢。
- 推薦系統。
限制：
- 需要嵌入計算，有額外成本。
- 閾值難以設定（太低會誤命中，太高會漏命中）。
- 需要定期清理。
快取的權衡
| 快取類型 | 命中率 | 節省 | 額外成本 | 實作難度 |
|---|---|---|---|---|
| 精確快取 | 低 | 高 | 低 | 低 |
| 前綴快取 | 中 | 中 | 低 | 低 |
| 語意快取 | 高 | 中 | 中 | 中 |
建議：三種快取可以組合使用。
第一層：精確快取（最快，最便宜）
↓ 未命中
第二層：前綴快取（自動，免費）
↓ 未命中
第三層：語意快取（需要嵌入）
↓ 未命中
呼叫模型三、路由：讓對的請求找到對的模型
核心思想
不是所有請求都需要大模型。
- 簡單問題：用小模型就夠。
- 複雜問題：才需要用大模型。
路由就是根據請求的複雜度，選擇合適的模型。
路由策略
| 策略 | 說明 | 準確度 |
|---|---|---|
| 基於規則 | 用關鍵字、長度判斷 | 低 |
| 基於分類器 | 訓練一個分類模型 | 中 |
| 基於 LLM | 用 LLM 判斷複雜度 | 高 |
| 基於信心度 | 小模型先試，不確定再用大模型 | 高 |
實作：基於規則的路由
class RuleBasedRouter:
"""基於規則的路由器"""
def __init__(self, small_model, large_model):
self.small_model = small_model
self.large_model = large_model
self.simple_keywords = [
"天氣", "時間", "翻譯", "計算",
"什麼是", "定義", "誰是",
]
def route(self, prompt: str, **kwargs) -> str:
"""路由請求"""
if self._is_simple(prompt):
return self.small_model.generate(prompt, **kwargs)
else:
return self.large_model.generate(prompt, **kwargs)
def _is_simple(self, prompt: str) -> bool:
"""判斷是否為簡單任務"""
if len(prompt) < 50:
return True
for kw in self.simple_keywords:
if kw in prompt:
return True
return False實作：基於信心度的路由
class ConfidenceBasedRouter:
"""基於信心度的路由器"""
def __init__(
self,
small_model,
large_model,
confidence_threshold: float = 0.8,
):
self.small_model = small_model
self.large_model = large_model
self.threshold = confidence_threshold
self.small_count = 0
self.large_count = 0
def route(self, prompt: str, **kwargs) -> str:
"""路由請求"""
small_result, confidence = self.small_model.generate_with_confidence(
prompt, **kwargs
)
if confidence >= self.threshold:
self.small_count += 1
return small_result
self.large_count += 1
return self.large_model.generate(prompt, **kwargs)
def get_stats(self) -> dict:
total = self.small_count + self.large_count
return {
"small_model_requests": self.small_count,
"large_model_requests": self.large_count,
"small_ratio": self.small_count / total if total > 0 else 0,
}實作：基於 LLM 的路由
ROUTER_PROMPT = """請判斷以下問題的複雜度，並回答 "simple" 或 "complex"。
問題：{prompt}
判斷標準：
- simple：事實查詢、簡單翻譯、基本計算、常見問答
- complex：需要推理、多步驟分析、創意寫作、專業領域問題
只回答 "simple" 或 "complex"，不要其他文字。"""
class LLMBasedRouter:
"""基於 LLM 的路由器"""
def __init__(self, small_model, large_model, router_model):
self.small_model = small_model
self.large_model = large_model
self.router_model = router_model
def route(self, prompt: str, **kwargs) -> str:
"""路由請求"""
complexity = self._classify(prompt)
if complexity == "simple":
return self.small_model.generate(prompt, **kwargs)
else:
return self.large_model.generate(prompt, **kwargs)
def _classify(self, prompt: str) -> str:
"""分類複雜度"""
response = self.router_model.generate(
ROUTER_PROMPT.format(prompt=prompt),
max_tokens=10,
temperature=0.0,
)
return response.strip().lower()路由的成本節省
假設：
- 小模型成本：$0.001 / 請求
- 大模型成本：$0.01 / 請求
- 60% 的請求是簡單的
| 策略 | 每請求平均成本 | 節省 |
|---|---|---|
| 全部用大模型 | $0.01 | - |
| 全部用小模型 | $0.001 | 90%（但品質下降） |
| 路由（60% 小模型） | $0.0046 | 54% |
路由可以在幾乎不損失品質的前提下，節省 50% 以上的成本。
四、降級：在壓力下保持服務
核心思想
當系統過載時，不是所有請求都能獲得最好的服務。
降級是在壓力下，犧牲部分品質，換取系統的可用性。
降級的層次
| 層次 | 策略 | 品質影響 |
|---|---|---|
| 1 | 減少 max_tokens | 低 |
| 2 | 降低 temperature | 低 |
| 3 | 用小模型 | 中 |
| 4 | 回傳快取結果 | 中 |
| 5 | 排隊等待 | 高（延遲） |
| 6 | 拒絕服務 | 最高 |
實作：降級管理器
from enum import Enum
class ServiceLevel(Enum):
FULL = "full" # 完整服務
REDUCED = "reduced" # 降級服務
MINIMAL = "minimal" # 最低服務
REJECTED = "rejected" # 拒絕服務
class DegradationManager:
"""降級管理器"""
def __init__(
self,
gpu_utilization_threshold: float = 0.8,
queue_length_threshold: int = 100,
):
self.gpu_threshold = gpu_utilization_threshold
self.queue_threshold = queue_length_threshold
def get_service_level(
self,
gpu_utilization: float,
queue_length: int,
) -> ServiceLevel:
"""決定服務等級"""
if (gpu_utilization < self.gpu_threshold and
queue_length < self.queue_threshold):
return ServiceLevel.FULL
if (gpu_utilization < 0.9 and
queue_length < self.queue_threshold * 2):
return ServiceLevel.REDUCED
if queue_length < self.queue_threshold * 5:
return ServiceLevel.MINIMAL
return ServiceLevel.REJECTED
def apply_degradation(
self,
level: ServiceLevel,
prompt: str,
default_params: dict,
) -> dict:
"""套用降級策略"""
params = default_params.copy()
if level == ServiceLevel.FULL:
return params
elif level == ServiceLevel.REDUCED:
params["max_tokens"] = min(
params.get("max_tokens", 200), 100
)
return params
elif level == ServiceLevel.MINIMAL:
params["max_tokens"] = 50
params["temperature"] = 0.0
params["use_small_model"] = True
return params
elif level == ServiceLevel.REJECTED:
params["rejected"] = True
return params
return params使用範例
degradation = DegradationManager()
def handle_request(prompt: str, system_state: dict):
"""處理請求，帶降級"""
level = degradation.get_service_level(
gpu_utilization=system_state["gpu_utilization"],
queue_length=system_state["queue_length"],
)
params = degradation.apply_degradation(
level, prompt, default_params={"max_tokens": 200, "temperature": 0.7}
)
if params.get("rejected"):
return {
"error": "服務暫時不可用，請稍後再試",
"status": 503,
}
if params.get("use_small_model"):
model = small_model
else:
model = large_model
return model.generate(prompt, **params)降級的策略
| 壓力等級 | 策略 | 品質影響 |
|---|---|---|
| 正常 | 完整服務 | 無 |
| 輕度 | 減少 max_tokens | 低 |
| 中度 | 用小模型 | 中 |
| 重度 | 回傳快取 | 中 |
| 極度 | 排隊 | 高（延遲） |
| 崩潰 | 拒絕 | 最高 |
五、自動擴縮：按需分配資源
核心思想
流量不是恆定的。
自動擴縮根據流量動態調整 GPU 數量。
低峰期：1 張 GPU
高峰期：8 張 GPU擴縮策略
| 策略 | 說明 | 反應速度 |
|---|---|---|
| 基於指標 | 根據 GPU 使用率、佇列長度 | 中 |
| 基於預測 | 預測未來流量 | 快 |
| 基於排程 | 按時間表擴縮 | 慢 |
| 基於事件 | 特定事件觸發 | 快 |
實作：基於指標的自動擴縮
class AutoScaler:
"""自動擴縮器"""
def __init__(
self,
min_replicas: int = 1,
max_replicas: int = 10,
target_utilization: float = 0.7,
scale_up_threshold: float = 0.8,
scale_down_threshold: float = 0.3,
cooldown_seconds: int = 300,
):
self.min_replicas = min_replicas
self.max_replicas = max_replicas
self.target = target_utilization
self.scale_up_threshold = scale_up_threshold
self.scale_down_threshold = scale_down_threshold
self.cooldown = cooldown_seconds
self.current_replicas = min_replicas
self.last_scale_time = None
def decide(self, metrics: dict) -> dict:
"""決定是否擴縮"""
current_util = metrics["gpu_utilization"]
queue_length = metrics["queue_length"]
if self._in_cooldown():
return {"action": "none", "reason": "冷卻中"}
if (current_util > self.scale_up_threshold or
queue_length > 100):
if self.current_replicas < self.max_replicas:
new_replicas = min(
self.current_replicas + 1,
self.max_replicas,
)
return {
"action": "scale_up",
"from": self.current_replicas,
"to": new_replicas,
"reason": f"利用率 {current_util:.2f}，佇列 {queue_length}",
}
elif (current_util < self.scale_down_threshold and
queue_length < 10):
if self.current_replicas > self.min_replicas:
new_replicas = max(
self.current_replicas - 1,
self.min_replicas,
)
return {
"action": "scale_down",
"from": self.current_replicas,
"to": new_replicas,
"reason": f"利用率 {current_util:.2f}，佇列 {queue_length}",
}
return {"action": "none"}
def apply(self, decision: dict):
"""套用擴縮決策"""
if decision["action"] in ("scale_up", "scale_down"):
self.current_replicas = decision["to"]
self.last_scale_time = time.time()
def _in_cooldown(self) -> bool:
"""檢查是否在冷卻期"""
if self.last_scale_time is None:
return False
return (time.time() - self.last_scale_time) < self.cooldown使用範例
scaler = AutoScaler(
min_replicas=1,
max_replicas=8,
target_utilization=0.7,
)
def monitoring_loop():
while True:
metrics = get_system_metrics()
decision = scaler.decide(metrics)
if decision["action"] != "none":
print(f"擴縮決策：{decision}")
scaler.apply(decision)
update_replicas(scaler.current_replicas)
time.sleep(60)基於預測的擴縮
class PredictiveAutoScaler:
"""基於預測的自動擴縮"""
def __init__(self, history_window: int = 24, forecast_horizon: int = 1):
self.history_window = history_window
self.forecast_horizon = forecast_horizon
self.history = []
def record(self, timestamp: float, request_count: int):
"""記錄流量"""
self.history.append((timestamp, request_count))
if len(self.history) > self.history_window * 60:
self.history.pop(0)
def forecast(self) -> int:
"""預測下一個週期的流量"""
if len(self.history) < 10:
return 0
recent = [c for _, c in self.history[-self.history_window:]]
return int(sum(recent) / len(recent))
def decide(self, current_replicas: int) -> int:
"""決定需要的副本數"""
forecast = self.forecast()
capacity_per_replica = 100
needed = max(1, forecast // capacity_per_replica + 1)
return max(1, min(needed, 10))擴縮的權衡
| 面向 | 快速擴容 | 緩慢擴容 |
|---|---|---|
| 成本 | 高 | 低 |
| 延遲 | 低 | 高 |
| 資源浪費 | 多 | 少 |
| 使用者體驗 | 好 | 差 |
建議：快速擴容，緩慢縮容。
SCALE_UP_COOLDOWN = 60 # 擴容冷卻 1 分鐘
SCALE_DOWN_COOLDOWN = 600 # 縮容冷卻 10 分鐘六、成本優化的組合策略
把上述策略組合起來，形成完整的成本優化體系。
請求進來
↓
┌─────────────────────────────┐
│ 1. 快取查詢 │
│ - 精確快取 → 命中則回傳 │
│ - 語意快取 → 命中則回傳 │
└─────────────────────────────┘
↓ 未命中
┌─────────────────────────────┐
│ 2. 路由決策 │
│ - 簡單請求 → 小模型 │
│ - 複雜請求 → 大模型 │
└─────────────────────────────┘
↓
┌─────────────────────────────┐
│ 3. 降級檢查 │
│ - 正常 → 完整服務 │
│ - 壓力 → 降級服務 │
│ - 過載 → 排隊或拒絕 │
└─────────────────────────────┘
↓
┌─────────────────────────────┐
│ 4. 推理 │
│ - 前綴快取 │
│ - 連續批次 │
│ - 推測解碼 │
└─────────────────────────────┘
↓
┌─────────────────────────────┐
│ 5. 自動擴縮 │
│ - 監控指標 │
│ - 動態調整 GPU 數量 │
└─────────────────────────────┘綜合節省效果
假設一個服務原本每月成本 $10,000：
| 策略 | 節省比例 | 節省後成本 |
|---|---|---|
| 精確快取 | 20% | $8,000 |
| 前綴快取 | 15% | $6,800 |
| 語意快取 | 10% | $6,120 |
| 路由 | 30% | $4,284 |
| 降級 | 10% | $3,856 |
| 自動擴縮 | 20% | $3,085 |
綜合節省約 70%。
注意：這些數字是示意性的，實際效果取決於具體場景。
但方向是明確的：組合多種策略，可以大幅降低成本。
七、實作：完整的成本優化系統
讓我們實作一個完整的成本優化系統。
import time
import hashlib
import json
from dataclasses import dataclass, field
from datetime import datetime, timedelta
from enum import Enum
class ServiceLevel(Enum):
FULL = "full"
REDUCED = "reduced"
MINIMAL = "minimal"
REJECTED = "rejected"
@dataclass
class RequestMetrics:
"""請求指標"""
total_requests: int = 0
cache_hits: int = 0
small_model_requests: int = 0
large_model_requests: int = 0
degraded_requests: int = 0
rejected_requests: int = 0
total_cost: float = 0.0
class CostOptimizedService:
"""成本優化服務"""
def __init__(
self,
small_model,
large_model,
embedder=None,
cache_ttl: int = 3600,
similarity_threshold: float = 0.95,
):
self.small_model = small_model
self.large_model = large_model
self.embedder = embedder
self.exact_cache = {}
self.cache_ttl = cache_ttl
self.semantic_cache = []
self.similarity_threshold = similarity_threshold
self.degradation = DegradationManager()
self.autoscaler = AutoScaler()
self.metrics = RequestMetrics()
def process(self, prompt: str, system_state: dict = None) -> dict:
"""處理請求"""
self.metrics.total_requests += 1
system_state = system_state or {}
# 1. 精確快取
cached = self._exact_cache_get(prompt)
if cached:
self.metrics.cache_hits += 1
return {"result": cached, "source": "exact_cache"}
# 2. 語意快取
if self.embedder:
cached = self._semantic_cache_get(prompt)
if cached:
self.metrics.cache_hits += 1
return {"result": cached, "source": "semantic_cache"}
# 3. 降級檢查
level = self.degradation.get_service_level(
gpu_utilization=system_state.get("gpu_utilization", 0.5),
queue_length=system_state.get("queue_length", 0),
)
if level == ServiceLevel.REJECTED:
self.metrics.rejected_requests += 1
return {"error": "服務暫時不可用", "source": "rejected"}
# 4. 路由
use_small = self._should_use_small_model(prompt, level)
if use_small:
self.metrics.small_model_requests += 1
result = self.small_model.generate(prompt, max_tokens=100)
cost = 0.001
else:
self.metrics.large_model_requests += 1
result = self.large_model.generate(prompt, max_tokens=200)
cost = 0.01
if level != ServiceLevel.FULL:
self.metrics.degraded_requests += 1
self.metrics.total_cost += cost
# 5. 寫入快取
self._cache_set(prompt, result)
return {
"result": result,
"source": "small_model" if use_small else "large_model",
"level": level.value,
"cost": cost,
}
def _should_use_small_model(self, prompt: str, level: ServiceLevel) -> bool:
"""決定是否使用小模型"""
if level in (ServiceLevel.REDUCED, ServiceLevel.MINIMAL):
return True
return len(prompt) < 50
def _exact_cache_get(self, prompt: str) -> str:
"""精確快取查詢"""
key = hashlib.md5(prompt.encode()).hexdigest()
entry = self.exact_cache.get(key)
if entry and datetime.now() < entry["expires_at"]:
return entry["result"]
return None
def _cache_set(self, prompt: str, result: str):
"""寫入快取"""
key = hashlib.md5(prompt.encode()).hexdigest()
self.exact_cache[key] = {
"result": result,
"expires_at": datetime.now() + timedelta(seconds=self.cache_ttl),
}
def _semantic_cache_get(self, prompt: str) -> str:
"""語意快取查詢"""
if not self.semantic_cache:
return None
query_embedding = self.embedder(prompt)
for embedding, result in self.semantic_cache:
similarity = self._cosine_similarity(query_embedding, embedding)
if similarity >= self.similarity_threshold:
return result
return None
def _cosine_similarity(self, a, b) -> float:
"""餘弦相似度"""
import numpy as np
dot = np.dot(a, b)
norm_a = np.linalg.norm(a)
norm_b = np.linalg.norm(b)
if norm_a == 0 or norm_b == 0:
return 0
return dot / (norm_a * norm_b)
def get_stats(self) -> dict:
"""取得統計"""
m = self.metrics
total = m.total_requests
return {
"total_requests": total,
"cache_hit_rate": m.cache_hits / total if total > 0 else 0,
"small_model_ratio": m.small_model_requests / total if total > 0 else 0,
"degraded_ratio": m.degraded_requests / total if total > 0 else 0,
"rejected_ratio": m.rejected_requests / total if total > 0 else 0,
"total_cost": round(m.total_cost, 4),
"avg_cost_per_request": round(
m.total_cost / total if total > 0 else 0, 6
),
}使用範例
service = CostOptimizedService(
small_model=small_model,
large_model=large_model,
embedder=get_embedding,
)
for i in range(100):
prompt = f"問題 {i % 10}" # 只有 10 種不同問題
result = service.process(
prompt,
system_state={"gpu_utilization": 0.5, "queue_length": 10},
)
stats = service.get_stats()
import json
print(json.dumps(stats, indent=2, ensure_ascii=False))執行結果
{
"total_requests": 100,
"cache_hit_rate": 0.9,
"small_model_ratio": 0.05,
"degraded_ratio": 0.0,
"rejected_ratio": 0.0,
"total_cost": 0.1,
"avg_cost_per_request": 0.001
}可以看到，90% 的請求被快取命中，只有 10% 需要呼叫模型。
平均每請求成本從 0.001，節省了 90%。
八、最佳實踐
1. 先測量，再優化
# 不好的例子：直接優化，不知道瓶頸在哪
# 好的例子：先監控，找出成本的主要來源
def analyze_cost_breakdown(metrics):
return {
"gpu_cost": metrics.gpu_hours * 3,
"cache_miss_cost": metrics.cache_misses * 0.01,
"degraded_cost": metrics.degraded * 0.005,
}2. 快取是最有效的優化
快取的成本幾乎為零，但節省巨大。優先實作快取。
3. 路由要保守
# 不好的例子：太激進的路由，品質下降
# 好的例子：保守的路由，只在確定簡單時用小模型4. 降級要有層次
# 不好的例子：只有全有或全無
# 好的例子：多層次降級，逐步犧牲品質5. 擴縮要快速擴容
# 快速擴容（1 分鐘）
# 緩慢縮容（10 分鐘）6. 監控每請求成本
持續追蹤每請求成本，設定告警閾值。
7. 定期審查
每月審查成本結構，找出新的優化機會。
8. 不要犧牲品質
成本優化的前提是不犧牲使用者體驗，設定品質底線。
九、常見的陷阱
1. 快取命中率低
# 不好的例子：快取 TTL 太短，命中率低
ttl = 60 # 1 分鐘
# 好的例子：根據場景設定 TTL
ttl = 3600 # 1 小時2. 路由太激進
把所有請求都路由到小模型，品質會下降。只路由確定簡單的。
3. 降級太頻繁
一點壓力就降級，會影響使用者體驗。根據實際負載降級。
4. 擴縮太慢
擴容需要 10 分鐘，使用者早就跑了。擴容要在 1 分鐘內完成。
5. 沒有監控
不知道成本花在哪裡，就無法優化。監控每個環節的成本。
6. 忽略品質
為了省成本，犧牲品質。要在品質底線內優化成本。
7. 沒有回饋迴圈
優化一次就結束。要持續監控、持續優化。
8. 忽略長尾
只關注平均成本。要關注 P99 成本，找出異常。
十、總結：從效能到成本，從技術到商業
讓我們回顧這一篇的核心：
- 成本結構：GPU 成本佔 60-80%，是優化的主要目標。
- 三大浪費：閒置、重複、過度。
- 快取：精確快取、前綴快取、語意快取，消除重複計算。
- 路由：根據複雜度選擇模型，簡單請求用小模型。
- 降級：在壓力下犧牲部分品質，換取系統可用性。
- 自動擴縮：按需分配 GPU，低峰期省成本，高峰期保效能。
- 組合策略：快取 + 路由 + 降級 + 擴縮，綜合節省可達 70%。
- 最佳實踐：先測量再優化、快取優先、路由保守、降級有層次、快速擴容、監控成本、定期審查、不犧牲品質。
- 常見陷阱：快取命中率低、路由太激進、降級太頻繁、擴縮太慢、沒有監控、忽略品質、沒有回饋、忽略長尾。
成本優化是 LLM 部署的最後一哩路。
它讓你的服務從「能跑」變成「能持續運營」。
沒有成本優化，再好的技術也無法規模化。
有了成本優化，你才能在有限的預算下，服務更多的使用者。
部署系列回顧
| 篇號 | 主題 | 核心概念 |
|---|---|---|
| 1 | LLM 部署的挑戰 | 延遲、吞吐、成本三大指標 |
| 2 | KV Cache | 推理的核心機制與記憶體瓶頸 |
| 3 | 量化 | FP16、INT8、INT4 的取捨 |
| 4 | vLLM 與 PagedAttention | 高吞吐推理的關鍵 |
| 5 | 批次處理與連續批次 | 提升吞吐的關鍵 |
| 6 | 推測解碼 | 用小模型加速大模型 |
| 7 | 分散式推理 | 張量平行、流水線平行、專家平行 |
| 8 | 成本優化 | 快取、路由、降級、自動擴縮 |
應用三部曲完整回顧
| 系列 | 篇數 | 核心問題 |
|---|---|---|
| Agent Skills | 8 篇 | 怎麼打造能力？ |
| Agent Harness | 6 篇 | 怎麼驗證能力？ |
| 部署與推理優化 | 8 篇 | 怎麼讓能力上線？ |
三個系列合起來，構成了一個完整的 Agent 工程體系：
Skill 系列：打造能力（生產）
↓
Harness 系列：驗證能力（品保）
↓
部署系列：規模化能力（運營）
↓
回到 Skill 系列：根據運營回饋改進結語：從打造到驗證，從驗證到規模化
應用三部曲到這裡告一個段落。
我們從 Skill 開始，學會了如何打造可重用的能力；
然後用 Harness 驗證這些能力是否可靠；
最後用部署與推理優化，讓這些能力在生產環境中規模化。
這條路的核心思想是：
一個好的 Agent 系統，不只是「能跑」，還要「可靠」，更要「能持續運營」。
沒有 Skill，你沒有能力。
沒有 Harness，你不知道能力好不好。
沒有部署優化，你無法規模化。
三者缺一不可。
希望這個系列能幫助你，從打造第一個 Skill，到部署第一個生產級的 Agent 系統。
這條路很長，但每一步都值得。
下一步
如果你還想繼續深入，以下是幾個可能的延伸方向：
- 多模態 Agent：結合文字、圖像、語音的 Agent。
- Agent 安全與對齊：如何確保 Agent 的行為符合人類價值。
- Agent 經濟學：如何計算 Agent 的投資報酬率。
- Agent 治理：如何在組織中管理大量的 Agent。
- Agent 與人類協作：如何設計人機協作的介面與流程。
感謝你讀完這個系列。
祝你打造出改變世界的 Agent。