這個 pull request 的標題是「Add GPT-5.5 fallback」。diff 是：本來二十行就能搞定的事，變成 340 行——複製貼上的 retry decorator、寫死的 model 字串、五個幾乎一樣的 call_llm_with_retry 變體。
你的 review：「這裡需要抽象化。」作者：「具體是怎樣的抽象化？」
這篇文章回答這個問題。
六個 Gang of Four 模式翻譯成 LLM API 領域：模型選擇的 Factory、prompt 的 Strategy、串流的 Observer、retry 與日誌的 Decorator、fallback 的 Chain of Responsibility、agent loop 的 Template Method——每個都附上生產環境的 Python 程式碼，以及它取代的反模式。
模式一：Factory——集中式的模型實例化
問題
chat.py 裡有 "gpt-5.5"。summarizer.py 裡有 "claude-sonnet-4-20250514"。classifier.py 裡有 "deepseek-v4-flash"。模型遷移，意味著要在整個 codebase 裡搜尋取代——還得祈禱沒漏掉某個只在生產環境載入的 config 檔。
模式
一個帶集中式 registry 的 ModelFactory。決定用哪個模型的是任務類型，不是寫死的字串。環境變數讓 canary 部署與 rollback 成為可能。模型的 metadata——capabilities、cost tier、context window——跟 model ID 放在一起。實作用的是 OpenAI Python SDK——OpenAI 相容 API 的標準客戶端函式庫——它的 AsyncOpenAI 客戶端撐起底下的 Factory。
from dataclasses import dataclass
from openai import AsyncOpenAI
@dataclass
class ModelSpec:
model_id: str
provider: str
capabilities: list[str] # ["chat", "vision", "tools", "json_mode"]
cost_tier: str # "cheap", "mid", "frontier"
context_window: int
class ModelFactory:
def __init__(self, base_url: str, api_key: str):
self.client = AsyncOpenAI(base_url=base_url, api_key=api_key)
self.registry: dict[str, ModelSpec] = {}
self._load_registry()
def create(self, task_type: str, requirements: list[str] = None) -> tuple[AsyncOpenAI, ModelSpec]:
model_id = os.getenv(f"MODEL_OVERRIDE_{task_type.upper()}", None)
if model_id:
spec = self.registry[model_id]
else:
spec = self._select_by_capability(task_type, requirements or [])
return self.client, spec
def _select_by_capability(self, task_type: str, requirements: list[str]) -> ModelSpec:
candidates = [
m for m in self.registry.values()
if all(req in m.capabilities for req in requirements)
]
tier_map = {"classification": "cheap", "generation": "mid", "review": "frontier"}
tier = tier_map.get(task_type, "mid")
return next((m for m in candidates if m.cost_tier == tier), candidates[0])統一的 API endpoint——所有 provider 共用一個 base_url——把 Factory 的 config 從 O(N 個 provider) 縮小成 O(1 個 endpoint ＋ N 個 model 字串)。一個 API key。一個 client 實例。registry 裡每個模型都能透過它存取。建立這種單一入口架構，要從 API key 驗證開始——一份憑證就管住 registry 裡所有模型的存取，消掉 N 把 key × M 個 provider 的混亂擴散。
它取代的反模式
每個呼叫點都寫死 model 字串。模型被棄用時，得對整個 codebase 做一次搜尋取代——而漏掉的第一個跡象，就是在生產環境出現 404 錯誤。
模式二：Strategy——可插拔的 Prompt 範本
問題
prompt 字串內嵌在 business logic 裡。要改 checkout 的語氣，得找出散落在 checkout、support、onboarding 程式裡的每一個 "You are a helpful shopping assistant..."。要 A/B 測試兩個 prompt 變體，就得在每個呼叫點寫 if/else 義大利麵。
模式
一個 PromptStrategy 介面。每個 use case 或實驗變體都有對應的具體實作。執行時期靠 feature flag 或 A/B 測試分桶來選擇。每個 strategy 都是帶版本的產物——你的 prompt registry 把版本對應到 strategy class。
from abc import ABC, abstractmethod
class PromptStrategy(ABC):
version: str
@abstractmethod
def build_messages(self, context: dict) -> list[dict]:
"""Build the messages array for this prompt strategy."""
class CheckoutV3(PromptStrategy):
version = "checkout_v3.2"
def build_messages(self, context: dict) -> list[dict]:
return [
{"role": "system", "content": CHECKOUT_SYSTEM_V3},
{"role": "user", "content": f"<cart>{context['cart']}</cart>"}
]
class PromptRouter:
def __init__(self, strategies: dict[str, PromptStrategy]):
self.strategies = strategies
def select(self, feature_flags: dict, task: str) -> PromptStrategy:
variant = feature_flags.get(f"prompt_{task}", "default")
return self.strategies[variant]當你對 checkout prompt V3 和 V4 做 A/B 測試，只要切一個 feature flag。零程式變更。eval 套件（見我們的測試指南）量出哪個變體勝出。
它取代的反模式
prompt 字串散落整個 business logic。改語氣，就得找出每一份複製貼上的變體。不翻部署日誌的話，根本無從知道使用者收到的是哪個版本的 prompt。
模式三：Observer——解耦的串流消費者
問題
你的串流迴圈把 TTS 合成、UI chunk 渲染、成本追蹤、日誌全部纏在一起。要加新的消費者——analytics、翻譯浮層、稽核記錄——就得改核心的生成迴圈。加三次之後，迴圈變成 200 行，沒人想碰它。
模式
一個 StreamObserver 介面。每個消費者都有具體的 observer。generator 通知 observer——但不知道它們做什麼。鬆散耦合。observer 可以獨立新增、移除或替換。
class StreamObserver(ABC):
@abstractmethod
async def on_token(self, token: str, sequence: int): ...
@abstractmethod
async def on_complete(self, full_response: str, usage: dict): ...
@abstractmethod
async def on_error(self, error: Exception): ...
class StreamObservable:
def __init__(self, client: AsyncOpenAI):
self.client = client
self.observers: list[StreamObserver] = []
def attach(self, observer: StreamObserver): self.observers.append(observer)
async def stream(self, **kwargs):
stream = await self.client.chat.completions.create(stream=True, **kwargs)
full_response = ""
try:
async for chunk in stream:
token = chunk.choices[0].delta.content or ""
full_response += token
await asyncio.gather(*[
o.on_token(token, len(full_response)) for o in self.observers
])
await asyncio.gather(*[
o.on_complete(full_response, usage) for o in self.observers
])
except Exception as e:
await asyncio.gather(*[o.on_error(e) for o in self.observers])一個 observer 當掉不會弄死串流——錯誤按 observer 各自隔離。加一個 CostTracker observer。再加一個 TTSOutput observer。誰都不知道另一個的存在。
它取代的反模式
所有串流消費者邏輯都內嵌在生成迴圈裡。要加 analytics 儀表，得去改處理 TTS 的同一個函式——還可能因為打錯一個變數名稱，害音訊輸出出迴歸。
模式四：Decorator——不雜亂的營運層
問題
一個 10 行的 LLM 呼叫，被 60 行的 retry 邏輯、成本追蹤、結構化日誌、錯誤處理團團圍住。在八個呼叫點之間，用稍微不同的參數複製貼上。
模式
分層的 decorator 包住核心的 LLM 呼叫。每個 decorator 只有單一職責。針對不同呼叫點，組合成不同組合。
@with_retry(max_retries=3, backoff="exponential", retry_on=[429, 503])
@with_cost_tracking(budget_per_call=5.00)
@with_structured_logging(log_level="DEBUG")
async def core_llm_call(client, model_spec, messages):
return await client.chat.completions.create(
model=model_spec.model_id, messages=messages
)retry decorator 用 exponential backoff 與 jitter 處理暫時性錯誤——不該重試的錯誤類型（400、401、403）直接放行。rate limiting 的機制與完整的 429 處理架構，我們在rate limit 處理指南講過了——這個模式把那份邏輯封裝起來，好讓每個呼叫點一致套用。成本追蹤 decorator 記錄 gen_ai.usage，單次呼叫成本超過預算就發警報。兩個 decorator 誰都不知道另一個的存在。堆疊順序很重要：retry 放最外層（這樣重試失敗也還是會被追蹤成本），日誌放最內層（這樣它才看得到最終回應）。
這個模式示範怎麼封裝 rate limit 處理邏輯，讓它在每個呼叫點被一致套用。同樣的封裝原則也適用於 prompt caching——@with_cache decorator 在重複或相似請求造成 API 呼叫之前，先把它攔下來。TokSpan 的prompt caching 文件涵蓋這個 decorator 包住的 API 層級快取機制。
它取代的反模式
營運樣板程式碼複製貼上在每個 LLM 呼叫周圍。retry 參數不一致。八個呼叫點裡有三個漏掉成本追蹤。沒人知道哪種日誌格式是「正確」的，因為每個呼叫點做得都稍微不同。
模式五：Chain of Responsibility——Fallback 管線
問題
模型 failover 寫死在巢狀的 try/except 區塊裡。try gpt-5.5 —except: try claude-sonnet —except: try deepseek —except: return error。要加一個 fallback 模型或改 chain 順序，就得整個區塊重寫。每個呼叫點的 chain 都稍微不同。
模式
一串 ModelHandler 物件。每個 handler 知道自己的模型和怎麼處理請求。如果失敗了——非暫時性錯誤、timeout、品質低於門檻——就交給下一個 handler。chain 的組成活在 config 裡，不在程式裡。
class ModelHandler(ABC):
def __init__(self, model_spec: ModelSpec):
self.model_spec = model_spec
self._next: ModelHandler | None = None
def set_next(self, handler: "ModelHandler") -> "ModelHandler":
self._next = handler
return handler
async def handle(self, request: dict) -> dict | None:
try:
result = await self._call_model(request)
if self._quality_check(result):
return result
except NonRetryableError:
pass
if self._next:
return await self._next.handle(request)
return None
class FallbackChain:
def __init__(self):
self.head: ModelHandler | None = None
self.circuit_breaker: dict[str, int] = {} # model_id —consecutive failures
async def execute(self, request: dict) -> dict:
if not self.head:
raise RuntimeError("Empty fallback chain")
return await self.head.handle(request)某個 handler 連續三次失敗——circuit breaker 會暫時把它移出 chain。冷卻期過後，會用一個測試請求把它加回去。我們的多模型架構指南深入講過路由策略——這個模式提供正式化的 chain 實作。
它取代的反模式
巢狀 try/except 的 fallback 邏輯在呼叫點之間複製貼上。chain 順序不一致。沒有 circuit breaker——排在第二位置的劣化模型，永遠成功不了，卻讓每個 fallback 都多一層延遲。
模式六：Template Method——標準化的 Agent 迴圈
問題
每個 agent 的 tool-calling 迴圈都稍微不同。有的用 while True。有的用 for i in range(max_iterations)。有的根本忘了迴圈上限。agent 之間行為不一致。能無限迴圈的那個 agent，有成本失控的風險。
模式
一個帶固定骨架的 AgentLoop template method：plan——執行工具——observe——決定下一步。子類別覆寫 hook 方法來做自訂行為。骨架保證每個 agent 都繼承相同的安全特性——迴圈上限、timeout、成本上限、結構化錯誤處理。
class AgentLoop(ABC):
def __init__(self, max_iterations: int = 15, timeout: float = 120.0, cost_cap: float = 5.00):
self.max_iterations = max_iterations
self.timeout = timeout
self.cost_cap = cost_cap
async def run(self, task: str) -> dict:
start = time.time()
total_cost = 0.0
for i in range(self.max_iterations):
if time.time() - start > self.timeout:
return {"status": "timeout", "partial_result": self._build_partial()}
if total_cost > self.cost_cap:
return {"status": "cost_cap_exceeded"}
plan = await self.plan(task) # Hook: override
tool = await self.select_tool(plan) # Hook: override
result = await self.execute(tool) # Hook: override
total_cost += result.get("cost", 0)
if await self.should_stop(i, result): # Hook: override
return await self.synthesize()
@abstractmethod
async def plan(self, task: str) -> dict: ...
@abstractmethod
async def select_tool(self, plan: dict) -> str: ...
@abstractmethod
async def execute(self, tool: str) -> dict: ...我們的單一 agent 指南講過 tool-calling 迴圈的基礎。這個模式提供設計模式的角度：一個正式化的範本，讓安全保證是結構性的，而不是理想喊話。
它取代的反模式
每個 agent 都自己實作迴圈。安全防護不一致。那個能無限迴圈的 agent，是因為有人複製了 “while True” 版本，卻沒帶 max_iterations 檢查。
速查表：什麼時候用哪個模式？
| 你的情況 | 建議採用 |
|---|---|
| 你的 codebase 中有 3 個以上的 model 字串 | Factory —集中管理 model 選擇 |
| 用 if/else 實作的 Prompt A/B 測試 | Strategy —封裝 prompt 變體 |
| 與生成程式碼耦合的 Streaming 消費者 | Observer —用事件驅動設計解耦 |
| 每個 LLM 呼叫周遭的 60 行樣板程式碼 | Decorator —分層處理營運關注點 |
| 為 model 容錯切換而做的巢狀 try/except | Chain of Responsibility —可設定的備援機制 |
| 多個迴圈不一致的 agent | Template Method —用安全防護標準化 |
依 codebase 規模的導入順序： 小（<5K 行，1–2 個 use case）——從 Decorator 和 Factory 開始。中（5–50K 行）——加 Strategy 和 Chain of Responsibility。大（50K 行以上，多個 agent）——加 Observer 和 Template Method。
六個模式都能搭配標準的 OpenAI 相容 SDK。統一的 API endpoint 意味著 Factory 的 config 是 1 個 base_url 加 N 個 model 字串——不是 N 個 base URL × M 個 provider。
常見問題
這些模式不會把簡單的 API 呼叫過度設計嗎？
如果你的 codebase 只有一個 LLM 呼叫、而且不會長超過兩個，對——直接寫 50 行的 client.chat.completions.create() 就是正解。當 codebase 長到 10+ 個 LLM 呼叫、3+ 個模型變體、還有生產環境的可靠度要求時，這些模式的 ROI 在第一個事件就浮現——第一次明明改個 config 就好的模型遷移、第一次因為漏了迴圈上限的成本失控、第一次沒有 rollback 路徑的 prompt 迴歸。
該先實作哪個模式？
Decorator。它疊在現有的 LLM 呼叫上，不必修改它們。一疊 decorator——retry、日誌、成本追蹤——套到每個呼叫點。立刻得到生產環境的可靠度提升。現有程式零重構。接著是 Factory——下次要換模型時，你改一個 config 值就好，不用改 15 個檔案。
這些模式能搭配 LangChain 或 LlamaIndex 用嗎？
它們可以共存。Factory 和 Strategy 在 LangChain 外面運作得更乾淨——它們防止模型選擇和 prompt 管理被框架鎖死。Observer 和 Template Method 可以活在 LangChain agent 裡面——迴圈結構和串流消費者能受惠於框架整合。這些模式不取代 LangChain。它們在圍繞你選的框架去結構化程式。
當模型分屬不同 provider API 時，這些模式表現如何？
模式會變得更好實作，不是更複雜。Factory：一個 client 實例就涵蓋所有模型——你的 config 是 1 個 base_url 加 N 個 model 字串，不是 N 個 base URL × M 個 provider。Chain of Responsibility：透過一個整合點，跨 provider 做 fallback。Decorator：因為所有呼叫都走同一個 gateway，成本追蹤保持一致。模式本身是 provider-agnostic 的。統一的 endpoint 縮小每個模式要管理的整合表面——這正是抽象化的重點。想更宏觀理解為什麼單一 endpoint 架構正成為業界預設，我們的AI API 聚合平台轉移分析講了這個趨勢背後的操作面與成本面驅動因素。
除了 GoF，有 LLM 專屬的模式嗎？
有。Semantic Router——用查詢的語義路由，而不是寫死的規則。Guard——在每個 LLM 呼叫前後執行的輸入／輸出驗證管線。Cache-Aside——在發出 API 呼叫前檢查 embedding 相似度的語義快取層。這些是 LLM 原生模式，值得各自一篇專文。這裡列的六個是刻意挑的：多數工程團隊本來就懂 GoF 模式。把它們對應到 LLM API，學習曲線幾乎降到零。
設計模式不是為了炫技。是為了不要因為程式被複製貼上而非結構化，讓同一個 bug 出現在八個地方。
從 Decorator 開始。加上 Factory。下一次模型遷移只要 30 秒——不是一個早上的搜尋取代，加一個下午去除錯那個你漏掉的呼叫點。
把這份參考存到書籤。下次你發現自己第四次在複製貼上 retry 邏輯時，會知道該打開哪個抽屜。想獲得更多生產模式與 LLM API 架構指南，讓你的 codebase 在擴張時保持結構，訂閱我們的部落格。