Mini-Proyecto: Smart Chatbot con OpenRouter

Descripción del proyecto

Proyecto final del Módulo 5: Chatbot inteligente que auto-selecciona modelo óptimo según query, con fallback robusto y cost optimization.

Tiempo: 60 minutos
Dificultad: Media-Alta

---

🎯 Objetivo

Crear chatbot production-ready con:

• ✅ Auto-selection de modelo (cheap → expensive)
• ✅ Fallback chain (3+ modelos)
• ✅ Cost tracking detallado
• ✅ Performance metrics
• ✅ User tier support (free/pro/enterprise)

---

💻 Código Completo

#!/usr/bin/env python3
"""
Smart Chatbot con OpenRouter
Módulo 5 - Proyecto Final
"""

import os
import json
import time
from datetime import datetime
from pathlib import Path
from typing import List, Dict
from dotenv import load_dotenv
from openai import OpenAI

load_dotenv()

# ============================================================================
# CONFIGURACIÓN
# ============================================================================

Path("conversations").mkdir(exist_ok=True)
Path("logs").mkdir(exist_ok=True)

# ============================================================================
# SMART ROUTER
# ============================================================================

class SmartChatbot:
    """Chatbot inteligente con OpenRouter."""
    
    def __init__(self, user_tier: str = "free"):
        self.client = OpenAI(
            base_url="https://openrouter.ai/api/v1",
            api_key=os.getenv("OPENROUTER_API_KEY")
        )
        
        self.user_tier = user_tier
        self.messages = [
            {"role": "system", "content": "Eres un asistente útil."}
        ]
        
        self.conversation_id = datetime.now().strftime("%Y%m%d_%H%M%S")
        self.total_cost = 0.0
        self.query_count = 0
        self.model_usage = {}
        
        # Budget por tier
        self.daily_budgets = {
            "free": 0.10,
            "pro": 1.00,
            "enterprise": 100.00
        }
    
    def classify_query(self, prompt: str) -> str:
        """Clasifica tipo de query."""
        prompt_lower = prompt.lower()
        
        if any(w in prompt_lower for w in ["código", "code", "function", "python"]):
            return "code"
        elif len(prompt) > 300:
            return "complex"
        else:
            return "simple"
    
    def select_model(self, query_type: str) -> List[str]:
        """
        Selecciona modelo(s) según tipo y tier.
        Retorna lista (primary + fallbacks).
        """
        
        # Budget check
        budget = self.daily_budgets[self.user_tier]
        remaining = budget - self.total_cost
        
        # Si cerca del límite → Solo modelos baratos
        if remaining < budget * 0.2:
            return [
                "mistralai/mixtral-8x7b-instruct",
                "anthropic/claude-3-haiku"
            ]
        
        # Routing por tier y tipo
        if self.user_tier == "free":
            return [
                "mistralai/mixtral-8x7b-instruct",
                "anthropic/claude-3-haiku"
            ]
        
        elif self.user_tier == "pro":
            if query_type == "code":
                return [
                    "mistralai/codestral-latest",
                    "openai/gpt-3.5-turbo",
                    "mistralai/mixtral-8x7b-instruct"
                ]
            else:
                return [
                    "openai/gpt-3.5-turbo",
                    "anthropic/claude-3-haiku",
                    "mistralai/mixtral-8x7b-instruct"
                ]
        
        else:  # enterprise
            if query_type == "complex":
                return [
                    "openai/gpt-4-turbo",
                    "anthropic/claude-3-opus",
                    "openai/gpt-3.5-turbo"
                ]
            elif query_type == "code":
                return [
                    "openai/gpt-4-turbo",
                    "mistralai/codestral-latest"
                ]
            else:
                return [
                    "openai/gpt-3.5-turbo",
                    "anthropic/claude-3-haiku"
                ]
    
    def chat(self, user_message: str) -> str:
        """Chat con auto-selection y fallback."""
        
        self.messages.append({"role": "user", "content": user_message})
        
        # Classify query
        query_type = self.classify_query(user_message)
        
        # Select models (primary + fallbacks)
        models = self.select_model(query_type)
        
        # Try models con fallback
        for i, model in enumerate(models):
            try:
                start = time.time()
                
                response = self.client.chat.completions.create(
                    model=model,
                    messages=self.messages,
                    timeout=10.0
                )
                
                latency = time.time() - start
                
                # Extract response
                assistant_message = response.choices[0].message.content
                self.messages.append({"role": "assistant", "content": assistant_message})
                
                # Track metrics
                self._track_usage(model, response.usage, latency, query_type)
                
                # Success message
                fallback_msg = f" (fallback #{i})" if i > 0 else ""
                print(f"✅ {model}{fallback_msg} | {latency:.2f}s | ${self._estimate_cost(response.usage):.6f}")
                
                return assistant_message
            
            except Exception as e:
                print(f"⚠️ {model} failed: {e}")
                if i == len(models) - 1:
                    return "Error: All models exhausted. Please try again."
                continue
        
        return "Error: Unexpected failure"
    
    def _estimate_cost(self, usage) -> float:
        """Estima costo basado en tokens."""
        # Simplified: assume avg $0.50/1M
        return (usage.total_tokens / 1_000_000) * 0.50
    
    def _track_usage(self, model: str, usage, latency: float, query_type: str):
        """Trackea métricas."""
        
        cost = self._estimate_cost(usage)
        
        self.total_cost += cost
        self.query_count += 1
        
        if model not in self.model_usage:
            self.model_usage[model] = {
                "count": 0,
                "total_tokens": 0,
                "total_cost": 0.0,
                "total_latency": 0.0
            }
        
        self.model_usage[model]["count"] += 1
        self.model_usage[model]["total_tokens"] += usage.total_tokens
        self.model_usage[model]["total_cost"] += cost
        self.model_usage[model]["total_latency"] += latency
    
    def show_stats(self):
        """Muestra estadísticas."""
        print("\n" + "="*60)
        print("📊 ESTADÍSTICAS")
        print("="*60)
        print(f"Tier: {self.user_tier}")
        print(f"Queries: {self.query_count}")
        print(f"Total cost: ${self.total_cost:.6f}")
        print(f"Budget used: {(self.total_cost/self.daily_budgets[self.user_tier])*100:.1f}%")
        
        print("\nBy model:")
        for model, stats in self.model_usage.items():
            avg_latency = stats["total_latency"] / stats["count"]
            print(f"\n  {model}:")
            print(f"    Queries: {stats['count']}")
            print(f"    Tokens: {stats['total_tokens']:,}")
            print(f"    Cost: ${stats['total_cost']:.6f}")
            print(f"    Avg latency: {avg_latency:.2f}s")
        
        print("="*60 + "\n")
    
    def save_conversation(self):
        """Guarda conversación."""
        filepath = f"conversations/conversation_{self.conversation_id}.json"
        
        data = {
            "conversation_id": self.conversation_id,
            "timestamp": datetime.now().isoformat(),
            "user_tier": self.user_tier,
            "messages": self.messages[1:],  # Skip system
            "total_cost": self.total_cost,
            "query_count": self.query_count,
            "model_usage": self.model_usage
        }
        
        with open(filepath, "w") as f:
            json.dump(data, f, indent=2)
        
        print(f"💾 Saved: {filepath}")

