Proyecto Final: Unified AI Client Production-Ready

Descripción

Proyecto final integrador de toda la guía. Crearás un Unified AI Client production-ready que abstrae 5 proveedores (OpenAI, OpenRouter, Ollama, LM Studio, Modal).

Tiempo: 2-3 horas
Dificultad: Alta

---

🎯 Objetivo

Cliente unificado con:

• ✅ Multi-provider support (5 proveedores)
• ✅ Factory pattern (extensible)
• ✅ Fallback chain automático
• ✅ Cost optimization
• ✅ Retry strategies
• ✅ Monitoring completo
• ✅ Testing suite

---

💻 Arquitectura Completa

#!/usr/bin/env python3
"""
Unified AI Client - Proyecto Final
Abstrae múltiples proveedores LLM
"""

import os
import time
import logging
from abc import ABC, abstractmethod
from typing import List, Dict, Optional
from dataclasses import dataclass
from openai import OpenAI

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# ============================================================================
# DATACLASSES
# ============================================================================

@dataclass
class LLMResponse:
    """Response estandarizado."""
    content: str
    model: str
    provider: str
    tokens: int
    latency: float
    cost: float

# ============================================================================
# ABSTRACT PROVIDER
# ============================================================================

class LLMProvider(ABC):
    """Interface abstracta para proveedores."""
    
    @abstractmethod
    def chat(self, messages: List[Dict], **kwargs) -> LLMResponse:
        """Envía chat y retorna response estandarizado."""
        pass
    
    @abstractmethod
    def get_name(self) -> str:
        """Nombre del proveedor."""
        pass

# ============================================================================
# PROVIDERS IMPLEMENTATION
# ============================================================================

class OpenAIProvider(LLMProvider):
    """OpenAI API provider."""
    
    def __init__(self, api_key: str, model: str = "gpt-3.5-turbo"):
        self.client = OpenAI(api_key=api_key)
        self.model = model
    
    def chat(self, messages: List[Dict], **kwargs) -> LLMResponse:
        start = time.time()
        
        response = self.client.chat.completions.create(
            model=self.model,
            messages=messages,
            **kwargs
        )
        
        latency = time.time() - start
        tokens = response.usage.total_tokens
        
        # Costo (GPT-3.5)
        cost = (tokens / 1_000_000) * 0.50
        
        return LLMResponse(
            content=response.choices[0].message.content,
            model=self.model,
            provider="openai",
            tokens=tokens,
            latency=latency,
            cost=cost
        )
    
    def get_name(self) -> str:
        return "OpenAI"

class OpenRouterProvider(LLMProvider):
    """OpenRouter multi-provider."""
    
    def __init__(self, api_key: str, model: str = "mistralai/mixtral-8x7b-instruct"):
        self.client = OpenAI(
            base_url="https://openrouter.ai/api/v1",
            api_key=api_key
        )
        self.model = model
    
    def chat(self, messages: List[Dict], **kwargs) -> LLMResponse:
        start = time.time()
        
        response = self.client.chat.completions.create(
            model=self.model,
            messages=messages,
            **kwargs
        )
        
        latency = time.time() - start
        tokens = response.usage.total_tokens
        cost = (tokens / 1_000_000) * 0.24  # Mixtral pricing
        
        return LLMResponse(
            content=response.choices[0].message.content,
            model=self.model,
            provider="openrouter",
            tokens=tokens,
            latency=latency,
            cost=cost
        )
    
    def get_name(self) -> str:
        return "OpenRouter"

class OllamaProvider(LLMProvider):
    """Ollama local provider."""
    
    def __init__(self, model: str = "mistral"):
        self.client = OpenAI(
            base_url="http://localhost:11434/v1",
            api_key="ollama"
        )
        self.model = model
    
    def chat(self, messages: List[Dict], **kwargs) -> LLMResponse:
        start = time.time()
        
        response = self.client.chat.completions.create(
            model=self.model,
            messages=messages,
            **kwargs
        )
        
        latency = time.time() - start
        tokens = response.usage.total_tokens if hasattr(response.usage, 'total_tokens') else 0
        
        return LLMResponse(
            content=response.choices[0].message.content,
            model=self.model,
            provider="ollama",
            tokens=tokens,
            latency=latency,
            cost=0.0  # Local = $0
        )
    
    def get_name(self) -> str:
        return "Ollama"

# ============================================================================
# UNIFIED CLIENT
# ============================================================================

