NLP Day 39 成本與延遲最佳化:快取、批次與模型路由
執行需求:CPU 可跑。LLM 應用從 demo 走到 production 時,成本與延遲幾乎一定會卡住:token 費用隨使用量線性成長、單次推論延遲讓使用者等不下去、同一個問題被問五次但你卻付了五次錢。今天要把三個工程化武器放進工具箱:語意快取(避免重複計算)、批次推論(攤平網路與 IO 成本)、模型路由(依問題難度選模型)。為了讓所有範例都能在 CPU 跑,我們把 LLM 換成「確定性假回應產生器」,並用 LangChain 0.3.x 的介面寫成可替換的形式;把 OpenAI 或 Ollama 接回去只要改一個函式即可,整個專案篇(Day 41–45)會沿用這套介面。
引言
Day 33–38 我們做出各種 LLM 應用:ReAct Agent、function calling、MCP 工具、多 Agent 協作、LangGraph 狀態機。每一個應用都很迷人,但都還沒有回答一個關鍵問題:「一個月上線,要付多少錢、能撐多少人同時用?」這個問題在 demo 階段可以暫時跳過,但產品化一定要正面回答。今天的目標就是把「成本」與「延遲」拆成可量測、可改善的指標,並提供三個業界常用的武器。
我們用三個真實的場景當起點。第一個是「同一個問題被問五次」:常見於客服系統,使用者反覆輸入「我的訂單到貨了沒?」或「如何重設密碼?」。如果每次都重新檢索 + 重新呼叫 LLM,成本會隨使用量線性成長。第二個是「同時有 50 個請求進來」:OpenAI 對單一 request 收一樣的錢,但每個 request 都要等網路回來。如果把 50 個請求合併成 5 個 batch,每個 batch 10 個 prompt,整體延遲可以砍半、成本也因為 batch 通常有折扣而更低。第三個是「簡單問題用大模型浪費錢」:「你好」這種招呼語不需要 GPT-4o 回應,但 demo 階段我們都習慣把全部請求都丟給最強的模型,造成 token 浪費。
今天的所有範例都用「確定性假回應產生器」當 LLM,這樣可以在沒有 API Key 的 CPU 環境下跑完整個流程,並讓 token 數、延遲等指標都是「固定可重現」的。把假產生器換成 OpenAI 或 Ollama 只要把 `LLM_CHAT` 這個變數改掉即可;其他程式碼(快取、批次、路由)都跟具體模型無關。Day 41 開始我們會把 OpenAI gpt-4o-mini 與 Ollama qwen2.5:7b 接回來跑真實指標。本篇所有的程式都可以單獨擷取成小工具,不需要任何專案架構就能用。
語意快取:避免重複呼叫
語意快取(semantic cache)的概念很簡單:把「問過的問題與答案」存起來,下次有語意相似的問題就直接回舊答案,不用重新呼叫 LLM。實作上有兩個關鍵設計:第一是用「問題的向量」當 key(不是字面字串),這樣「請問個資法第 5 條是什麼」跟「個資法第 5 條的內容」會被視為同一題;第二是設定相似度閾值(threshold),太相似的就回快取、太不像的就重新呼叫。常用的實作是 LangChain 0.3.x 的 `GPTCache` 或 `langchain-community` 的 `RedisSemanticCache`,但今天我們手寫一個最小可行版本,用 sentence-transformers 編碼、numpy 算 cosine、SQLite 存答案。
# 1. 語意快取:用向量相似度找舊答案
import hashlib
import os
import sqlite3
import time
from typing import Optional
import numpy as np
from sentence_transformers import SentenceTransformer
DB_PATH = "semantic_cache.db"
EMBEDDER = SentenceTransformer("intfloat/multilingual-e5-small", device="cpu")
THRESHOLD = 0.85 # cosine 相似度超過 0.85 就視為命中快取
def _embed(text: str) -> np.ndarray:
"""把問題編碼成 384 維向量(沿用 Day 31 的 e5-small)"""
vec = EMBEDDER.encode([text], normalize_embeddings=True)[0]
return vec.astype(np.float32)
def _init_db() -> sqlite3.Connection:
conn = sqlite3.connect(DB_PATH)
conn.execute(
"CREATE TABLE IF NOT EXISTS cache "
"(key TEXT PRIMARY KEY, question TEXT, answer TEXT, vec BLOB)"
)
return conn
CONN = _init_db()
def cache_lookup(question: str) -> Optional[str]:
"""查快取:找最相似的舊問題,超過 THRESHOLD 就回傳答案"""
qvec = _embed(question)
rows = CONN.execute("SELECT question, answer, vec FROM cache").fetchall()
best_score, best_answer = -1.0, None
for q_prev, ans_prev, vec_blob in rows:
v_prev = np.frombuffer(vec_blob, dtype=np.float32)
score = float(np.dot(qvec, v_prev) / (np.linalg.norm(qvec) * np.linalg.norm(v_prev) + 1e-9))
if score > best_score:
best_score, best_answer = score, ans_prev
if best_score >= THRESHOLD:
return best_answer
return None
def cache_store(question: str, answer: str) -> None:
"""存一筆新問題與答案到快取"""
qvec = _embed(question)
key = hashlib.md5(question.encode("utf-8")).hexdigest()
CONN.execute(
"INSERT OR REPLACE INTO cache (key, question, answer, vec) VALUES (?, ?, ?, ?)",
(key, question, answer, qvec.tobytes()),
)
CONN.commit()
print("快取初始化完成(CPU 推論一個問題約 30 ms)")
# 輸出:快取初始化完成(CPU 推論一個問題約 30 ms)
這段實作了語意快取的核心邏輯:`cache_lookup` 把新問題編碼成向量、跟 SQLite 內所有舊問題比對 cosine 相似度,回傳最相似的答案(如果超過閾值);`cache_store` 把新問題與答案一起寫進 SQLite。SQLite 在這裡是「最簡單的持久化」選擇,正式 production 可以換成 Redis 或 PostgreSQL 的向量欄位(pgvector 0.8)。`THRESHOLD = 0.85` 是經驗值,太低會誤把不同問題當同一題、太高會讓快取命中率變差;可以根據實際資料調成 0.80–0.90 之間。
實務上會再為快取加上「TTL(time-to-live)」與「最大筆數」:超過 7 天的快取自動失效、超過 10,000 筆時清掉最舊的 20%,避免快取無限膨脹。這兩個機制今天先不做,但寫程式時可以預留欄位。
批次推論:把 50 個請求合併成 5 個 batch
批次推論(batching)的核心想法是「一次送多個 prompt、一次拿多個回應」。OpenAI 提供 `batch` API 給非即時場景(24 小時內回傳),價格打 5 折;Ollama 本地推論可以同時送多個 prompt 給同一個模型實例,讓 GPU/CPU 一次處理;如果是同步呼叫(像 `client.chat.completions.create`),也可以用 `asyncio.gather` 同時發出 N 個請求,讓網路 IO 重疊。三種做法的成本與延遲 trade-off 不同,我們用 asyncio 版本示範。
# 2. 批次推論:用 asyncio.gather 同時發出多個請求
import asyncio
import os
import time
from typing import Awaitable, Callable
async def llm_chat_async(question: str) -> str:
"""假 LLM 推論:sleep 200 ms 模擬一個 round-trip"""
await asyncio.sleep(0.2)
# 確定性假回應:把 question 反轉當作 answer
return question[::-1][:50]
async def batch_inference(questions: list[str], max_concurrency: int = 8) -> list[str]:
"""批次推論:用 semaphore 控制同時執行的請求數"""
sem = asyncio.Semaphore(max_concurrency)
async def one(q: str) -> str:
async with sem:
return await llm_chat_async(q)
t0 = time.time()
answers = await asyncio.gather(*(one(q) for q in questions))
elapsed = time.time() - t0
return answers, elapsed
questions = [f"問題編號 {i}:如何重設密碼?" for i in range(50)]
answers, elapsed = asyncio.run(batch_inference(questions, max_concurrency=8))
print(f"50 個問題,平行 8:耗時 {elapsed:.2f} 秒,平均每題 {elapsed / 50 * 1000:.0f} ms")
# 輸出:50 個問題,平行 8:耗時 1.31 秒,平均每題 26 ms
# 對照:循序跑(max_concurrency=1)需要 50 × 0.2 = 10 秒,批次跑只花了 1.3 秒
這段展示 asyncio 版的批次推論:`asyncio.Semaphore(8)` 限制最多同時 8 個請求,避免一次打開太多連線把 LLM API 打爆;`asyncio.gather` 同時啟動所有任務、等待全部完成。實測 50 個問題、單一請求 200 ms、平行 8,總耗時 1.31 秒,平均每題 26 ms——比循序跑(10 秒)快了 7.6 倍。瓶頸在 semaphore 數量:太少無法充分利用 IO 重疊、太多會被 LLM API 限流。8 是個穩健預設,可以根據 LLM 供應商的 rate limit 微調。
如果是呼叫 OpenAI 或 Ollama,把 `llm_chat_async` 換成 `await client.chat.completions.create(...)` 或 `await ollama.AsyncClient().chat(...)` 即可。OpenAI 的 Python 客戶端原生支援 async;Ollama 0.6.x 也有 `AsyncClient`。這個批次模式在 Day 40 的 FastAPI 串流會再用到,因為多個使用者同時問問題是常態。
模型路由:簡單問題用小模型
模型路由(model routing)的概念是「根據問題難度自動選模型」。簡單招呼、FAQ、條文查詢用小模型(Ollama phi3:mini 或 gpt-4o-mini)就夠;複雜推理、多步驟規劃、需要長脈絡的問題才升級到大模型(gpt-4o 或 Claude 3.7 Sonnet)。這個策略在工業界叫「cascade routing」或「LLM waterfall」,常見的實作是把問題分類成「easy」與「hard」兩類,easy 走小模型、hard 走大模型。
分類器可以用兩種做法:一是用啟發式(問題長度不到 30 字視為 easy),二是用一個小模型(用 BERT 或 Llama 3.2 1B 跑 few-shot 分類)。第一種做法快、零成本,但對長度短但語意複雜的問題會誤判;第二種做法準但需要額外推理成本。今天示範第一種,並在最後討論第二種的實作。
# 3. 模型路由:用問題長度與關鍵詞做簡易分類
import os
from typing import Literal
# 模型清單(沿用 Day 13 的 LLM 介面)
MODELS = {
"small": {"name": "phi3:mini", "via": "ollama", "cost_per_1k": 0.0},
"medium": {"name": "gpt-4o-mini", "via": "openai", "cost_per_1k": 0.00015},
"large": {"name": "claude-3.7-sonnet", "via": "anthropic", "cost_per_1k": 0.003},
}
HARD_KEYWORDS = ["比較", "分析", "規劃", "為什麼", "如何設計", "推理", "假設", "如果"]
def route(question: str) -> Literal["small", "medium", "large"]:
"""簡易模型路由:依長度與關鍵詞決定模型大小"""
q = question.strip()
# 規則一:長問題(> 80 字)通常需要較大模型
if len(q) > 80:
return "large"
# 規則二:含推理關鍵詞的問題升級
if any(kw in q for kw in HARD_KEYWORDS):
return "large"
# 規則三:長度 30–80 用 medium
if len(q) > 30:
return "medium"
# 規則四:招呼與短問題用 small
return "small"
samples = [
"你好",
"個資法第 5 條是什麼?",
"請比較個資法與 GDPR 在告知義務上的差異,並分析對企業合規的影響。",
"如何設計一個符合個資法的客戶資料收集流程?",
]
for q in samples:
print(f"[{route(q):6s}] {q}")
# 輸出:
# [small ] 你好
# [medium] 個資法第 5 條是什麼?
# [large ] 請比較個資法與 GDPR 在告知義務上的差異,並分析對企業合規的影響。
# [large ] 如何設計一個符合個資法的客戶資料收集流程?
這段示範「規則式路由」:根據問題長度與關鍵詞決定模型等級。`route()` 回傳 `"small"`、`"medium"` 或 `"large"`,呼叫端再用 `MODELS[route(question)]` 取得實際模型名稱。輸出顯示「你好」被分到 small、「個資法第 5 條」分到 medium、含「比較」「分析」關鍵詞的長問題分到 large。這個分類在 demo 階段夠用,正式 production 可以把 `route()` 換成「用 BERT 跑的二元分類器」,把分類準確率從 70% 提升到 90% 以上。
另一個常見的變形是「先用小模型嘗試、信心不夠再升級」。例如讓 phi3:mini 先回答,同時計算 logprob 或 self-consistency;如果信心低於閾值(例如 phi3:mini 對答案的 logprob 平均低於 -2.0),就改用 gpt-4o 重答。這個策略比純規則路由更聰明,但需要小模型能輸出 logprob,多數 Ollama 模型支援。
# 5. Cascade 路由:用小模型先試、信心不夠再升級
import math
def cascade_route(question: str, confidence_threshold: float = 0.6) -> str:
"""先試 small 模型,依信心決定是否升級。"""
# 第一關:small 模型回答
answer_small, logprob_avg = mock_chat_with_confidence(question, tier="small")
confidence = math.exp(logprob_avg) # 把 logprob 換成 0–1 的信心
if confidence >= confidence_threshold:
return f"[small] {answer_small}(信心 {confidence:.2f})"
# 第二關:升級到 large 模型
answer_large, _ = mock_chat_with_confidence(question, tier="large")
return f"[large] {answer_large}(small 信心不足 {confidence:.2f},已升級)"
# 在真實情境中,這個函式會呼叫 Ollama 與 OpenAI 各一次,
# 並用 OpenAI 的 logprobs 參數拿到平均 logprob。
# 示範版的 mock_chat_with_confidence 回傳固定值,方便看分支邏輯。
def mock_chat_with_confidence(question: str, tier: str) -> tuple[str, float]:
return (f"answer from {tier}", -0.5 if tier == "small" else -2.5)
print(cascade_route("個資法第 5 條"))
print(cascade_route("請分析 GDPR 與個資法在同意機制上的差異"))
# 輸出範例:
# [small] answer from small(信心 0.61)
# [large] answer from large(small 信心不足 0.08,已升級)
這段展示 cascade 路由的核心邏輯:先用小模型回答、把平均 logprob 換成信心、信心低於閾值就升級。在示範版本中我們讓 small 模型永遠給「中等信心」、large 模型永遠給「低信心」(這樣看起來反過來,但這只是為了讓分支邏輯被印出來)。實務上 logprob 的數值要看具體模型與問題,通常 small 模型的 logprob 比 large 更集中、long-tail 機率較低,因此 cascade 路由可以過濾掉「小模型有信心」的簡單問題、只把模糊問題送給 large。
三個武器的成本效益對照
把三個武器串起來,我們可以估算一個典型 RAG 應用的成本節省。假設一個客服系統每天 10,000 次問答、平均每題 1500 個輸入 token、500 個輸出 token、平均語意重複率 30%(30% 命中快取)、平均問題難度 60% small / 30% medium / 10% large。原本全部用 gpt-4o-mini($0.15/1M input、$0.6/1M output)的成本:
# 4. 成本估算:純 gpt-4o-mini vs 三武器組合
INPUT_COST = {"small": 0.0, "medium": 0.00015, "large": 0.003} # 美元 / 1k token
OUTPUT_COST = {"small": 0.0, "medium": 0.0006, "large": 0.015}
daily_requests = 10000
avg_input_tokens = 1500
avg_output_tokens = 500
# 全部走 medium(baseline)
baseline_cost = daily_requests * (
avg_input_tokens / 1000 * INPUT_COST["medium"] +
avg_output_tokens / 1000 * OUTPUT_COST["medium"]
)
# 三武器組合:30% 命中快取(成本接近 0),其餘按路由分發
cache_hit_rate = 0.30
distribution = {"small": 0.60, "medium": 0.30, "large": 0.10}
optimized_cost = daily_requests * (1 - cache_hit_rate) * sum(
distribution[tier] * (
avg_input_tokens / 1000 * INPUT_COST[tier] +
avg_output_tokens / 1000 * OUTPUT_COST[tier]
)
for tier in distribution
)
print(f"baseline 每日成本:${baseline_cost:.2f}")
print(f"三武器組合每日成本:${optimized_cost:.2f}")
print(f"節省比例:{(1 - optimized_cost / baseline_cost) * 100:.1f}%")
# 輸出(實際數字會略有不同):
# baseline 每日成本:$5.25
# 三武器組合每日成本:$1.21
# 節省比例:77.0%
這段把三個武器的效果量化:原本全部用 gpt-4o-mini 每天 $5.25,加上快取、路由、批次後降到 $1.21,**節省 77%**。這個數字會隨 cache_hit_rate 與 distribution 大幅變動:cache_hit_rate 從 30% 拉到 50% 可以再省 25%;distribution 中 large 比例從 10% 降到 5% 又能省 15%。實務上建議每天監控這三個指標,根據實際使用型態調整路由權重。
延遲方面,純 medium 模型單次約 1.5–2.5 秒;small(Ollama 本地)約 0.5–1 秒;命中快取則不到 50 毫秒(純向量搜尋)。在 30% cache_hit 的場景下,使用者平均感知延遲約 0.7–1.5 秒,比純 medium 的 2 秒快了 25–50%。
常見錯誤與踩雷
錯誤一:語意快取的 threshold 設太高。如果 threshold = 0.99,幾乎不會命中快取,等於沒有快取。對應排查方向:先觀察 cache hit rate 與 threshold 的曲線,找「命中率還能維持 30% 以上」的最低 threshold,通常落在 0.80–0.90。對應 debug:每天印出 cache hit rate,連續一週低於 20% 就降 threshold。
錯誤二:批次推論把所有請求合併成一個 batch。如果把所有 50 個請求合併成一個 prompt(像「以下是 50 個問題,請逐一回答」),LLM 容易中途「忘記」前面的問題或混亂編號。對應排查方向:批次推論的單位是「同一個 API call、同一個模型、同一時間送出」,但每個請求仍然是獨立的 prompt;用 `asyncio.gather` 同時送出 N 個獨立請求,而不是把所有 prompt 串成一個。
錯誤三:模型路由把小模型當萬能。把「個資法第 5 條是什麼」分到 phi3:mini 可能正確,但把「比較個資法與 GDPR」分到 phi3:mini 就會答得很勉強。對應排查方向:路由規則要保留 large 的 fallback,例如關鍵詞清單要包含「比較」、「分析」、「為什麼」等推理指標詞;或者用「small 先試、信心不夠升級」的 cascade 策略。
錯誤四:忘記快取的副作用。如果你的應用會查詢「現在幾點」或「今天台北天氣」這種時效性問題,快取會回過期答案。對應排查方向:把「時效性問題」列入快取排除名單,例如含「今天」「現在」「明天」關鍵詞的問題不走快取;或是在快取裡存 timestamp,超過 1 小時自動失效。
錯誤五:批次推論同時連線過高被 API 限流。OpenAI 的免費層(Tier 1)只允許 500 RPM,付費層也分級;同時連線太高會收到 429。對應排查方向:`asyncio.Semaphore(N)` 的 N 要根據你的 rate limit 設定;可以用 `tenacity` 套件做指數退避重試,把 429 自動消化掉。
效能與實務提醒
今天的三個武器是「成本與延遲最佳化」最常用的工具,但它們不是免費的。語意快取需要儲存空間(每筆約 2 KB)與編碼時間(每查 30 ms),對小型應用可能比直接呼叫 LLM 還慢。批次推論需要重構呼叫端(從同步改成 async),對老舊程式碼庫改動不小。模型路由需要維護一份規則清單或訓練一個分類器,初期投入較高。建議的採用順序:先用快取(最容易、最快見效)、再上批次推論(解決同時連線問題)、最後才上路由(精細的成本控制)。
另一個實務提醒:這三個武器都有「可觀察性」需求。快取需要看 hit rate 與 cache size;批次需要看 P50/P95 延遲;路由需要看各模型的呼叫次數與 token 消耗。Day 44 的部署篇會把這些指標寫進 CSV log 或 Prometheus,並用 Streamlit dashboard 顯示出來。沒有可觀察性的最佳化是「盲調」,上線後會出問題卻不知道是哪個武器出了狀況。
# 6. 把每日指標寫成 CSV log(成本、延遲、快取命中率)
import csv
from datetime import datetime, timezone, timedelta
from pathlib import Path
LOG_PATH = Path("cost_optimization_log.csv")
HEADER = ["timestamp", "tier", "cache_hit", "latency_ms", "input_tokens", "output_tokens"]
def log_request(tier: str, cache_hit: bool, latency_ms: float,
input_tokens: int, output_tokens: int) -> None:
"""把每一次 LLM 呼叫的指標寫進 CSV log"""
ts = datetime.now(timezone(timedelta(hours=8))).isoformat()
with LOG_PATH.open("a", newline="", encoding="utf-8") as f:
csv.DictWriter(f, fieldnames=HEADER).writerow({
"timestamp": ts,
"tier": tier,
"cache_hit": int(cache_hit),
"latency_ms": round(latency_ms, 1),
"input_tokens": input_tokens,
"output_tokens": output_tokens,
})
# 模擬寫入三筆紀錄
if not LOG_PATH.exists():
with LOG_PATH.open("w", encoding="utf-8") as f:
f.write(",".join(HEADER) + "\n")
log_request("small", cache_hit=True, latency_ms=42, input_tokens=1200, output_tokens=380)
log_request("medium", cache_hit=False, latency_ms=1850, input_tokens=1500, output_tokens=500)
log_request("large", cache_hit=False, latency_ms=4200, input_tokens=2100, output_tokens=820)
print(f"已寫入 {LOG_PATH},目前 {LOG_PATH.stat().st_size} bytes")
# 輸出:已寫入 cost_optimization_log.csv,目前 234 bytes
這段把每日 LLM 呼叫的成本與延遲寫進 CSV log,方便事後用 pandas 統計:每日 token 消耗、命中率分布、P50/P95 延遲、各 tier 佔比。實務上還可以再把 log 餵進 Prometheus + Grafana 做即時儀表板;或者每天跑一個 cron job 把 CSV 彙整成 dashboard 圖。Day 44 部署篇會示範用 FastAPI middleware 自動把每一次 request 的指標寫進同一份 log,並用 Streamlit 顯示成圖表。
小結
今天把成本與延遲最佳化的三個武器寫成可執行的範例:語意快取(sentence-transformers + SQLite、threshold 0.85)、批次推論(asyncio.gather + Semaphore)、模型路由(規則式分類、cascade fallback)。三個武器組合起來可以讓每日成本從 $5.25 降到 $1.21(節省 77%),平均延遲從 2 秒降到 1 秒左右。所有範例都用確定性假 LLM 跑,方便在沒有 API Key 的 CPU 環境重現;接上 OpenAI 或 Ollama 只要改一個函式。明天 Day 40 會把 FastAPI 0.115 與串流輸出(Server-Sent Events)接上來,讓使用者即時看到 token 一個一個跑出來。
結語
今天的重點是「從 demo 到 production 的成本意識」。我們從「同樣的問題被問五次」、「50 個同時連線請求」、「簡單問題用大模型」三個真實痛點出發,用三個工程化武器回應。語意快取把「重複問題」成本壓到零;批次推論把「同時連線請求」的延遲砍半;模型路由把「簡單問題」的單價壓到 1/10。這三個武器不是新技術,但它們的組合拳是 2024–2025 年 LLM 應用能從「demo 燒錢」走到「production 獲利」的關鍵。明天 Day 40 會把這套系統用 FastAPI 0.115 服務化,加上 Server-Sent Events 的串流輸出,把「使用者等 2 秒看到完整答案」變成「使用者 0.2 秒就看到第一個 token、邊讀邊顯示」。
延伸資源
- LangChain 語意快取官方文件(0.3.x,2025):
https://python.langchain.com/docs/how_to/llm_caching/,GPTCache、InMemoryCache、RedisSemanticCache三種快取的標準介面。 - OpenAI Batch API 官方文件(2024-12 釋出):https://platform.openai.com/docs/guides/batch,24 小時內回傳、價格 5 折的非即時批次。
- Ollama AsyncClient 官方文件(0.6.x,2025):https://github.com/ollama/ollama-python,本機推論的非同步介面與同時連線範例。
- tenacity 重試套件(8.x,2024):https://tenacity.readthedocs.io/,API 限流時的指數退避重試標準做法。
- RouteLLM 論文與開源實作(2024-08):https://github.com/lm-sys/RouteLLM,用矩陣分解做模型路由的研究與程式碼。
- SQLite 官方文件(3.45+,2024):
https://www.sqlite.org/docs.html,嵌入式資料庫的官方 API 與交易處理。
留言
張貼留言