# ============================================================================
# MAIN CLI
# ============================================================================

def main():
    """CLI principal."""
    
    print("\n" + "="*60)
    print("🤖 SMART CHATBOT (OpenRouter)")
    print("="*60)
    
    # Select tier
    print("\nSelect tier:")
    print("  1. Free ($0.10/day budget)")
    print("  2. Pro ($1.00/day budget)")
    print("  3. Enterprise (unlimited)")
    
    tier_choice = input("\nTier (1-3): ").strip()
    tier_map = {"1": "free", "2": "pro", "3": "enterprise"}
    tier = tier_map.get(tier_choice, "free")
    
    print(f"\n✅ Tier: {tier}")
    print("\nCommands:")
    print("  - 'stats' → Show statistics")
    print("  - 'salir' → Exit")
    print("\n" + "="*60 + "\n")
    
    bot = SmartChatbot(user_tier=tier)
    
    while True:
        try:
            user_input = input("Tú: ").strip()
            
            if user_input.lower() in ["salir", "exit", "quit"]:
                print("\n👋 ¡Adiós!\n")
                bot.show_stats()
                bot.save_conversation()
                break
            
            if user_input.lower() == "stats":
                bot.show_stats()
                continue
            
            if not user_input:
                continue
            
            response = bot.chat(user_input)
            print(f"\nBot: {response}\n")
        
        except KeyboardInterrupt:
            print("\n\n👋 Interrupted\n")
            bot.show_stats()
            bot.save_conversation()
            break

if __name__ == "__main__":
    main()

---

🚀 Uso

python smart_chatbot.py

Conversación ejemplo:

Select tier:
  1. Free ($0.10/day budget)
  2. Pro ($1.00/day budget)
  3. Enterprise (unlimited)

Tier (1-3): 2

✅ Tier: pro

Tú: Hola
✅ openai/gpt-3.5-turbo | 1.5s | $0.000025
Bot: ¡Hola! ¿En qué puedo ayudarte hoy?

Tú: Escribe función Python para fibonacci
✅ mistralai/codestral-latest | 2.1s | $0.000120
Bot: def fibonacci(n):
    if n <= 1:
        return n
    return fibonacci(n-1) + fibonacci(n-2)

Tú: stats

============================================================
📊 ESTADÍSTICAS
============================================================
Tier: pro
Queries: 2
Total cost: $0.000145
Budget used: 0.0%

By model:

  openai/gpt-3.5-turbo:
    Queries: 1
    Tokens: 50
    Cost: $0.000025
    Avg latency: 1.50s

  mistralai/codestral-latest:
    Queries: 1
    Tokens: 240
    Cost: $0.000120
    Avg latency: 2.10s
============================================================

Tú: salir

👋 ¡Adiós!

[Stats displayed again]
💾 Saved: conversations/conversation_20240215_103045.json

---

✅ Rúbrica

Funcionalidad (40 pts):

• (10) Auto-selection funciona
• (10) Fallback funciona
• (10) Cost tracking preciso
• (10) Tier support (free/pro/enterprise)

Optimization (30 pts):

• (10) Budget limits respetados
• (10) Cheap models para queries simples
• (10) Fallback chain lógico

Production-ready (30 pts):

• (10) Error handling robusto
• (10) Stats completas
• (10) Conversaciones guardadas

Total: ___/100

---

✅ Resumen del Módulo 5

Lo que dominaste:

• ✅ OpenRouter API (100+ modelos)
• ✅ Cost optimization strategies
• ✅ Fallback robusto
• ✅ Auto-selection inteligente
• ✅ Multi-tier support

Ventajas vs single provider:

• 50-95% ahorro (modelos baratos)
• High availability (fallback)
• Sin vendor lock-in
• Flexibility

---

➡️ Próximos Módulos

Módulo 6: Modal (Serverless) Módulo 7: Comparación Técnica (Benchmarks) Módulo 8: Unified Client (Proyecto final)

---

¡Felicitaciones! Completaste el Módulo 5. 🎉