class UnifiedAIClient:
    """Cliente unificado con multi-provider support."""
    
    def __init__(self, fallback_chain: List[str] = None):
        """
        Args:
            fallback_chain: Lista de proveedores en orden (primary → fallback)
                           Default: ["openai", "openrouter", "ollama"]
        """
        self.fallback_chain = fallback_chain or ["openai", "openrouter", "ollama"]
        self.providers = self._init_providers()
        
        self.metrics = {
            "total_requests": 0,
            "successful_requests": 0,
            "failed_requests": 0,
            "total_cost": 0.0,
            "total_tokens": 0,
            "by_provider": {}
        }
    
    def _init_providers(self) -> Dict[str, LLMProvider]:
        """Inicializa providers disponibles."""
        providers = {}
        
        # OpenAI (si API key existe)
        if os.getenv("OPENAI_API_KEY"):
            providers["openai"] = OpenAIProvider(
                api_key=os.getenv("OPENAI_API_KEY")
            )
        
        # OpenRouter (si API key existe)
        if os.getenv("OPENROUTER_API_KEY"):
            providers["openrouter"] = OpenRouterProvider(
                api_key=os.getenv("OPENROUTER_API_KEY")
            )
        
        # Ollama (intenta connect)
        try:
            test_client = OpenAI(base_url="http://localhost:11434/v1", api_key="ollama")
            test_client.models.list()
            providers["ollama"] = OllamaProvider()
        except:
            logger.warning("Ollama not available")
        
        return providers
    
    def chat(self, prompt: str, **kwargs) -> LLMResponse:
        """
        Envía chat con fallback automático.
        
        Args:
            prompt: Mensaje del usuario
            **kwargs: Parameters adicionales (temperature, max_tokens, etc.)
        
        Returns:
            LLMResponse con resultado
        """
        
        self.metrics["total_requests"] += 1
        
        messages = [{"role": "user", "content": prompt}]
        
        # Intenta cada provider en orden
        for provider_name in self.fallback_chain:
            provider = self.providers.get(provider_name)
            
            if not provider:
                logger.warning(f"Provider {provider_name} not configured, skipping")
                continue
            
            try:
                logger.info(f"Trying {provider_name}...")
                
                response = provider.chat(messages, **kwargs)
                
                # Track success
                self._track_success(provider_name, response)
                
                logger.info(f"✅ Success with {provider_name} | {response.latency:.2f}s | ${response.cost:.6f}")
                
                return response
            
            except Exception as e:
                logger.error(f"❌ {provider_name} failed: {e}")
                self._track_failure(provider_name)
                continue
        
        # All failed
        self.metrics["failed_requests"] += 1
        raise Exception("All providers exhausted")
    
    def _track_success(self, provider: str, response: LLMResponse):
        """Track successful request."""
        self.metrics["successful_requests"] += 1
        self.metrics["total_cost"] += response.cost
        self.metrics["total_tokens"] += response.tokens
        
        if provider not in self.metrics["by_provider"]:
            self.metrics["by_provider"][provider] = {
                "requests": 0,
                "cost": 0.0,
                "tokens": 0,
                "latency_total": 0.0
            }
        
        self.metrics["by_provider"][provider]["requests"] += 1
        self.metrics["by_provider"][provider]["cost"] += response.cost
        self.metrics["by_provider"][provider]["tokens"] += response.tokens
        self.metrics["by_provider"][provider]["latency_total"] += response.latency
    
    def _track_failure(self, provider: str):
        """Track failed request."""
        if provider not in self.metrics["by_provider"]:
            self.metrics["by_provider"][provider] = {
                "requests": 0,
                "failures": 0
            }
        
        self.metrics["by_provider"][provider]["failures"] = \
            self.metrics["by_provider"][provider].get("failures", 0) + 1
    
    def get_stats(self) -> dict:
        """Retorna métricas."""
        return self.metrics
    
    def print_stats(self):
        """Imprime métricas."""
        print("\n" + "="*60)
        print("📊 UNIFIED CLIENT STATS")
        print("="*60)
        print(f"Total requests: {self.metrics['total_requests']}")
        print(f"Successful: {self.metrics['successful_requests']}")
        print(f"Failed: {self.metrics['failed_requests']}")
        print(f"Total cost: ${self.metrics['total_cost']:.6f}")
        print(f"Total tokens: {self.metrics['total_tokens']:,}")
        
        print("\nBy provider:")
        for provider, stats in self.metrics["by_provider"].items():
            if stats.get("requests", 0) > 0:
                avg_latency = stats["latency_total"] / stats["requests"]
                print(f"\n  {provider}:")
                print(f"    Requests: {stats['requests']}")
                print(f"    Cost: ${stats['cost']:.6f}")
                print(f"    Tokens: {stats['tokens']:,}")
                print(f"    Avg latency: {avg_latency:.2f}s")
        
        print("="*60 + "\n")

# ============================================================================
# MAIN - DEMO
# ============================================================================

