Módulo 8: Unified AI Client — Proyecto integrador final

Arquitectura y diseño

Siguiente →

Antes de teclear una sola línea de código, vamos a diseñar la interfaz. Esto no es perfectionism — es la diferencia entre implementar una vez vs. refactorear tres veces. Cada decisión que tomes acá te ahorra trabajo y te da consistencia entre los 5 adapters que vas a escribir.

Al terminar vas a poder:

  • Decidir el contrato de UnifiedClient y BaseAdapter con tipos claros
  • Diseñar el flujo de configuración (en código vs YAML vs env vars)
  • Mapear los patrones Adapter, Factory y Strategy a piezas concretas
  • Documentar decisiones de diseño con razones (no solo qué, sino por qué)

Las decisiones que vamos a tomar

Cinco decisiones grandes, en orden de impacto:

  1. ¿Qué métodos expone la interfaz pública? (mínimo + extensible)
  2. ¿Qué tipo de inputs/outputs? (string simple vs estructuras ricas)
  3. ¿Cómo se configura? (constructor params vs config object vs YAML)
  4. ¿Cómo se manejan errores? (excepciones tipadas vs Result wrapper)
  5. ¿Cómo se inyectan extensiones (fallback, métricas, retry) sin contaminar la interfaz?

Vamos una por una.

Decisión 1 — La interfaz pública

Opción A — Mínima:

class UnifiedClient:
    def __init__(self, provider: str, **kwargs): ...
    def chat(self, prompt: str) -> str: ...

Opción B — Rica:

class UnifiedClient:
    def __init__(self, provider: str, **kwargs): ...
    def chat(self, prompt: str, **opts) -> ChatResponse: ...
    def stream_chat(self, prompt: str, **opts) -> Iterator[str]: ...
    def chat_with_history(self, messages: list[Message], **opts) -> ChatResponse: ...
    def embed(self, texts: list[str]) -> list[Embedding]: ...

Decisión: algo entre los dos. Empezamos con:

class UnifiedClient:
    def __init__(self, primary: str, fallback: list[str] = [], **opts): ...

    def chat(
        self,
        prompt: str,
        *,
        max_tokens: int = 256,
        temperature: float = 0.7,
        system: str | None = None,
    ) -> ChatResponse: ...

    def chat_with_messages(
        self,
        messages: list[Message],
        **opts
    ) -> ChatResponse: ...

    def get_metrics(self) -> Metrics: ...

Por qué:

  • chat() para el caso simple (90% del uso).
  • chat_with_messages() para multi-turn (cuando tu app gestiona historia).
  • Sin streaming, embeddings, function calling (scope explícito M08).
  • Métodos públicos pequeños; extensiones (fallback, métricas) inyectadas, no nuevos métodos.

Decisión 2 — Inputs y outputs

Strings simples vs estructuras:

# Opción A — strings: simple pero pierde info
def chat(self, prompt: str) -> str: ...

# Opción B — Pydantic: estructurado pero más código
def chat(self, prompt: str) -> ChatResponse:
    ...

class ChatResponse(BaseModel):
    text: str
    model: str
    provider: str
    tokens_input: int
    tokens_output: int
    duration_ms: int
    cost_usd: float | None
    raw_response: dict  # acceso al response original si necesitas

Decisión: Opción B (estructurado).

Por qué:

  • Necesitas tokens y cost para tracking (cápsula 06)
  • Necesitas provider para saber quién respondió (importante con fallback)
  • Necesitas raw_response para casos avanzados sin contaminar la API base
  • El costo extra de Pydantic es trivial; el valor de info estructurada es alto

Decisión 3 — Configuración

Opción A — Constructor params:

client = UnifiedClient(
    primary="openai",
    fallback=["openrouter", "ollama"],
    openai_api_key="sk-...",
    openrouter_api_key="sk-or-...",
)

Opción B — Config object:

config = ClientConfig(
    primary=ProviderConfig(name="openai", model="gpt-4o-mini"),
    fallback=[
        ProviderConfig(name="openrouter", model="mistralai/mistral-7b-instruct"),
        ProviderConfig(name="ollama", model="mistral"),
    ],
)
client = UnifiedClient(config)

Opción C — YAML/JSON externo:

client = UnifiedClient.from_yaml("clients.yaml")

Decisión: A para la API en código, C para producción. Implementamos ambas.

Por qué:

  • A es ergonomic para desarrollo y notebooks
  • C es lo correcto para producción (config en archivo, no hardcoded)
  • B introduce ceremonia que no se justifica en el caso simple
  • API keys siempre vienen de env vars o secret manager, no constructor params

Estructura de YAML:

# clients.yaml
primary: openai-gpt4o-mini
fallback:
  - openrouter-mistral
  - ollama-mistral

