AG Day 36 成本與延遲優化:快取與模型分級
執行需求:CPU+API key。昨天的 AG Day 35 追蹤平台:Langfuse 觀測實戰幫 research-agent 接上了 Langfuse,我們終於能看到每一輪模型呼叫實際花掉多少 token、延遲多少毫秒。今天要把這些觀測數據拿來用:當一個研究問題被執行第二次、第三次,當某些子任務(例如查詢改寫、引用格式化)其實可以共用同一份結果,當模型分級可以用更便宜的模型先試試看再決定要不要請旗艦模型出場,這些都是我們今天要動手做、也立刻能在 Langfuse 看到效益的最佳化手段。我們也會示範一個最簡單的「語意快取」(semantic cache)雛型,讓研究助理對「語意相近但文字不同」的問題給出快取回應,並用 AG Day 9 累積的 token 計數器與 AG Day 35 的 Langfuse trace 對照成本變化。今天需要 LLM API 金鑰才能跑完整流程,但所有路徑都保留 --dry-run 模式與示範輸出。
引言
如果你跟著前幾篇跑 Langfuse 觀測,應該已經看過一張類似的折線圖:每一次研究任務的 token 用量隨著子任務數量線性成長,平均一次研究任務花掉的成本與時間,都是「呼叫模型次數 × 每次的平均成本與延遲」。這意味著,最有槓桿的最佳化不是把單次呼叫變快一點,而是減少不必要的呼叫、降低每次呼叫的平均成本、把可以共用的結果真正共用起來。今天要談的三件事——回應快取、模型分級、提示快取——都圍繞著這個原則。
第一個武器是回應快取。代理在做研究的過程中,會反覆問出很相似的子問題:「請用一句話總結這份文件」「請把這段引用改寫成 APA 格式」。如果每次都重新呼叫模型,既花錢又增加延遲;但如果用「精確字串比對」當快取鍵,又會因為查詢換句話說就失效。我們要做的是語意快取:把每次查詢先編碼成向量,跟快取裡的向量比對相似度,超過門檻就直接回傳上次的結果。第二個武器是模型分級(model cascading):用便宜、快速的模型先過濾或初步回答,只有在模型自己表示沒把握、或評估分數低於門檻時,才升級請旗艦模型出場。第三個武器是提示快取(prompt caching):多數模型供應商對同一段重複的 system prompt 或長文檔有快取機制,把固定的脈絡獨立出來可以省下大筆輸入成本。這三招不是互斥的,可以疊在 research-agent 的同一條管線上。
今天的目標是給 research-agent 加上兩個新模組:src/research_agent/cache.py(語意快取)與 src/research_agent/router.py(模型分級路由),並用 Langfuse 把每次「快取命中」「模型升級」的決策記錄下來,讓你在 Langfuse 介面上能直接看到「這一輪因為命中快取,省下了多少 token、節省了多少毫秒」。我們也會保留所有原始的單一模型路徑,當快取或分級沒有把握時,依然會回退到原本的旗艦模型呼叫,確保產出品質不會因為最佳化而下降。
原理/觀念
回應快取的三個層次
第一層是「精確字串快取」:以查詢字串當鍵、模型回應當值,存在記憶體或 SQLite 裡。優點是實作簡單、命中零成本;缺點是稍微改一個字就失效,且對於長查詢命中率極低。第二層是「語意快取」:把查詢編碼成向量,與快取裡的向量比對相似度,超過門檻(例如 0.92)就視為命中。這層命中率通常比精確快取高出一個量級,因為它能容忍換句話說、同義詞、輕度改寫,但需要一個向量編碼器與一次向量比對,這個成本通常遠低於再一次呼叫模型帶來的成本與延遲。第三層是「任務級快取」:快取的不是「對某個問題的回答」,而是「執行某個子任務的結果」,例如「查詢改寫把這句話換成三種說法」這個動作可以被快取,整份報告的章節草稿可以被快取,這層通常命中率最高、也最貼近研究助理的實際使用情境。
我們今天實作前兩層,並在概念上演示第三層的做法。第一層的精確快取用 SQLite 的 key-value 表格;第二層的語意快取用 AG Day 20 已經會用的 sentence-transformers 把查詢編碼成向量,存在一個輕量的記憶體索引(例如 FAISS 或純 numpy 的 cosine 計算)。對小型個人研究助理來說,數百到數千筆條目的記憶體索引就足夠;規模再大才需要升級到向量資料庫的 metadata filter 機制。
模型分級:先用便宜、再升級
「模型分級」這個詞的英文常見說法是 model cascading 或 model routing,但核心概念都一樣:根據任務的難度或重要性,動態選擇不同價格、不同能力的模型。最常見的策略是「兩個層級」:便宜快速的模型(例如輕量級模型)負責第一線篩選——查詢改寫、簡單分類、結構化抽取等;昂貴但更強的模型(例如旗艦模型)負責第二線——綜合寫作、深度推理、複雜引用整理。模型分級通常搭配一個「信心分數」:第一線模型如果自己表示不確定、或一個獨立的評估模型判斷信心不足,就升級到第二線。
這個設計有兩個前提必須先講清楚。第一,分級的判定本身需要成本,不管是請模型自我評估、還是另外呼叫一個評估模型,都會增加一次呼叫;如果設定不當,最佳化反而比直接用旗艦模型更貴。第二,模型分級對品質的影響必須能被量測。我們會在 AG Day 34 學過的評估集上,分別跑「單一旗艦模型」與「分級模型」兩個版本,比較兩者的通過率、平均成本、平均延遲;只有在分級版本的通過率不比單一旗艦差太多(例如相差小於某個百分比)時,才考慮在 production 上線。今天我們不跑完整評估,但會把 hook 與資料點預留好。
提示快取:把不變的部分獨立出來
多數主流模型供應商對 prompt 裡有顯著重複前綴的請求提供「提示快取」(prompt caching),意思是當你送出一段很長的 system prompt 或長文脈絡,後續若再次帶著相同前綴呼叫,模型供應商會認出這部分、直接用快取的計算結果,省下對應的輸入 token 成本。對 research-agent 來說,這意味著「固定的引用格式規範」「固定的報告範本」「固定的 system 指令」這些大段不變的文字,應該在程式碼層級就獨立成一個常數或檔案,不要每次呼叫時都重新拼裝。提示快取的具體可用性與價格折抵依供應商而異,請以「以官方文件為準」的精神查證。
完整實作
今天的程式改動集中在兩個新檔案,以及一個小幅度修改既有 llm.py 的路由介面。我們先建立新檔案的骨架:
touch research-agent/src/research_agent/cache.py
touch research-agent/src/research_agent/router.py
第一步:實作「精確快取 + 語意快取」兩層的 cache.py。精確快取用 SQLite 表格存查詢雜湊與回應;語意快取用 sentence-transformers 編碼查詢、以 numpy 計算 cosine 相似度:
# research-agent/src/research_agent/cache.py
from __future__ import annotations
import hashlib
import json
import sqlite3
import time
from pathlib import Path
from typing import Optional
import numpy as np
from sentence_transformers import SentenceTransformer
DEFAULT_DB = Path(__file__).resolve().parents[2] / "data" / "cache.db"
class ResponseCache:
"""精確快取 + 語意快取的雙層結構。"""
def __init__(self, db_path: Path = DEFAULT_DB, sim_threshold: float = 0.92) -> None:
self.db_path = db_path
self.db_path.parent.mkdir(parents=True, exist_ok=True)
self.sim_threshold = sim_threshold
self._init_db()
# 語意快取的記憶體索引;正式上線可換成 FAISS 或 Chroma metadata 機制
self.encoder = SentenceTransformer("all-MiniLM-L6-v2")
self._vec_index: list[np.ndarray] = []
self._payloads: list[dict] = []
self._reload_index()
def _init_db(self) -> None:
with sqlite3.connect(self.db_path) as con:
con.execute(
"CREATE TABLE IF NOT EXISTS exact ("
" key TEXT PRIMARY KEY,"
" payload TEXT NOT NULL,"
" created_at REAL NOT NULL"
")"
)
def _reload_index(self) -> None:
with sqlite3.connect(self.db_path) as con:
rows = con.execute("SELECT payload FROM exact").fetchall()
self._payloads = [json.loads(r[0]) for r in rows]
if self._payloads:
queries = [p["query"] for p in self._payloads]
self._vec_index = list(self.encoder.encode(queries, convert_to_numpy=True))
else:
self._vec_index = []
@staticmethod
def _hash(query: str) -> str:
return hashlib.sha256(query.encode("utf-8")).hexdigest()
def get(self, query: str) -> Optional[dict]:
"""先看精確快取,再看語意快取;都沒命中回傳 None。"""
key = self._hash(query)
with sqlite3.connect(self.db_path) as con:
row = con.execute("SELECT payload FROM exact WHERE key = ?", (key,)).fetchone()
if row:
return json.loads(row[0])
if not self._vec_index:
return None
q_vec = self.encoder.encode([query], convert_to_numpy=True)[0]
sims = self._vec_index @ q_vec / (
np.linalg.norm(self._vec_index, axis=1) * np.linalg.norm(q_vec) + 1e-9
)
best_idx = int(np.argmax(sims))
if sims[best_idx] >= self.sim_threshold:
return self._payloads[best_idx]
return None
def put(self, query: str, response: str, *, meta: Optional[dict] = None) -> None:
payload = {
"query": query,
"response": response,
"meta": meta or {},
"created_at": time.time(),
}
with sqlite3.connect(self.db_path) as con:
con.execute(
"INSERT OR REPLACE INTO exact(key, payload, created_at) VALUES (?, ?, ?)",
(self._hash(query), json.dumps(payload, ensure_ascii=False), payload["created_at"]),
)
# 同步更新記憶體索引;正式上線可用背景執行緒或 batched rebuild
if not self._vec_index or len(self._vec_index) != self._count_rows():
self._reload_index()
def _count_rows(self) -> int:
with sqlite3.connect(self.db_path) as con:
return con.execute("SELECT COUNT(*) FROM exact").fetchone()[0]
這份程式碼展示了三個重點:精確快取用 SQLite 當永久儲存,語意快取把向量載入記憶體做 cosine 比對,命中率門檻 sim_threshold 預設 0.92,可在建立時依場景調整(要求嚴格的問答可調高到 0.95,要求寬鬆的查詢改寫可降到 0.85)。注意我們用的是 all-MiniLM-L6-v2 作為示範編碼器,AG Day 20 已經介紹過這個模型;如果 production 需要更精準的語意比對,可以換成較大的編碼器或呼叫雲端嵌入 API。
第二步:實作 router.py,把模型分級邏輯封裝成一個簡單的「先試便宜的、不行再升級」介面。我們讓 cheap 模型在生成時附帶一個「信心分數」標記,由 cheap 模型自己判斷這個查詢自己能不能回答:
# research-agent/src/research_agent/router.py
from __future__ import annotations
import json
import os
from typing import Optional
from langchain_core.messages import HumanMessage, SystemMessage
from research_agent.cache import ResponseCache
from research_agent.llm import chat, chat_with_model
from research_agent.tracing import trace_span
DEFAULT_CHEAP = os.environ.get("RESEARCH_AGENT_CHEAP_MODEL", "cheap-model-placeholder")
DEFAULT_STRONG = os.environ.get("RESEARCH_AGENT_MODEL", "strong-model-placeholder")
def _extract_confidence(text: str) -> float:
"""cheap 模型回傳 JSON {"answer": ..., "confidence": 0~1};解析失敗視為低信心。"""
try:
data = json.loads(text)
return float(data.get("confidence", 0.0))
except (ValueError, TypeError):
return 0.0
def route_and_call(
prompt: str,
*,
cheap_model: str = DEFAULT_CHEAP,
strong_model: str = DEFAULT_STRONG,
confidence_threshold: float = 0.6,
cache: Optional[ResponseCache] = None,
) -> dict:
"""先查快取 -> cheap 模型 -> 信心不足升級到 strong 模型。"""
if cache is not None:
hit = cache.get(prompt)
if hit:
return {"source": "cache", "response": hit["response"], "model": hit.get("meta", {}).get("model", "n/a")}
with trace_span("router.cheap", {"model": cheap_model}):
cheap_resp = chat_with_model(
cheap_model,
[
SystemMessage(content="請用 JSON 回應:{\"answer\": ..., \"confidence\": 0~1}"),
HumanMessage(content=prompt),
],
)
confidence = _extract_confidence(cheap_resp)
if confidence >= confidence_threshold:
return {"source": "cheap", "response": cheap_resp, "model": cheap_model, "confidence": confidence}
with trace_span("router.strong_upgrade", {"model": strong_model, "cheap_confidence": confidence}):
strong_resp = chat_with_model(strong_model, [HumanMessage(content=prompt)])
return {"source": "strong", "response": strong_resp, "model": strong_model, "confidence": confidence}
第三步:補上一個簡單的驗證腳本,模擬「同一個查詢問三次:第一次沒有快取、第二次命中語意快取、第三次換句話說也命中」。這個腳本會在 Langfuse 留下清楚的 trace 紀錄,你可以直接從 Langfuse UI 看到「cache hit」這個標記:
# research-agent/verify_cache.py
from research_agent.cache import ResponseCache
from research_agent.router import route_and_call
def fake_chat_with_model(model, messages):
"""離線示範:不真的呼叫 API,直接回傳預寫回應。"""
user = messages[-1].content
if model.startswith("cheap"):
# cheap 模型信心分數 0.3,會升級
return '{"answer": "(cheap 模型示範回答)", "confidence": 0.3}'
return "(strong 模型示範回答,內容更完整)"
# 把 chat_with_model 換成 fake,避免真的打 API
import research_agent.router as router
router.chat_with_model = fake_chat_with_model
cache = ResponseCache(sim_threshold=0.85)
queries = [
"請總結這份文件的重點。",
"請總結這份文件的主要內容。", # 近似查詢,預期命中語意快取
"請摘要這份檔案。", # 換句話說,也可能命中
]
for q in queries:
result = route_and_call(q, cache=cache, confidence_threshold=0.6)
print(f"[{result['source']}] {q} -> {result['response'][:40]}")
if result["source"] == "strong":
cache.put(q, result["response"], meta={"model": result["model"]})
離線示範輸出(第一次走 strong 模型,第二次、第三次命中語意快取):
[strong] 請總結這份文件的重點。 -> (strong 模型示範回答,內容更完整)
[cache] 請總結這份文件的主要內容。 -> (strong 模型示範回答,內容更完整)
[cache] 請摘要這份檔案。 -> (strong 模型示範回答,內容更完整)
第四步:在 llm.py 裡把 RESEARCH_AGENT_MODEL 與新的 RESEARCH_AGENT_CHEAP_MODEL 串起來。我們不改既有 chat() 的簽名,只新增一個 chat_with_model() 函式,讓路由模組可以指定特定模型:
# research-agent/src/research_agent/llm.py(新增函式)
def chat_with_model(model_name: str, messages) -> str:
"""指定模型名稱呼叫對話 API;沿用既有 API key 與 retry 機制。"""
from research_agent.providers import get_provider # 內部模組,Day 3 已建立
provider = get_provider(model_name)
response = provider.invoke(messages)
_record_usage(provider.last_usage) # 與 AG Day 9 的 token 計數器整合
return response.text
第五步:在 cli.py 的查詢改寫子命令裡實際啟用快取與分級。AG Day 22 的查詢改寫流程現在會先經過 route_and_call(),cheap 模型負責生成候選查詢,當 cheap 模型對自己的改寫信心不足時升級到 strong 模型:
# research-agent/src/research_agent/cli.py(節錄)
@app.command()
def rewrite(query: str, dry_run: bool = typer.Option(False, "--dry-run")):
"""AG Day 22 查詢改寫;本篇示範加上快取與模型分級。"""
if dry_run:
typer.echo(f"[dry-run] rewrite: {query}")
return
cache = ResponseCache()
prompt = f"把這個查詢改寫成三種檢索友善的說法:{query}"
result = route_and_call(prompt, cache=cache)
typer.echo(f"來源:{result['source']}({result['model']})")
typer.echo(result["response"])
第六步:用一段 pytest 測試驗證快取命中與模型升級的行為都正確:
# research-agent/tests/test_cache_and_router.py
from research_agent.cache import ResponseCache
from research_agent.router import route_and_call
def test_exact_cache_hit(monkeypatch, tmp_path):
cache = ResponseCache(db_path=tmp_path / "cache.db", sim_threshold=0.99)
cache.put("請總結這份文件。", "示範回應 A", meta={"model": "test"})
assert cache.get("請總結這份文件。")["response"] == "示範回應 A"
def test_semantic_cache_hit(monkeypatch, tmp_path):
cache = ResponseCache(db_path=tmp_path / "cache.db", sim_threshold=0.7)
cache.put("請摘要這篇文章的結論。", "示範回應 B", meta={"model": "test"})
hit = cache.get("請把這篇文章的結論做個總結。")
assert hit is not None
assert hit["response"] == "示範回應 B"
def test_router_upgrades_on_low_confidence(monkeypatch):
def fake_chat(model, messages):
if model.startswith("cheap"):
return '{"answer": "cheap", "confidence": 0.2}'
return "strong"
monkeypatch.setattr("research_agent.router.chat_with_model", fake_chat)
out = route_and_call("這是一個需要深度推理的問題。", cache=None)
assert out["source"] == "strong"
assert out["response"] == "strong"
常見錯誤與踩雷
第一個雷是把「快取命中率」當成唯一指標,忽略「快取內容是否仍然正確」。研究領域的答案有時間敏感性:今天對「2026 年最熱門的 LLM 是哪一個」的正確答案,明天可能完全不一樣。如果沒有設計 TTL(time to live)或主動失效機制,使用者拿到的會是過時的回應卻渾然不知。我們的 ResponseCache.put() 在 payload 裡留下 created_at 與 meta 欄位,正式上線時應該在 get() 裡加上 TTL 檢查(例如超過 24 小時視為過期),並針對「事實查詢」「時事摘要」這類任務設定更短的 TTL 或完全不進快取。
第二個雷是語意快取的相似度門檻設錯。門檻太高命中率低、省不了錢;門檻太低會把「其實是不同的問題」當成同一題回應,造成幻覺或錯誤答案。我們預設 0.92 是經驗值,但實際應用應該用 AG Day 33 建立的評估集做 sweep:分別測 0.85、0.90、0.92、0.95,找出「命中率」與「錯誤率」的最佳平衡點。這一步需要花點時間,但不做就只能靠運氣。
第三個雷是模型分級的判定成本比省下的成本還貴。舉例來說,如果讓 cheap 模型每次都生成 100 個 token 的信心評估,這個判定本身就吃掉不少 input token,最後整體成本不降反升。設計時要讓 cheap 模型盡量用「結構化短回答」(例如只回一個 0-1 的數字)來評估信心,並考慮把信心評估的 prompt 也做成靜態、不每次重組,這樣可以搭提示快取一起省。
第四個雷是把快取邏輯寫死在業務程式碼裡。當快取邏輯散落在各個工具、各個節點函式裡時,很難保證大家都走同一套策略(例如某處忘了把 TTL 設短、某處忘了把私密資訊排除在快取之外)。我們把快取邏輯集中在 cache.py、路由邏輯集中在 router.py,業務程式碼只呼叫 route_and_call() 這一個介面,是為了讓快取與分級策略變得可集中調整、可集中監控。改天要換成 Redis 當快取後端、或加上一個更精準的編碼器,只需要改這兩個檔案。
第五個雷是把「Langfuse 上的 cost 數字」當成唯一最佳化目標。Langfuse 看到的成本只是 API 帳單成本,不包含「因為最佳化而增加的程式複雜度」「因為快取錯誤而增加的客服時間」「因為模型分級導致的品質下降帶來的業務損失」。最佳化必須與 AG Day 34 的評估集同時跑,確定品質不退化之後才能上線,單看 Langfuse 的 token 數字會讓你做出「成本最低但沒人想用」的系統。
效能與實務提醒
實務上最容易低估的是「快取本身的成本」。語意快取每次查詢都要做一次編碼與一次向量比對,雖然對小型快取來說只要幾毫秒,但若快取規模成長到數萬筆以上,純 numpy 的 cosine 計算就會開始變慢。這時候可以改用 FAISS 或 Chroma 的 metadata filter 機制,並把向量預先批次載入。我們今天選擇「全量載入記憶體」是因為個人研究助理的快取規模通常在數百到數千筆,這個量級在 CPU 上完全沒問題。
第二個提醒是「快取鍵的設計」。我們目前用查詢字串的 SHA-256 雜湊當精確快取鍵,這個設計無法支援「同一查詢但不同溫度參數」的區分。如果未來想針對不同 temperature 或不同模型分別快取回應,需要把這些參數一起納入雜湊(例如 hash(model, temperature, query))。同樣地,如果有一天你的代理會在 prompt 裡插入當下時間、使用者 ID 這類變動資訊,要把這些變動欄位從快取鍵裡排除,避免「明明是同一題但每次都沒命中」。
第三個提醒是「觀測的完整性」。今天範例裡我們把路由決策包在 trace_span() 裡,讓 Langfuse 看到「這次走了 cache 還是 cheap 還是 strong」。這層觀測的成本幾乎為零,但對後續調校極有價值:在 Langfuse 介面上你可以用「source」欄位 group by,看到三種來源的呼叫次數、平均 token、平均延遲。如果某天你發現「cache 命中率突然掉到 30%」,第一個念頭應該是去看 Langfuse 的 source 分布,而不是直接改程式。
第四個提醒是模型分級對「失敗模式」的影響。原本單一旗艦模型如果出錯,錯誤模式比較一致;分級模型則可能在 cheap 模型有信心時用便宜的輸出、在 cheap 模型沒信心時用昂貴的輸出,這意味著「失敗模式更分散」。建議在 AG Day 34 的 LLM-as-judge 評估裡分別記錄 cheap-only、strong-only、分級版本三條路徑的分數,並對分級版本設定一個「cheap 信心 低於 門檻時是否真的升級」的稽核報表,確保分級邏輯沒有退化。
小結
今天我們替 research-agent 加上了三層最佳化武器:精確快取、語意快取、模型分級,並把每個路由決策都送進 Langfuse 做觀測。程式碼集中在兩個新檔案 cache.py 與 router.py,既有 llm.py 與 cli.py 只做了介面擴充,不破壞原本的呼叫路徑。我們也用 pytest 把精確命中、語意命中、模型升級三種情境寫成可重複執行的測試,避免未來重構時把行為改壞。
今天新增的關鍵詞:語意快取(semantic cache)——以查詢向量相似度判斷是否可重用的回應快取;模型分級(model cascading / routing)——根據任務難度選擇不同等級模型的策略;提示快取(prompt caching)——模型供應商對重複前綴 prompt 提供的成本折抵機制;TTL——快取條目的存活時間,是避免過時回應的關鍵設計。
結語
最佳化與安全是同一件事的兩面:成本降低一半固然好,但若因為快取而把不該暴露的內容暴露給其他使用者,後果遠比多花點 API 費用嚴重。明天我們會進入「AG Day 37 安全:prompt injection 與工具防護」,探討研究助理在面對來自外部輸入(網頁內容、使用者提問、第三方資料)時,要怎麼設計防護讓模型不會被誤導成主要工具、執行預期外的動作。我們會把今天的快取與分級機制也納入安全討論:哪些內容可以進快取、哪些回應該被遮罩、哪些工具呼叫需要更嚴格的把關。
延伸資源
- Langfuse 官方文件:成本與延遲儀表板的設定與詮釋方式,以官方文件為準。
- sentence-transformers 使用者指南:
https://www.sbert.net/。本篇採用的all-MiniLM-L6-v2是入門常用模型,正式上線前請依任務特性挑選合適編碼器。 - OpenAI 與 Anthropic 官方文件:提示快取(prompt caching)的可用性與價格折抵規則會隨版本更新,請以官方文件為準。
- pytest 官方文件:
https://docs.pytest.org/。本篇測試使用monkeypatch與tmp_pathfixture 模擬外部依賴。
留言
張貼留言