El pull request se titula “Agregar fallback a GPT-5.5.” El diff: 340 líneas para lo que deberían ser veinte —decoradores de retry copiados y pegados, strings de modelo hardcodeados, cinco variantes de call_llm_with_retry casi idénticas.
Revisas: “Necesitamos abstracción aquí.” El autor: “¿Qué abstracción, específicamente?”
Este artículo responde esa pregunta.
Seis patrones del Gang of Four traducidos al dominio de las LLM APIs: Factory para selección de modelos, Strategy para prompts, Observer para streaming, Decorator para retry y logging, Chain of Responsibility para fallback, Template Method para loops de agentes —cada uno con código Python de producción y el anti-patrón que reemplaza.
Patrón 1: Factory —Instanciación centralizada de modelos
El Problema
"gpt-5.5" en chat.py. "claude-sonnet-4-20250514" en summarizer.py. "deepseek-v4-flash" en classifier.py. La migración de modelos significa buscar-y-reemplazar en toda tu base de código —y esperar no haber dejado uno en un archivo de configuración que solo se carga en producción.
El Patrón
Un ModelFactory con un registro centralizado. Los tipos de tareas, no los strings hardcodeados, determinan qué modelo se usa. Las variables de entorno habilitan despliegues canary y rollbacks. Los metadatos del modelo —capacidades, nivel de costo, ventana de contexto— viven junto al ID del modelo. La implementación usa el OpenAI Python SDK —la biblioteca cliente estándar para APIs compatibles con OpenAI— cuyo cliente AsyncOpenAI impulsa el Factory que sigue.
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])Un endpoint de API unificado —un solo base_url para todos los proveedores— reduce la configuración del Factory de O(N proveedores) a O(1 endpoint + N strings de modelo). Una API key. Una instancia de cliente. Cada modelo del registro accesible a través de él. Configurar esta arquitectura de punto único de entrada comienza con autenticación de API key —una credencial que controla el acceso a cada modelo de tu registro, eliminando el desorden de N keys por M proveedores.
El Anti-Patrón Que Reemplaza
Strings de modelo hardcodeados en cada punto de llamada. Una deprecación de modelo dispara una búsqueda y reemplazo en toda la base de código —y la primera señal de que dejaste uno atrás es un error 404 en producción.
Patrón 2: Strategy —Plantillas de prompt conectables
El Problema
Strings de prompt inline en la lógica de negocio. Cambiar el tono del checkout significa encontrar cada "You are a helpful shopping assistant..." disperso por el código de checkout, soporte y onboarding. Probar A/B dos variantes de prompt significa spaghetti de if/else en cada punto de llamada.
El Patrón
Una interfaz PromptStrategy. Implementaciones concretas por caso de uso o variante de experimento. Selección en tiempo de ejecución por feature flag o bucket de prueba A/B. Cada estrategia es un artefacto versionado —tu registro de prompts mapea versiones a clases de estrategia.
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]Cuando haces A/B test del prompt de checkout V3 contra V4, activas un feature flag. Cero cambios de código. La suite de evaluación (ver nuestra guía de testing) mide qué variante gana.
El Anti-Patrón Que Reemplaza
Strings de prompt dispersos por la lógica de negocio. Cambiar el tono significa encontrar cada variante copiada y pegada. No hay forma de saber qué versión del prompt recibió un usuario sin rastrear los logs de despliegue.
Patrón 3: Observer —Consumidores de streaming desacoplados
El Problema
Tu loop de streaming tiene síntesis TTS, renderizado de chunks en la UI, tracking de costo y logging enredados juntos. Agregar un nuevo consumidor —analytics, overlay de traducción, registro de auditoría— significa modificar el loop de generación principal. Después de tres adiciones, el loop tiene 200 líneas y nadie quiere tocarlo.
El Patrón
Una interfaz StreamObserver. Observadores concretos para cada consumidor. El generador notifica a los observadores —pero no sabe qué hacen ellos. Acoplamiento débil. Los observadores pueden agregarse, eliminarse o reemplazarse de forma independiente.
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])Que un observador se caiga no mata el stream —los errores se aíslan por observador. Agrega un observador CostTracker. Agrega un TTSOutput. Ninguno sabe que el otro existe.
El Anti-Patrón Que Reemplaza
Toda la lógica de consumidores de streaming inline en el loop de generación. Agregar instrumentación de analytics requiere editar la misma función que maneja TTS —y arriesgar una regresión en la salida de audio porque tecleaste mal un nombre de variable.
Patrón 4: Decorator —Capas operativas sin desorden
El Problema
Una llamada LLM de 10 líneas rodeada de 60 líneas de lógica de retry, tracking de costo, logging estructurado y manejo de errores. Copiada y pegada con parámetros ligeramente diferentes en ocho puntos de llamada.
El Patrón
Decoradores en capas envolviendo la llamada LLM principal. Cada decorador tiene una responsabilidad. Se componen en diferentes combinaciones para diferentes puntos de llamada.
@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
)El decorador de retry maneja errores transitorios con backoff exponencial y jitter —los tipos de error que no deben reintentarse (400, 401, 403) pasan de inmediato. Los mecanismos de rate limiting y la arquitectura completa de manejo de 429 están cubiertos en nuestra guía de manejo de rate limits —este patrón encapsula esa lógica para una aplicación consistente en todos los puntos de llamada. El decorador de tracking de costo registra gen_ai.usage y alerta si el costo por llamada excede el presupuesto. Ningún decorador sabe del otro. El orden de la pila importa: retry más externo (para que los reintentos fallidos igual se trackeen en costo), logging más interno (para que vea la respuesta final).
Este patrón muestra cómo encapsular la lógica de manejo de rate limits para aplicarla consistentemente en cada punto de llamada. El mismo principio de encapsulación aplica al prompt caching —un decorador @with_cache intercepta requests repetidas o similares antes de incurrir en una llamada API. La documentación de prompt caching de TokSpan cubre los mecanismos de caching a nivel de API que envuelve el decorador.
El Anti-Patrón Que Reemplaza
Boilerplate operativo copiado y pegado alrededor de cada llamada LLM. Parámetros de retry inconsistentes. Tracking de costo ausente en tres de ocho puntos de llamada. Nadie sabe qué formato de logging es el “correcto” porque cada punto de llamada lo hace ligeramente distinto.
Patrón 5: Chain of Responsibility —Pipelines de fallback
El Problema
Failover de modelos hardcodeado en bloques try/except anidados. try gpt-5.5 —except: try claude-sonnet —except: try deepseek —except: return error. Agregar un modelo de fallback o cambiar el orden de la cadena significa reescribir todo el bloque. Cada punto de llamada tiene una cadena ligeramente diferente.
El Patrón
Una cadena de objetos ModelHandler. Cada handler conoce su modelo y cómo procesar un request. Si falla —error no transitorio, timeout, calidad por debajo del umbral— pasa al siguiente handler. La composición de la cadena vive en config, no en código.
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)Tres fallos consecutivos en un handler —el circuit breaker lo remueve temporalmente de la cadena. Se re-agrega después de un período de enfriamiento con un request de prueba. Nuestra guía de arquitectura multi-modelo cubre las estrategias de enrutamiento en profundidad —este patrón provee la implementación formalizada de la cadena.
El Anti-Patrón Que Reemplaza
Lógica de fallback con try/except anidados copiada y pegada entre puntos de llamada. Orden de cadena inconsistente. Sin circuit breaker —un modelo degradado en la posición dos agrega latencia a cada fallback sin nunca tener éxito.
Patrón 6: Template Method —Loop de agente estandarizado
El Problema
Cada agente tiene un loop de llamada de herramientas ligeramente diferente. Unos usan while True. Otros usan for i in range(max_iterations). Otros olvidaron el límite del loop por completo. Comportamiento inconsistente entre agentes. Riesgo de costo descontrolado en el agente que puede loopear indefinidamente.
El Patrón
Un método template AgentLoop con un esqueleto fijo: plan —ejecuta herramienta —observa —decide el siguiente paso. Las subclases sobrescriben métodos hook para comportamiento personalizado. El esqueleto garantiza que cada agente herede las mismas características de seguridad —límite de loop, timeout, tope de costo, manejo de errores estructurado.
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: ...Nuestra guía de agente único cubre los fundamentos del loop de llamada de herramientas. Este patrón provee la perspectiva de diseño: un template formalizado que hace las garantías de seguridad estructurales, no aspiracionales.
El Anti-Patrón Que Reemplaza
Cada agente implementa su propio loop. Guardias de seguridad inconsistentes. El agente que puede loopear para siempre porque alguien copió la versión “while True” sin el chequeo de max_iterations.
Referencia rápida: ¿Qué patrón usar cuándo?
| Tienes… | Usa… |
|---|---|
| >3 strings de modelo en tu base de código | Factory —centraliza la selección de modelo |
| Pruebas A/B de prompts implementadas con if/else | Strategy —encapsula las variantes de prompt |
| Consumidores de streaming acoplados al código de generación | Observer —desacopla con diseño basado en eventos |
| 60 líneas de código repetitivo alrededor de cada llamada LLM | Decorator —estratifica las preocupaciones operativas |
| try/except anidados para la conmutación por error de modelos | Chain of Responsibility —respaldo configurable |
| Múltiples agentes con loops inconsistentes | Template Method —estandariza con guardas de seguridad |
Orden de adopción por tamaño de base de código: Pequeña (<5K líneas, 1-2 casos de uso) —empieza con Decorator y Factory. Mediana (5-50K líneas) —agrega Strategy y Chain of Responsibility. Grande (50K+ líneas, múltiples agentes) —agrega Observer y Template Method.
Los seis patrones funcionan con SDKs estándar compatibles con OpenAI. Un endpoint de API unificado significa que la configuración del Factory es un base_url y N strings de modelo —no N base URLs por M proveedores.
FAQ
¿No sobre-ingenierizan estos patrones una llamada API simple?
Si tu base de código tiene una llamada LLM y no crecerá más allá de dos, sí —50 líneas de client.chat.completions.create() directo es la respuesta correcta. Cuando tu base de código llega a 10+ llamadas LLM, 3+ variantes de modelo y requisitos de confiabilidad de producción, el ROI de estos patrones se materializa en el primer incidente —la primera migración de modelo que debió ser un cambio de config, el primer costo descontrolado por un límite de loop ausente, la primera regresión de prompt sin camino de rollback.
¿Qué patrón debo implementar primero?
Decorator. Se superpone a las llamadas LLM existentes sin modificarlas. Una pila de decoradores —retry, logging, tracking de costo— aplicada a cada punto de llamada. Ganancia inmediata de confiabilidad en producción. Cero refactor de código existente. Factory segundo —cuando necesites cambiar de modelos, cambias un valor de config en lugar de 15 archivos.
¿Funcionan estos patrones con LangChain o LlamaIndex?
Coexisten. Factory y Strategy funcionan más limpios fuera de LangChain —previenen el lock-in de framework para la selección de modelos y la gestión de prompts. Observer y Template Method pueden vivir dentro de agentes LangChain —la estructura del loop y los consumidores de streaming se benefician de la integración con el framework. Estos patrones no reemplazan a LangChain. Estructuran el código alrededor del framework que elijas.
¿Cómo se sostienen los patrones cuando los modelos viven detrás de APIs de proveedores diferentes?
Los patrones se vuelven más simples de implementar, no más complejos. Factory: una instancia de cliente cubre cada modelo —tu configuración es un base_url y N strings de modelo, no N base URLs por M proveedores. Chain of Responsibility: fallback entre proveedores a través de un punto de integración. Decorator: tracking de costo consistente porque todas las llamadas fluyen por el mismo gateway. Los patrones en sí son agnósticos del proveedor. Un endpoint unificado reduce la superficie de integración que cada patrón tiene que manejar —que es el punto completo de la abstracción. Para una perspectiva más amplia de por qué la arquitectura de endpoint único se está volviendo el default en la industria, nuestro análisis del cambio a plataformas de agregación de AI APIs cubre los drivers operativos y de costo detrás de la tendencia.
¿Hay patrones específicos de LLM más allá de los GoF?
Sí. Semantic Router —enruta por semántica de la consulta, no por reglas hardcodeadas. Guard —pipeline de validación de entrada/salida que corre antes y después de cada llamada LLM. Cache-Aside —capa de caching semántico que verifica la similitud de embeddings antes de hacer una llamada API. Estos son patrones nativos de LLM que merecen su propio artículo dedicado. Los seis de aquí son deliberados: la mayoría de los equipos de ingeniería ya conocen los patrones GoF. Mapearlos a las LLM APIs reduce la curva de aprendizaje a casi cero.
Los patrones de diseño no se tratan de sofisticación. Se tratan de no tener el mismo bug en ocho lugares porque el código fue copiado y pegado en lugar de estructurado.
Empieza con Decorator. Agrega Factory. La próxima migración de modelo tomará 30 segundos —no una mañana de buscar-y-reemplazar y una tarde de depurar el punto de llamada que olvidaste.
Guarda esta referencia como favorito. La próxima vez que te sorprendas copiando lógica de retry por cuarta vez, sabrás qué cajón abrir. Para más patrones de producción y guías de arquitectura de LLM APIs que mantengan tu base de código estructurada mientras escalas, suscríbete a nuestro blog.