providers:
  openai-gpt4o-mini:
    type: openai
    model: gpt-4o-mini
    api_key_env: OPENAI_API_KEY
    base_url: https://api.openai.com/v1

  openrouter-mistral:
    type: openai_compatible
    model: mistralai/mistral-7b-instruct
    api_key_env: OPENROUTER_API_KEY
    base_url: https://openrouter.ai/api/v1

  ollama-mistral:
    type: openai_compatible
    model: mistral
    api_key_env: OLLAMA_API_KEY  # placeholder; Ollama ignora
    base_url: http://localhost:11434/v1

  modal-mistral:
    type: modal_custom
    base_url_env: MODAL_BASE_URL
    api_token_env: MODAL_API_TOKEN

Decisión 4 — Manejo de errores

Opción A — Excepciones tipadas:

class UnifiedClientError(Exception): ...
class ProviderError(UnifiedClientError): ...
class AllProvidersFailedError(UnifiedClientError): ...
class RateLimitError(ProviderError): ...
class AuthError(ProviderError): ...
class TimeoutError(ProviderError): ...

Opción B — Result wrapper:

@dataclass
class Result:
    success: bool
    response: ChatResponse | None
    error: str | None

Decisión: Opción A (excepciones tipadas).

Por qué:

  • Python idiomático usa excepciones
  • Tipos específicos permiten try/except RateLimitError para manejo selectivo
  • Result wrappers son comunes en Rust/Go, no en Python — terminan sintiéndose foráneos
  • Stack traces ayudan a debuggear

Decisión 5 — Inyectar extensiones

Necesitamos agregar fallback, métricas, retry — pero sin hacer la clase grande y desordenada. Dos opciones:

Opción A — Decoradores / mixins:

client = with_metrics(with_fallback(BaseUnifiedClient(primary="openai")))

Opción B — Composición declarada:

client = UnifiedClient(
    primary="openai",
    fallback=["openrouter"],
    metrics_enabled=True,
    retry_on=[RateLimitError, TimeoutError],
)

Decisión: Opción B (composición declarada).

Por qué:

  • Opción A es elegante pero menos descubrible
  • Opción B funciona como "config object por defecto"; explícita
  • La complejidad va dentro de UnifiedClient, no en la API de usuario

El diagrama de clases resultante

            ┌─────────────────────────────────────┐
            │           UnifiedClient             │
            │  - primary: BaseAdapter             │
            │  - fallback: list[BaseAdapter]      │
            │  - metrics: MetricsCollector        │
            │  + chat(prompt, **opts) -> Response │
            │  + chat_with_messages(...) -> Resp  │
            │  + get_metrics() -> Metrics         │
            └────────────────┬────────────────────┘
                             │ delega a
                             ▼
            ┌─────────────────────────────────────┐
            │   BaseAdapter (ABC)                 │
            │  + chat(...) -> Response (abstract) │
            │  + name: str                        │
            └────────────────┬────────────────────┘
                             │
       ┌─────────────────────┼─────────────────────┐
       ▼                     ▼                     ▼
┌──────────────┐      ┌──────────────┐      ┌──────────────┐
│OpenAIAdapter │      │OllamaAdapter │      │ModalAdapter  │
│              │      │              │      │              │
└──────────────┘      └──────────────┘      └──────────────┘
       │                     │                     │
       ▼                     ▼                     ▼
   OpenAI API            Ollama API            Modal endpoint

┌──────────────────────────┐
│  ProviderFactory         │
│  + create(config) ->     │
│    BaseAdapter           │
└──────────────────────────┘

ProviderFactory mapea config → instancia del adapter correcto.

┌──────────────────────────┐
│   MetricsCollector       │
│  + record(provider, ...) │
│  + summary() -> Metrics  │
└──────────────────────────┘

MetricsCollector es un colaborador que UnifiedClient opcionalmente usa (cápsula 06).

Los modelos de datos (Pydantic)

# unified_ai_client/models.py
from pydantic import BaseModel, Field
from typing import Literal

class Message(BaseModel):
    role: Literal["system", "user", "assistant"]
    content: str

class ChatResponse(BaseModel):
    text: str
    model: str
    provider: str
    tokens_input: int = 0
    tokens_output: int = 0
    duration_ms: int = 0
    cost_usd: float | None = None
    raw_response: dict | None = None

class ProviderConfig(BaseModel):
    name: str           # identificador único de este config (ej. "openai-gpt4o-mini")
    type: Literal["openai", "openai_compatible", "modal_custom"]
    model: str | None = None
    base_url: str | None = None
    api_key_env: str | None = None
    base_url_env: str | None = None
    api_token_env: str | None = None
    # Pricing opcional para metrics:
    price_input_per_1m: float | None = None
    price_output_per_1m: float | None = None

class ClientConfig(BaseModel):
    primary: str
    fallback: list[str] = Field(default_factory=list)
    providers: dict[str, ProviderConfig]
    metrics_enabled: bool = True

El contrato de BaseAdapter

# unified_ai_client/adapters/base.py
from abc import ABC, abstractmethod
from ..models import Message, ChatResponse, ProviderConfig