def main():
    """Demo del unified client."""
    
    print("\n🤖 UNIFIED AI CLIENT\n")
    
    # Initialize con fallback chain
    client = UnifiedAIClient(
        fallback_chain=["openai", "openrouter", "ollama"]
    )
    
    # Test queries
    prompts = [
        "¿Qué es Python?",
        "Explica qué es un servidor",
        "Dame un ejemplo de código"
    ]
    
    for prompt in prompts:
        try:
            response = client.chat(prompt)
            print(f"\nPrompt: {prompt}")
            print(f"Response: {response.content[:100]}...")
            print(f"Provider: {response.provider} | Model: {response.model}")
            print(f"Latency: {response.latency:.2f}s | Cost: ${response.cost:.6f}")
        
        except Exception as e:
            print(f"\n❌ Error: {e}")
    
    # Show stats
    client.print_stats()

if __name__ == "__main__":
    main()

---

🚀 Uso

1. Setup .env:

OPENAI_API_KEY=sk-proj-...
OPENROUTER_API_KEY=sk-or-v1-...

---

2. Ejecutar:

python unified_client.py

Output esperado:

🤖 UNIFIED AI CLIENT

Trying openai...
✅ Success with openai | 1.5s | $0.000025

Prompt: ¿Qué es Python?
Response: Python es un lenguaje de programación interpretado...
Provider: openai | Model: gpt-3.5-turbo
Latency: 1.50s | Cost: $0.000025

...

============================================================
📊 UNIFIED CLIENT STATS
============================================================
Total requests: 3
Successful: 3
Failed: 0
Total cost: $0.000075
Total tokens: 150

By provider:

  openai:
    Requests: 3
    Cost: $0.000075
    Tokens: 150
    Avg latency: 1.50s
============================================================

---

✅ Rúbrica de Auto-Evaluación

Arquitectura (30 pts):

• (10) Abstract interface (LLMProvider)
• (10) Factory pattern implementation
• (10) Extensible (fácil añadir providers)

Funcionalidad (40 pts):

• (10) Multi-provider support (3+)
• (10) Fallback chain funciona
• (10) Metrics tracking completo
• (10) Error handling robusto

Production-ready (30 pts):

• (10) Logging comprehensivo
• (10) Testing suite (pytest)
• (10) Documentación clara

Total: ___/100

---

🎯 Extensiones Opcionales

1. Añadir provider nuevo:

class AnthropicProvider(LLMProvider):
    """Anthropic Claude provider."""
    
    def chat(self, messages, **kwargs):
        # Implementation
        pass

---

2. Smart routing:

def smart_route(self, prompt: str) -> str:
    """Auto-select provider según query."""
    if len(prompt) < 50:
        return "ollama"  # Cheap
    else:
        return "openai"  # Quality

---

3. Caching layer:

from functools import lru_cache

@lru_cache(maxsize=100)
def cached_chat(self, prompt: str):
    return self.chat(prompt)

---

🎉 ¡Felicitaciones!

Completaste la guía completa "LLM Access Strategies"!

Lo que dominaste:

Estratégico:

• ✅ Framework de decisión (5 dimensiones)
• ✅ Matriz de decisión con 7+ casos
• ✅ Trade-offs cuantitativos

Cloud APIs:

• ✅ OpenAI API (production-ready)
• ✅ OpenRouter (100+ modelos)
• ✅ Modal (serverless)

Local LLMs:

• ✅ LM Studio (GUI)
• ✅ Ollama (CLI + Docker)

Integración:

• ✅ Unified client pattern
• ✅ Multi-provider abstraction
• ✅ Fallback strategies
• ✅ Cost optimization

Proyectos completados (8):

1. Decision Framework tool
2. Chatbot OpenAI API
3. Chatbot LM Studio (local)
4. Chatbot Ollama + Docker
5. Smart Chatbot OpenRouter
6. API Modal serverless
7. Decision tool comparativo
8. Unified AI Client (este)

---

📚 Recursos Finales

Docs oficiales:

1. OpenAI Platform
2. OpenRouter
3. Ollama
4. LM Studio
5. Modal

Comunidad:

1. r/LocalLLaMA
2. OpenAI Community
3. Modal Discord

---

🚀 Próximos Pasos

1. Profundiza en un proveedor:

• Si cloud: OpenAI/OpenRouter avanzado
• Si local: Ollama clustering
• Si serverless: Modal + Ray

2. Implementa en proyecto real:

• Chatbot producción
• API cliente
• Internal tools

3. Mantente actualizado:

• Pricing cambia (revisa cada 3 meses)
• Nuevos modelos (Llama 3, GPT-5, etc.)
• Nuevos providers

---

🎓 Certificación Completada

Has completado exitosamente:

✅ 8 Módulos (64 cápsulas)
✅ 8 Proyectos integradores
✅ 5 Proveedores dominados
✅ 15-20 horas de contenido práctico

Estás listo para implementar LLMs en producción con confianza.

---

¡Enhorabuena por completar el bootcamp completo! 🎉🚀