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

Implementación base

Siguiente →

Tiempo de teclear código. En esta cápsula construyes el esqueleto del paquete unified_ai_client con el primer adapter funcionando (OpenAI). Al final vas a poder hacer pip install -e . y usar tu librería en cualquier script.

Al terminar vas a poder:

  • Crear una estructura Python instalable con pyproject.toml
  • Implementar BaseAdapter y OpenAIAdapter con tipado completo
  • Construir UnifiedClient mínimo viable (solo primary, sin fallback aún)
  • Cargar configuración desde YAML
  • Verificar que todo funciona con un script de prueba

Setup del paquete

Crea la estructura de archivos:

mkdir unified_ai_client_pkg
cd unified_ai_client_pkg
mkdir -p unified_ai_client/adapters tests
touch unified_ai_client/__init__.py
touch unified_ai_client/adapters/__init__.py
touch unified_ai_client/{models.py,exceptions.py,factory.py,client.py}
touch unified_ai_client/adapters/{base.py,openai_adapter.py}
touch tests/__init__.py
touch README.md pyproject.toml

Tu árbol:

unified_ai_client_pkg/
├── pyproject.toml
├── README.md
├── unified_ai_client/
│   ├── __init__.py
│   ├── models.py
│   ├── exceptions.py
│   ├── factory.py
│   ├── client.py
│   └── adapters/
│       ├── __init__.py
│       ├── base.py
│       └── openai_adapter.py
└── tests/
    └── __init__.py

pyproject.toml

[build-system]
requires = ["setuptools>=68.0"]
build-backend = "setuptools.build_meta"

[project]
name = "unified-ai-client"
version = "0.1.0"
description = "Cliente unificado para múltiples providers LLM"
requires-python = ">=3.10"
dependencies = [
    "openai>=1.30.0",
    "pydantic>=2.0",
    "pyyaml>=6.0",
    "httpx>=0.25",
]

[project.optional-dependencies]
dev = ["pytest", "pytest-mock"]

[tool.setuptools.packages.find]
include = ["unified_ai_client*"]

Instala:

python -m venv .venv
source .venv/bin/activate
pip install -e ".[dev]"

pip list debería mostrar unified-ai-client 0.1.0.

unified_ai_client/models.py

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


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
    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
    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

unified_ai_client/exceptions.py

# unified_ai_client/exceptions.py
class UnifiedClientError(Exception):
    """Base para errores del cliente."""


class ConfigError(UnifiedClientError):
    """Error en configuración (provider no existe, env var faltante, etc.)."""


class ProviderError(UnifiedClientError):
    """Error reportado por un provider específico."""

    def __init__(self, provider: str, message: str, original: Exception | None = None):
        super().__init__(f"[{provider}] {message}")
        self.provider = provider
        self.original = original


class AuthError(ProviderError):
    """API key inválida, token expirado, etc."""


class RateLimitError(ProviderError):
    """Provider devolvió 429."""


class TimeoutError(ProviderError):
    """Request excedió el timeout."""


class AllProvidersFailedError(UnifiedClientError):
    """Primary y todos los fallback fallaron."""

    def __init__(self, errors: dict[str, Exception]):
        msg = "Todos los providers fallaron:\n" + "\n".join(
            f"  - {p}: {e}" for p, e in errors.items()
        )
        super().__init__(msg)
        self.errors = errors

unified_ai_client/adapters/base.py

# 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."""
        ...

unified_ai_client/adapters/openai_adapter.py

Cubre OpenAI nativo y todos los providers OpenAI-compatible (OpenRouter, Ollama, LM Studio).

# unified_ai_client/adapters/openai_adapter.py
import os
import time
from openai import OpenAI, APIStatusError, RateLimitError as OpenAIRateLimitError
from openai import APIConnectionError, AuthenticationError, APITimeoutError

from .base import BaseAdapter
from ..models import Message, ChatResponse, ProviderConfig
from ..exceptions import (
    ProviderError,
    AuthError,
    RateLimitError,
    TimeoutError,
    ConfigError,
)