class BaseAdapter(ABC):
    def __init__(self, config: ProviderConfig):
        self.config = config
        self.name = config.name

    @abstractmethod
    def chat(
        self,
        messages: list[Message],
        max_tokens: int = 256,
        temperature: float = 0.7,
    ) -> ChatResponse:
        """Genera respuesta a partir de mensajes. Implementación por provider."""
        ...

Reglas que todos los adapters deben cumplir:

  1. Aceptan list[Message], no prompt: str (es responsabilidad de UnifiedClient convertir prompt simple a [Message(role="user", content=prompt)])
  2. Devuelven ChatResponse completo (text + tokens + duración + provider name)
  3. Lanzan excepciones tipadas (RateLimitError, AuthError, etc.) en caso de error
  4. Mide su propio duration_ms internamente (con time.perf_counter())

Estructura de archivos del proyecto

unified_ai_client/
├── __init__.py
├── client.py             # UnifiedClient
├── factory.py            # ProviderFactory
├── exceptions.py         # excepciones tipadas
├── models.py             # ChatResponse, Message, configs
├── metrics.py            # MetricsCollector
├── adapters/
│   ├── __init__.py
│   ├── base.py           # BaseAdapter
│   ├── openai_adapter.py
│   ├── ollama_adapter.py
│   └── modal_adapter.py
└── tests/
    ├── test_unified_client.py
    ├── test_adapters.py
    ├── test_fallback.py
    └── test_metrics.py

pyproject.toml            # package metadata
README.md                 # docs y ejemplos
clients.yaml.example      # ejemplo de config

Trampas comunes en el diseño

Trampa 1 — "Diseño demasiado completo antes de implementar." Mucha gente diseña 5 features y solo implementa 1. Construye un MVP que funcione (cápsula 03) antes de seguir agregando capas.

Trampa 2 — "Cada feature en su propia subclase." UnifiedClientWithFallback, UnifiedClientWithMetrics, UnifiedClientWithRetryAndFallback → combinatoria explosiva. Es por eso que decidimos composición con flags en vez de jerarquía de subclases.

Trampa 3 — "Soporte de todas las features de todos los providers." Si OpenAI tiene 47 parámetros y Anthropic tiene 38, no expongas 85 parámetros en chat(). Expone los comunes (max_tokens, temperature, system). Lo exclusivo va en provider_specific_options o se evita.

Trampa 4 — "Tipos sin Pydantic ni dataclasses." "Mi return es un dict con keys aleatorios" → maintenance hell. Usa Pydantic desde el día 1.

Trampa 5 — "API key en código." OpenAIAdapter(api_key="sk-real-key"). Nunca. Siempre via env var (que YAML referencia con api_key_env).

Ejercicio de diseño

Antes de implementar (cápsula 03), responde para tu propio proyecto:

  1. ¿Vas a soportar streaming? Si sí, ¿agregas stream_chat() o lo unificas con un parámetro stream=True?
  2. Tu producto recibe prompts en múltiples idiomas. ¿Tu Pydantic models manejan eso o asumen ASCII?
  3. Tu fallback debe ser automático o opt-in por request (chat(prompt, use_fallback=False))?
  4. Algunos providers cobran por request fallido. ¿Tu metrics tracking incluye errores? ¿Cobra por errores también?

No hay respuestas únicas — solo decisiones explícitas. Anota las tuyas.

Defaults que vamos a usar en las cápsulas siguientes
  1. No streaming (scope explícito). Lo agregas tú después si lo necesitas.
  2. UTF-8 por default en strings de Pydantic (manejado automático).
  3. Fallback automático. Una llamada chat(prompt) intenta primary → primer fallback → segundo fallback → falla.
  4. Errores se cuentan en metrics pero no se cobran (algunos providers sí cobran; ajusta si aplica).

Resumen

Decidiste:

  • ✅ Interfaz pública: chat() + chat_with_messages() + get_metrics()
  • ✅ Inputs/outputs: Pydantic models (Message, ChatResponse)
  • ✅ Configuración: constructor params + YAML para producción
  • ✅ Manejo de errores: excepciones tipadas (RateLimitError, AuthError, etc.)
  • ✅ Composición declarada (fallback, metrics como flags) vs jerarquía de subclases

Checkpoint: si tienes claros los 5 decisiones de arriba y puedes dibujar el diagrama de clases sin mirar, estás listo para implementar.

Siguiente cápsula

03 — Implementación base. Pasamos del diseño al código. Construyes el esqueleto del paquete + el primer adapter (OpenAI) + el UnifiedClient mínimo viable. Al final de la cápsula vas a tener algo importable y ejecutable, aunque solo con un provider.

Recursos

  1. Refactoring.guru — Adapter, Factory, Strategy patterns — definiciones canónicas con ejemplos Python.
  2. Pydantic v2 docs — modelos para tus dataclasses.
  3. LiteLLM source code — implementación open source; estudia su Router para fallback.
  4. Python typing — Protocol vs ABC — alternativa a ABC con duck typing.
  5. Architectural Decision Records template — para documentar tus decisiones.
← AnteriorMódulo 8: Unified AI Client — Proyecto integrador finalSiguiente →Implementación base