The pull request is titled “Add GPT-5.5 fallback.” The diff: 340 lines for what should be twenty —copy-pasted retry decorators, hardcoded model strings, five near-identical call_llm_with_retry variants.
You review: “We need abstraction here.” The author: “What abstraction, specifically?”
This article answers that question.
Six Gang of Four patterns translated to the LLM API domain: Factory for model selection, Strategy for prompts, Observer for streaming, Decorator for retry and logging, Chain of Responsibility for fallback, Template Method for agent loops —each with production Python code and the anti-pattern it replaces.
Pattern 1: Factory —Centralized Model Instantiation
The Problem
"gpt-5.5" in chat.py. "claude-sonnet-4-20250514" in summarizer.py. "deepseek-v4-flash" in classifier.py. Model migration means find-and-replace across your entire codebase —and hoping you didn’t miss one in a config file that only loads in production.
The Pattern
A ModelFactory with a centralized registry. Task types, not hardcoded strings, determine which model is used. Environment variables enable canary deployments and rollbacks. Model metadata —capabilities, cost tier, context window —lives alongside the model ID. The implementation uses the OpenAI Python SDK —the standard client library for OpenAI-compatible APIs —whose AsyncOpenAI client powers the Factory below.
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])A unified API endpoint —one base_url for all providers —shrinks the Factory’s config from O(N providers) to O(1 endpoint + N model strings). One API key. One client instance. Every model in the registry accessible through it. Setting up this single-entry-point architecture starts with API key authentication —one credential that gates access to every model in your registry, eliminating the N-keys-times-M-providers sprawl.
The Anti-Pattern It Replaces
Model strings hardcoded at every call site. A model deprecation triggers a codebase-wide search and replace —and the first sign you missed one is a 404 error in production.
Pattern 2: Strategy —Pluggable Prompt Templates
The Problem
Prompt strings inlined in business logic. Changing the checkout tone means finding every "You are a helpful shopping assistant..." scattered across checkout, support, and onboarding code. A/B testing two prompt variants means if/else spaghetti at every call site.
The Pattern
A PromptStrategy interface. Concrete implementations per use case or experiment variant. Runtime selection by feature flag or A/B test bucket. Each strategy is a versioned artifact —your prompt registry maps versions to strategy classes.
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]When you A/B test checkout prompt V3 against V4, you toggle a feature flag. Zero code changes. The eval suite (see our testing guide) measures which variant wins.
The Anti-Pattern It Replaces
Prompt strings scattered across business logic. Changing tone means finding every copy-pasted variant. There’s no way to know which version of the prompt a user received without tracing through deployment logs.
Pattern 3: Observer —Decoupled Streaming Consumers
The Problem
Your streaming loop has TTS synthesis, UI chunk rendering, cost tracking, and logging all tangled together. Adding a new consumer —analytics, translation overlay, audit recording —means modifying the core generation loop. After three additions, the loop is 200 lines and nobody wants to touch it.
The Pattern
A StreamObserver interface. Concrete observers for each consumer. The generator notifies observers —but doesn’t know what they do. Loose coupling. Observers can be added, removed, or replaced independently.
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])One observer crashing doesn’t kill the stream —errors are isolated per observer. Add a CostTracker observer. Add a TTSOutput observer. Neither knows the other exists.
The Anti-Pattern It Replaces
All streaming consumer logic inlined in the generation loop. Adding analytics instrumentation requires editing the same function that handles TTS —and risking a regression in audio output because you fat-fingered a variable name.
Pattern 4: Decorator —Operational Layers Without Clutter
The Problem
A 10-line LLM call surrounded by 60 lines of retry logic, cost tracking, structured logging, and error handling. Copy-pasted with slightly different parameters across eight call sites.
The Pattern
Layered decorators wrapping the core LLM call. Each decorator has one responsibility. They compose into different combinations for different call sites.
@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
)The retry decorator handles transient errors with exponential backoff and jitter —error types that should not be retried (400, 401, 403) are passed through immediately. Rate limiting mechanics and the full 429 handling architecture are covered in our rate limit handling guide —this pattern encapsulates that logic for consistent application across call sites. The cost tracking decorator logs gen_ai.usage and alerts if per-call cost exceeds budget. Neither decorator knows about the other. Stack order matters: retry outermost (so failed retries still get cost-tracked), logging innermost (so it sees the final response).
This pattern shows how to encapsulate rate-limit handling logic so it’s applied consistently across every call site. The same encapsulation principle applies to prompt caching —a @with_cache decorator intercepts repeated or similar requests before they incur an API call. TokSpan’s prompt caching documentation covers the API-level caching mechanics the decorator wraps.
The Anti-Pattern It Replaces
Operational boilerplate copy-pasted around every LLM call. Inconsistent retry parameters. Missing cost tracking on three of eight call sites. Nobody knows which logging format is “correct” because every call site does it slightly differently.
Pattern 5: Chain of Responsibility —Fallback Pipelines
The Problem
Model failover hardcoded in nested try/except blocks. try gpt-5.5 —except: try claude-sonnet —except: try deepseek —except: return error. Adding a fallback model or changing the chain order means rewriting the entire block. Every call site has a slightly different chain.
The Pattern
A chain of ModelHandler objects. Each handler knows its model and how to process a request. If it fails —non-transient error, timeout, quality below threshold —it passes to the next handler. Chain composition lives in config, not 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)Three consecutive failures on a handler —circuit breaker temporarily removes it from the chain. It gets re-added after a cooldown period with a test request. Our multi-model architecture guide covers the routing strategies in depth —this pattern provides the formalized chain implementation.
The Anti-Pattern It Replaces
Nested try/except fallback logic copy-pasted across call sites. Inconsistent chain order. No circuit breaker —a degraded model in position two adds latency to every fallback without ever succeeding.
Pattern 6: Template Method —Standardized Agent Loop
The Problem
Every agent has a slightly different tool-calling loop. Some use while True. Some use for i in range(max_iterations). Some forgot the loop limit entirely. Inconsistent behavior between agents. Cost runaway risk on the agent that can loop indefinitely.
The Pattern
An AgentLoop template method with a fixed skeleton: plan —execute tool —observe —decide next. Subclasses override hook methods for custom behavior. The skeleton guarantees every agent inherits the same safety characteristics —loop limit, timeout, cost cap, structured error handling.
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: ...Our single agent guide covers the tool-calling loop fundamentals. This pattern provides the design-pattern perspective: a formalized template that makes safety guarantees structural, not aspirational.
The Anti-Pattern It Replaces
Each agent implements its own loop. Inconsistent safety guards. The agent that can loop forever because someone copied the “while True” version without the max_iterations check.
Quick Reference: Which Pattern When?
| You Have… | Reach For… |
|---|---|
| >3 model strings in your codebase | Factory —centralize model selection |
| Prompt A/B tests implemented with if/else | Strategy —encapsulate prompt variants |
| Streaming consumers coupled to generation code | Observer —decouple with event-driven design |
| 60 lines of boilerplate around every LLM call | Decorator —layer operational concerns |
| Nested try/except for model failover | Chain of Responsibility —configurable fallback |
| Multiple agents with inconsistent loops | Template Method —standardize with safety guards |
Adoption order by codebase size: Small (<5K lines, 1-2 use cases) —start with Decorator and Factory. Medium (5-50K lines) —add Strategy and Chain of Responsibility. Large (50K+ lines, multiple agents) —add Observer and Template Method.
All six patterns work with standard OpenAI-compatible SDKs. A unified API endpoint means the Factory’s config is one base_url and N model strings —not N base URLs times M providers.
FAQ
Won’t these patterns over-engineer a simple API call?
If your codebase has one LLM call and won’t grow beyond two, yes —50 lines of direct client.chat.completions.create() is the right answer. When your codebase reaches 10+ LLM calls, 3+ model variants, and production reliability requirements, these patterns’ ROI materializes at the first incident —the first model migration that should have been a config change, the first cost runaway from a missing loop limit, the first prompt regression with no rollback path.
Which pattern should I implement first?
Decorator. It layers onto existing LLM calls without modifying them. One decorator stack —retry, logging, cost tracking —applied to every call site. Immediate production reliability gain. Zero refactoring of existing code. Factory second —when you next need to switch models, you’ll change one config value instead of 15 files.
Do these patterns work with LangChain or LlamaIndex?
They coexist. Factory and Strategy work cleaner outside LangChain —they prevent framework lock-in for model selection and prompt management. Observer and Template Method can live inside LangChain agents —the loop structure and streaming consumers benefit from framework integration. These patterns don’t replace LangChain. They structure the code around whatever framework you choose.
How do the patterns hold up when models live behind different provider APIs?
The patterns become simpler to implement, not more complex. Factory: one client instance covers every model —your config is one base_url and N model strings, not N base URLs times M providers. Chain of Responsibility: fallback across providers through one integration point. Decorator: consistent cost tracking because all calls flow through the same gateway. The patterns themselves are provider-agnostic. A unified endpoint reduces the integration surface each pattern has to manage —which is the whole point of abstraction. For a broader perspective on why the single-endpoint architecture is becoming the default across the industry, our analysis of the shift to AI API aggregation platforms covers the operational and cost drivers behind the trend.
Are there LLM-specific patterns beyond GoF?
Yes. Semantic Router —route by query semantics, not hardcoded rules. Guard —input/output validation pipeline that runs before and after every LLM call. Cache-Aside —semantic caching layer that checks embedding similarity before making an API call. These are LLM-native patterns worth their own dedicated article. The six here are deliberate: most engineering teams already know GoF patterns. Mapping them to LLM APIs reduces the learning curve to near zero.
Design patterns aren’t about sophistication. They’re about not having the same bug in eight places because the code was copy-pasted instead of structured.
Start with Decorator. Add Factory. The next model migration will take 30 seconds —not a morning of find-and-replace and an afternoon of debugging the one call site you missed.
Bookmark this reference. Next time you catch yourself copy-pasting retry logic for the fourth time, you’ll know which drawer to open. For more production patterns and LLM API architecture guides that keep your codebase structured as you scale, subscribe to our blog.