Mini-Project: Smart Chatbot with OpenRouter

Project overview

Final project of Module 5: An intelligent chatbot that auto-selects the optimal model based on the query, with robust fallback and cost optimization.

Time: 60 minutes
Difficulty: Medium-High

---

🎯 Objective

Build a production-ready chatbot with:

• ✅ Model auto-selection (cheap → expensive)
• ✅ Fallback chain (3+ models)
• ✅ Detailed cost tracking
• ✅ Performance metrics
• ✅ User tier support (free/pro/enterprise)

---

💻 Complete Code

#!/usr/bin/env python3
"""
Smart Chatbot with OpenRouter
Module 5 - Final Project
"""

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

# ============================================================================
# CONFIGURATION
# ============================================================================

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

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

class SmartChatbot:
    """Intelligent chatbot with 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": "You are a helpful assistant."}
        ]
        
        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 per tier
        self.daily_budgets = {
            "free": 0.10,
            "pro": 1.00,
            "enterprise": 100.00
        }
    
    def classify_query(self, prompt: str) -> str:
        """Classify the query type."""
        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]:
        """
        Select model(s) based on type and tier.
        Returns a list (primary + fallbacks).
        """
        
        # Budget check
        budget = self.daily_budgets[self.user_tier]
        remaining = budget - self.total_cost
        
        # If near the limit → only cheap models
        if remaining < budget * 0.2:
            return [
                "mistralai/mixtral-8x7b-instruct",
                "anthropic/claude-3-haiku"
            ]
        
        # Routing by tier and type
        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 with auto-selection and 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 with 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:
        """Estimate cost based on 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):
        """Track metrics."""
        
        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):
        """Show the statistics."""
        print("\n" + "="*60)
        print("📊 STATISTICS")
        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):
        """Save the conversation."""
        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():
    """Main CLI."""
    
    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("You: ").strip()
            
            if user_input.lower() in ["salir", "exit", "quit"]:
                print("\n👋 Bye!\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()

---

🚀 Usage

python smart_chatbot.py

Example conversation:

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

Tier (1-3): 2

✅ Tier: pro

You: Hi
✅ openai/gpt-3.5-turbo | 1.5s | $0.000025
Bot: Hi! How can I help you today?

You: Write a Python function for fibonacci
✅ mistralai/codestral-latest | 2.1s | $0.000120
Bot: def fibonacci(n):
    if n <= 1:
        return n
    return fibonacci(n-1) + fibonacci(n-2)

You: stats

============================================================
📊 STATISTICS
============================================================
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
============================================================

You: salir

👋 Bye!

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

---

✅ Rubric

Functionality (40 pts):

• (10) Auto-selection works
• (10) Fallback works
• (10) Accurate cost tracking
• (10) Tier support (free/pro/enterprise)

Optimization (30 pts):

• (10) Budget limits respected
• (10) Cheap models for simple queries
• (10) Logical fallback chain

Production-ready (30 pts):

• (10) Robust error handling
• (10) Complete stats
• (10) Conversations saved

Total: ___/100

---

✅ Module 5 Summary

What you mastered:

• ✅ OpenRouter API (100+ models)
• ✅ Cost optimization strategies
• ✅ Robust fallback
• ✅ Intelligent auto-selection
• ✅ Multi-tier support

Advantages vs a single provider:

• 50-95% savings (cheap models)
• High availability (fallback)
• No vendor lock-in
• Flexibility

---

➡️ Next Modules

Module 6: Modal (Serverless) Module 7: Technical Comparison (Benchmarks) Module 8: Unified Client (Final project)

---

Congratulations! You completed Module 5. 🎉