class OpenAIAdapter(BaseAdapter):
    """
    Sirve para OpenAI nativo y cualquier proveedor OpenAI-compatible:
    OpenRouter, Ollama, LM Studio, Anyscale, Together, etc.
    """

    def __init__(self, config: ProviderConfig):
        super().__init__(config)
        api_key = self._resolver_api_key(config)
        base_url = config.base_url or self._resolver_base_url(config)
        self.client = OpenAI(api_key=api_key, base_url=base_url, timeout=60.0)

    def _resolver_api_key(self, config: ProviderConfig) -> str:
        if not config.api_key_env:
            return "no-key-needed"  # caso Ollama local
        key = os.environ.get(config.api_key_env)
        if not key:
            raise ConfigError(
                f"Falta env var '{config.api_key_env}' para provider '{config.name}'"
            )
        return key

    def _resolver_base_url(self, config: ProviderConfig) -> str | None:
        if config.base_url_env:
            return os.environ.get(config.base_url_env)
        return None

    def chat(
        self,
        messages: list[Message],
        max_tokens: int = 256,
        temperature: float = 0.7,
    ) -> ChatResponse:
        if not self.config.model:
            raise ConfigError(f"Provider '{self.name}' no tiene 'model' configurado")

        msgs_openai = [{"role": m.role, "content": m.content} for m in messages]

        inicio = time.perf_counter()
        try:
            r = self.client.chat.completions.create(
                model=self.config.model,
                messages=msgs_openai,
                max_tokens=max_tokens,
                temperature=temperature,
            )
        except AuthenticationError as e:
            raise AuthError(self.name, "API key inválida o sin permisos", e) from e
        except OpenAIRateLimitError as e:
            raise RateLimitError(self.name, "Rate limit alcanzado", e) from e
        except APITimeoutError as e:
            raise TimeoutError(self.name, "Request timeout", e) from e
        except (APIConnectionError, APIStatusError) as e:
            raise ProviderError(self.name, f"Error del provider: {e}", e) from e

        duracion_ms = int((time.perf_counter() - inicio) * 1000)

        texto = r.choices[0].message.content or ""
        usage = r.usage
        tokens_input = usage.prompt_tokens if usage else 0
        tokens_output = usage.completion_tokens if usage else 0

        cost = self._calcular_costo(tokens_input, tokens_output)

        return ChatResponse(
            text=texto,
            model=self.config.model,
            provider=self.name,
            tokens_input=tokens_input,
            tokens_output=tokens_output,
            duration_ms=duracion_ms,
            cost_usd=cost,
            raw_response=r.model_dump(),
        )

    def _calcular_costo(self, tokens_input: int, tokens_output: int) -> float | None:
        if self.config.price_input_per_1m is None or self.config.price_output_per_1m is None:
            return None
        return (
            tokens_input * self.config.price_input_per_1m / 1_000_000
            + tokens_output * self.config.price_output_per_1m / 1_000_000
        )

unified_ai_client/factory.py

# unified_ai_client/factory.py
from .models import ProviderConfig
from .adapters.base import BaseAdapter
from .adapters.openai_adapter import OpenAIAdapter
from .exceptions import ConfigError


class ProviderFactory:
    """Crea adapters dado un ProviderConfig."""

    _REGISTRO: dict[str, type[BaseAdapter]] = {
        "openai": OpenAIAdapter,
        "openai_compatible": OpenAIAdapter,
        # "modal_custom" se agregará en cápsula posterior
    }

    @classmethod
    def crear(cls, config: ProviderConfig) -> BaseAdapter:
        clase_adapter = cls._REGISTRO.get(config.type)
        if not clase_adapter:
            raise ConfigError(
                f"Tipo de provider desconocido: '{config.type}'. "
                f"Tipos válidos: {list(cls._REGISTRO.keys())}"
            )
        return clase_adapter(config)

    @classmethod
    def registrar(cls, tipo: str, clase: type[BaseAdapter]) -> None:
        """Registra un adapter custom desde fuera (extensibilidad)."""
        cls._REGISTRO[tipo] = clase

unified_ai_client/client.py

# unified_ai_client/client.py
from pathlib import Path
from typing import Iterable
import yaml

from .models import Message, ChatResponse, ClientConfig
from .factory import ProviderFactory
from .exceptions import ConfigError


