Pull request có tiêu đề “Add GPT-5.5 fallback.” Diff: 340 dòng cho thứ lẽ ra phải là hai mươi —decorator retry copy-paste, chuỗi model hardcode, năm biến thể call_llm_with_retry gần như giống hệt.
Bạn review: “Chúng ta cần abstraction ở đây.” Tác giả: “Abstraction cụ thể nào?”
Bài viết này trả lời câu hỏi đó.
Sáu mẫu Gang of Four dịch sang miền LLM API: Factory cho chọn model, Strategy cho prompt, Observer cho streaming, Decorator cho retry và logging, Chain of Responsibility cho fallback, Template Method cho agent loop —mỗi cái kèm code Python production và anti-pattern nó thay thế.
Pattern 1: Factory —Khởi tạo model tập trung
Vấn đề
"gpt-5.5" trong chat.py. "claude-sonnet-4-20250514" trong summarizer.py. "deepseek-v4-flash" trong classifier.py. Di trú model nghĩa là tìm-và-thay trên toàn bộ codebase —và hy vọng bạn không sót một cái trong file cấu hình chỉ nạp lúc production.
Pattern
Một ModelFactory với registry tập trung. Loại tác vụ, không phải chuỗi hardcode, quyết định model nào được dùng. Biến môi trường kích hoạt canary deployment và rollback. Metadata model —capabilities, hạng chi phí, context window— sống cạnh ID model. Triển khai dùng OpenAI Python SDK —thư viện client chuẩn cho API tương thích OpenAI— có client AsyncOpenAI tiếp sức cho Factory bên dưới.
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])Một endpoint API thống nhất —một base_url cho mọi nhà cung cấp— thu nhỏ cấu hình Factory từ O(N nhà cung cấp) xuống O(1 endpoint + N chuỗi model). Một API key. Một instance client. Mọi model trong registry truy cập qua nó. Thiết lập kiến trúc điểm vào đơn lẻ này bắt đầu với xác thực API key —một chứng chỉ kiểm soát truy cập đến mọi model trong registry của bạn, loại bỏ sự phân tán N keys nhân M nhà cung cấp.
Anti-pattern Nó Thay Thế
Chuỗi model hardcode ở mọi điểm gọi. Model bị ngưng hỗ trợ kích hoạt tìm-và-thay trên toàn codebase —và dấu hiệu đầu tiên bạn sót một cái là lỗi 404 trong production.
Pattern 2: Strategy —Prompt template cắm được
Vấn đề
Chuỗi prompt inline trong logic nghiệp vụ. Đổi giọng checkout nghĩa là tìm mọi "You are a helpful shopping assistant..." rải rác trong code checkout, support và onboarding. A/B test hai biến thể prompt nghĩa là mì spaghetti if/else ở mọi điểm gọi.
Pattern
Một interface PromptStrategy. Triển khai cụ thể theo từng use case hoặc biến thể thử nghiệm. Chọn lựa runtime theo feature flag hoặc bucket A/B test. Mỗi strategy là một artifact phiên bản —registry prompt của bạn ánh xạ phiên bản sang class strategy.
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]Khi bạn A/B test prompt checkout V3 so với V4, bạn bật một feature flag. Không thay đổi code. Bộ đánh giá (xem hướng dẫn testing của chúng tôi) đo biến thể nào thắng.
Anti-pattern Nó Thay Thế
Chuỗi prompt rải rác trong logic nghiệp vụ. Đổi giọng nghĩa là tìm mọi biến thể copy-paste. Không có cách nào biết người dùng nhận phiên bản prompt nào mà không lần theo log triển khai.
Pattern 3: Observer —Streaming consumer tách rời
Vấn đề
Vòng lặp streaming của bạn có tổng hợp TTS, render chunk UI, theo dõi chi phí và logging đan xen với nhau. Thêm consumer mới —analytics, overlay dịch, ghi âm kiểm toán— nghĩa là phải sửa vòng lặp sinh lõi. Sau ba lần thêm, vòng lặp thành 200 dòng và không ai muốn động vào.
Pattern
Một interface StreamObserver. Observer cụ thể cho từng consumer. Generator thông báo cho observer —nhưng không biết chúng làm gì. Khớp nối lỏng. Observer có thể thêm, xóa, hoặc thay độc lập.
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])Một observer sập không giết stream —lỗi cô lập theo từng observer. Thêm observer CostTracker. Thêm TTSOutput. Không cái nào biết cái kia tồn tại.
Anti-pattern Nó Thay Thế
Mọi logic streaming consumer inline trong vòng lặp sinh. Thêm instrument analytics đòi sửa chính hàm xử lý TTS —và rủi ro regression đầu ra âm thanh vì bạn gõ sai tên biến.
Pattern 4: Decorator —Lớp vận hành gọn gàng
Vấn đề
Một lời gọi LLM 10 dòng bao quanh bởi 60 dòng logic retry, theo dõi chi phí, logging có cấu trúc và xử lý lỗi. Copy-paste với tham số hơi khác qua tám điểm gọi.
Pattern
Các decorator phân lớp bọc lời gọi LLM lõi. Mỗi decorator một trách nhiệm. Chúng kết hợp thành nhiều tổ hợp khác nhau cho các điểm gọi khác nhau.
@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
)Decorator retry xử lý lỗi tạm thời bằng exponential backoff và jitter —loại lỗi không nên retry (400, 401, 403) được chuyển qua ngay. Cơ chế rate limiting và toàn bộ kiến trúc xử lý 429 được trình bày trong hướng dẫn xử lý rate limit của chúng tôi —pattern này đóng gói logic đó để áp dụng nhất quán trên mọi điểm gọi. Decorator theo dõi chi phí ghi gen_ai.usage và cảnh báo nếu chi phí mỗi lời gọi vượt ngân sách. Không decorator nào biết về cái kia. Thứ tự stack quan trọng: retry ngoài cùng (để retry thất bại vẫn được theo dõi chi phí), logging trong cùng (để nó thấy phản hồi cuối).
Pattern này cho thấy cách đóng gói logic xử lý rate limit để áp dụng nhất quán trên mọi điểm gọi. Nguyên tắc đóng gói tương tự áp dụng cho prompt caching —decorator @with_cache chặn các request lặp lại hoặc tương tự trước khi chúng gây lời gọi API. Tài liệu prompt caching của TokSpan bao phủ cơ chế caching cấp API mà decorator bọc.
Anti-pattern Nó Thay Thế
Boilerplate vận hành copy-paste quanh mọi lời gọi LLM. Tham số retry không nhất quán. Thiếu theo dõi chi phí ở ba trong tám điểm gọi. Không ai biết định dạng logging nào “đúng” vì mỗi điểm gọi làm hơi khác.
Pattern 5: Chain of Responsibility —Pipeline fallback
Vấn đề
Failover model hardcode trong khối try/except lồng nhau. try gpt-5.5 —except: try claude-sonnet —except: try deepseek —except: return error. Thêm model fallback hoặc đổi thứ tự chuỗi nghĩa là viết lại cả khối. Mỗi điểm gọi có một chuỗi hơi khác.
Pattern
Một chuỗi đối tượng ModelHandler. Mỗi handler biết model của nó và cách xử lý request. Nếu nó thất bại —lỗi không tạm thời, timeout, chất lượng dưới ngưỡng— nó chuyển sang handler kế tiếp. Thành phần chuỗi nằm trong config, không phải code.
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)Ba lần thất bại liên tiếp trên một handler —circuit breaker tạm thời gỡ nó khỏi chuỗi. Nó được thêm lại sau thời gian hạ nhiệt với một request kiểm thử. Hướng dẫn kiến trúc đa model của chúng tôi trình bày chiến lược định tuyến sâu —pattern này cung cấp triển khai chuỗi chính thức hóa.
Anti-pattern Nó Thay Thế
Logic fallback try/except lồng nhau copy-paste giữa các điểm gọi. Thứ tự chuỗi không nhất quán. Không circuit breaker —model xuống cấp ở vị trí hai thêm độ trễ vào mọi fallback mà không bao giờ thành công.
Pattern 6: Template Method —Agent loop chuẩn hóa
Vấn đề
Mỗi agent có vòng lặp gọi công cụ hơi khác. Vài cái dùng while True. Vài cái dùng for i in range(max_iterations). Vài cái quên giới hạn vòng lặp hoàn toàn. Hành vi không nhất quán giữa các agent. Rủi ro chi phí chạy trốn trên agent có thể lặp vô hạn.
Pattern
Một template method AgentLoop với khung cố định: plan —thực thi công cụ —quan sát —quyết định bước tiếp. Subclass override phương thức hook cho hành vi tùy chỉnh. Khung đảm bảo mỗi agent kế thừa cùng đặc tính an toàn —giới hạn vòng lặp, timeout, trần chi phí, xử lý lỗi có cấu trúc.
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: ...Hướng dẫn agent đơn lẻ của chúng tôi trình bày nền tảng vòng lặp gọi công cụ. Pattern này cho góc nhìn design pattern: một template chính thức hóa biến đảm bảo an toàn thành cấu trúc, không phải nguyện vọng.
Anti-pattern Nó Thay Thế
Mỗi agent triển khai vòng lặp riêng. Lớp bảo vệ an toàn không nhất quán. Agent có thể lặp mãi vì ai đó copy phiên bản “while True” mà không có kiểm tra max_iterations.
Tham chiếu nhanh: Dùng pattern nào khi nào?
| Bạn có… | Nên sử dụng… |
|---|---|
| Hơn 3 chuỗi model trong codebase của bạn | Factory —tập trung hóa việc chọn model |
| Bài kiểm tra A/B prompt triển khai bằng if/else | Strategy —đóng gói các biến thể prompt |
| Consumer streaming gắn liền với code tạo nội dung | Observer —tách rời bằng thiết kế hướng sự kiện |
| 60 dòng boilerplate quanh mỗi lệnh gọi LLM | Decorator —phân lớp các mối quan tâm vận hành |
| try/except lồng nhau cho việc chuyển đổi dự phòng model | Chain of Responsibility —dự phòng có thể cấu hình |
| Nhiều agent có vòng lặp không nhất quán | Template Method —chuẩn hóa bằng lớp bảo vệ an toàn |
Thứ tự áp dụng theo quy mô codebase: Nhỏ (<5K dòng, 1-2 use case) —bắt đầu với Decorator và Factory. Trung bình (5-50K dòng) —thêm Strategy và Chain of Responsibility. Lớn (50K+ dòng, nhiều agent) —thêm Observer và Template Method.
Cả sáu pattern hoạt động với SDK tương thích OpenAI chuẩn. Một endpoint API thống nhất nghĩa là cấu hình Factory chỉ là một base_url và N chuỗi model —không phải N base URL nhân M nhà cung cấp.
Câu hỏi thường gặp
Chẳng phải các pattern này over-engineer một lời gọi API đơn giản sao?
Nếu codebase của bạn có một lời gọi LLM và sẽ không vượt quá hai, đúng —50 dòng client.chat.completions.create() trực tiếp là câu trả lời đúng. Khi codebase của bạn đạt 10+ lời gọi LLM, 3+ biến thể model và yêu cầu độ tin cậy production, ROI của các pattern này hiện ra ở sự cố đầu tiên —cuộc di trú model đầu tiên lẽ ra là thay đổi config, chi phí chạy trốn đầu tiên từ thiếu giới hạn vòng lặp, regression prompt đầu tiên không có đường rollback.
Pattern nào tôi nên triển khai trước?
Decorator. Nó xếp lớp lên lời gọi LLM hiện có mà không sửa chúng. Một stack decorator —retry, logging, theo dõi chi phí— áp dụng cho mọi điểm gọi. Tăng độ tin cậy production ngay lập tức. Không refactor code hiện có. Factory thứ hai —khi bạn cần đổi model lần tới, bạn đổi một giá trị config thay vì 15 file.
Các pattern này có hoạt động với LangChain hoặc LlamaIndex không?
Chúng cùng tồn tại. Factory và Strategy hoạt động sạch hơn ngoài LangChain —chúng ngăn khóa framework cho chọn model và quản lý prompt. Observer và Template Method có thể sống trong agent LangChain —cấu trúc vòng lặp và streaming consumer hưởng lợi từ tích hợp framework. Các pattern này không thay thế LangChain. Chúng cấu trúc code quanh framework bạn chọn.
Các pattern giữ vững thế nào khi model nằm sau API nhà cung cấp khác?
Các pattern trở nên đơn giản hơn để triển khai, không phức tạp hơn. Factory: một instance client bao phủ mọi model —config của bạn là một base_url và N chuỗi model, không phải N base URL nhân M nhà cung cấp. Chain of Responsibility: fallback qua các nhà cung cấp qua một điểm tích hợp. Decorator: theo dõi chi phí nhất quán vì mọi lời gọi chảy qua cùng gateway. Bản thân các pattern không phụ thuộc nhà cung cấp. Một endpoint thống nhất giảm bề mặt tích hợp mỗi pattern phải quản lý —đó chính là toàn bộ điểm của abstraction. Cho góc nhìn rộng hơn về lý do kiến trúc endpoint đơn đang thành mặc định toàn ngành, phân tích của chúng tôi về sự chuyển dịch sang nền tảng tổng hợp AI API bao phủ các động lực vận hành và chi phí đằng sau xu hướng.
Có pattern riêng của LLM ngoài GoF không?
Có. Semantic Router —định tuyến theo ngữ nghĩa truy vấn, không phải quy tắc hardcode. Guard —pipeline xác thực đầu vào/đầu ra chạy trước và sau mỗi lời gọi LLM. Cache-Aside —lớp caching ngữ nghĩa kiểm tra độ tương đồng embedding trước khi gọi API. Đây là các pattern bản địa LLM đáng một bài viết riêng. Sáu pattern ở đây có chủ đích: hầu hết đội kỹ thuật đã biết pattern GoF. Ánh xạ chúng sang LLM API giảm đường cong học tập gần về không.
Design pattern không phải về sự tinh xảo. Chúng về việc không có cùng bug ở tám nơi vì code copy-paste thay vì được cấu trúc.
Bắt đầu với Decorator. Thêm Factory. Cuộc di trú model kế tiếp sẽ mất 30 giây —không phải một buổi sáng tìm-và-thay và một buổi chiều debug điểm gọi bạn bỏ sót.
Đánh dấu tài liệu tham chiếu này. Lần tới bạn bắt gặp mình copy-paste logic retry lần thứ tư, bạn sẽ biết mở ngăn kéo nào. Để biết thêm pattern production và hướng dẫn kiến trúc LLM API giữ codebase của bạn có cấu trúc khi mở rộng, hãy đăng ký blog của chúng tôi.