class UnifiedClient:
    """Cliente unificado para acceder a LLMs a través de múltiples providers."""

    def __init__(self, config: ClientConfig):
        self.config = config

        if config.primary not in config.providers:
            raise ConfigError(f"Primary '{config.primary}' no existe en providers")
        for fb in config.fallback:
            if fb not in config.providers:
                raise ConfigError(f"Fallback '{fb}' no existe en providers")

        self.primary = ProviderFactory.crear(config.providers[config.primary])
        self.fallbacks = [
            ProviderFactory.crear(config.providers[name]) for name in config.fallback
        ]

    @classmethod
    def from_yaml(cls, path: str | Path) -> "UnifiedClient":
        with open(path) as f:
            data = yaml.safe_load(f)
        config = ClientConfig(**data)
        return cls(config)

    @classmethod
    def from_dict(cls, data: dict) -> "UnifiedClient":
        return cls(ClientConfig(**data))

    def chat(
        self,
        prompt: str,
        *,
        system: str | None = None,
        max_tokens: int = 256,
        temperature: float = 0.7,
    ) -> ChatResponse:
        """Versión simple: un prompt → una respuesta."""
        messages: list[Message] = []
        if system:
            messages.append(Message(role="system", content=system))
        messages.append(Message(role="user", content=prompt))
        return self.chat_with_messages(messages, max_tokens=max_tokens, temperature=temperature)

    def chat_with_messages(
        self,
        messages: list[Message],
        max_tokens: int = 256,
        temperature: float = 0.7,
    ) -> ChatResponse:
        """Versión con historia: lista completa de mensajes."""
        # En esta cápsula solo usamos primary. Fallback viene en cápsula 04.
        return self.primary.chat(messages, max_tokens=max_tokens, temperature=temperature)

unified_ai_client/__init__.py

# unified_ai_client/__init__.py
from .client import UnifiedClient
from .models import Message, ChatResponse, ProviderConfig, ClientConfig
from .factory import ProviderFactory
from .exceptions import (
    UnifiedClientError,
    ConfigError,
    ProviderError,
    AuthError,
    RateLimitError,
    TimeoutError,
    AllProvidersFailedError,
)

__all__ = [
    "UnifiedClient",
    "Message",
    "ChatResponse",
    "ProviderConfig",
    "ClientConfig",
    "ProviderFactory",
    "UnifiedClientError",
    "ConfigError",
    "ProviderError",
    "AuthError",
    "RateLimitError",
    "TimeoutError",
    "AllProvidersFailedError",
]

Verificación: usa tu librería

Crea examples/quickstart.py fuera del paquete:

# examples/quickstart.py
"""Verifica que la librería funciona con un provider real (OpenAI)."""
from unified_ai_client import UnifiedClient

CONFIG = {
    "primary": "openai-mini",
    "providers": {
        "openai-mini": {
            "name": "openai-mini",
            "type": "openai",
            "model": "gpt-4o-mini",
            "api_key_env": "OPENAI_API_KEY",
            "price_input_per_1m": 0.15,
            "price_output_per_1m": 0.60,
        }
    },
}

client = UnifiedClient.from_dict(CONFIG)
response = client.chat(
    "Explica REST en una sola frase.",
    max_tokens=80,
)

print(f"\nProvider: {response.provider}")
print(f"Modelo:   {response.model}")
print(f"Texto:    {response.text}")
print(f"Tokens:   {response.tokens_input} input + {response.tokens_output} output")
print(f"Duración: {response.duration_ms}ms")
print(f"Costo:    ${response.cost_usd:.6f}" if response.cost_usd else "Costo:    n/d")

Ejecuta:

export OPENAI_API_KEY=sk-...
python examples/quickstart.py

Output esperado:

Provider: openai-mini
Modelo:   gpt-4o-mini
Texto:    REST es un estilo de arquitectura para diseñar APIs que usa HTTP estándar para crear, leer, actualizar y eliminar recursos identificados por URLs.
Tokens:   18 input + 32 output
Duración: 1843ms
Costo:    $0.000022

Verificación con YAML

Crea examples/clients.yaml:

primary: openai-mini
fallback: []

providers:
  openai-mini:
    name: openai-mini
    type: openai
    model: gpt-4o-mini
    api_key_env: OPENAI_API_KEY
    price_input_per_1m: 0.15
    price_output_per_1m: 0.60

Y examples/quickstart_yaml.py:

from unified_ai_client import UnifiedClient
client = UnifiedClient.from_yaml("examples/clients.yaml")
print(client.chat("Di solo hola").text)

Configuración de OpenRouter (mismo adapter)

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

Cambia primary: openrouter-mistral y vuelve a correr el script. Mismo código, distinto provider. Esto es la magia que el path entero apunta a lograr.

Configuración de Ollama local

providers:
  ollama-mistral:
    name: ollama-mistral
    type: openai_compatible
    model: mistral
    base_url: http://localhost:11434/v1
    # Ollama no requiere api_key, pero el SDK lo exige
    api_key_env: OLLAMA_DUMMY_KEY
export OLLAMA_DUMMY_KEY=ollama   # cualquier valor
# Asegúrate de tener Ollama corriendo + el modelo descargado
ollama pull mistral
ollama serve  # si no corre como daemon

python examples/quickstart.py  # con primary: ollama-mistral

Trampas comunes

Trampa 1 — "Olvidé activar el venv y pip install -e . instaló global." Esto contamina tu Python global. Siempre activa el venv antes de instalar.

Trampa 2 — "Cambié código pero al importar veo la versión vieja." pip install -e . instala en "editable mode" — los cambios se reflejan. Si no, verifica que tu IDE/notebook esté usando el venv correcto.

Trampa 3 — "Ollama me da model not found." El nombre del modelo debe coincidir con lo que tienes descargado. Ejecuta ollama list para ver tus modelos y usa exactamente ese nombre en el config.

Trampa 4 — "Tengo errores de import circular." Pasa cuando models.py importa de adapters/ y viceversa. Mantén models.py como hoja del grafo de imports (no importa de otros archivos del paquete).

Trampa 5 — "El cost_usd me sale None." Solo se calcula si configuraste price_input_per_1m y price_output_per_1m. Es null si no.

Ejercicio

Modifica el quickstart para:

  1. Pasar un system prompt ("Eres un asistente que responde en estilo de Hemingway")
  2. Imprimir el JSON completo del raw_response además del texto
  3. Manejar el caso donde OPENAI_API_KEY no está seteada y mostrar un mensaje útil
Ver solución 
# examples/quickstart_avanzado.py
import os
import json
from unified_ai_client import UnifiedClient, ConfigError

if not os.environ.get("OPENAI_API_KEY"):
    print("Error: setea la variable OPENAI_API_KEY antes de correr este script.")
    exit(1)

CONFIG = {
    "primary": "openai-mini",
    "providers": {
        "openai-mini": {
            "name": "openai-mini",
            "type": "openai",
            "model": "gpt-4o-mini",
            "api_key_env": "OPENAI_API_KEY",
            "price_input_per_1m": 0.15,
            "price_output_per_1m": 0.60,
        }
    },
}

try:
    client = UnifiedClient.from_dict(CONFIG)
    response = client.chat(
        "Describe el océano en 3 frases.",
        system="Eres Ernest Hemingway. Frases cortas. Imágenes concretas. Sin adverbios.",
        max_tokens=120,
    )

    print(f"Texto:\n{response.text}\n")
    print(f"Raw response (JSON):")
    print(json.dumps(response.raw_response, indent=2, default=str))
except ConfigError as e:
    print(f"Config error: {e}")

Resumen

Tienes:

  • ✅ Paquete Python instalable (pip install -e .)
  • ✅ Modelos Pydantic para inputs/outputs
  • ✅ Excepciones tipadas para manejo de errores
  • ✅ BaseAdapter abstracto + OpenAIAdapter que cubre OpenAI, OpenRouter, Ollama, LM Studio
  • ✅ ProviderFactory que crea adapters desde config
  • ✅ UnifiedClient con chat() y chat_with_messages()
  • ✅ Carga desde dict y desde YAML
  • ✅ Quickstart funcionando con un provider real

Checkpoint: si python examples/quickstart.py te devuelve respuesta de OpenAI con tokens y costo correctos, estás listo.

Siguiente cápsula

04 — Fallback strategy. Hasta acá tu cliente usa solo primary. Vamos a agregar fallback automático con manejo de errores diferenciado (rate limit → retry, auth error → no retry, network error → siguiente provider). También vemos un circuit breaker simple para no martillear un provider caído.

Recursos

  1. setuptools — pyproject.toml — config de paquetes Python modernos.
  2. pip install -e (editable mode) — para desarrollo de librerías.
  3. OpenAI Python SDK — types — los errores que capturamos.
  4. Pydantic v2 — Validators — para agregar validación custom a tus models.
← AnteriorArquitectura y diseñoSiguiente →Fallback strategy