Final Project: Unified AI Client Production-Ready

Description

Final integrating project for the whole guide. You'll create a production-ready Unified AI Client that abstracts 5 providers (OpenAI, OpenRouter, Ollama, LM Studio, Modal).

Time: 2-3 hours
Difficulty: High

---

🎯 Objective

Unified client with:

• ✅ Multi-provider support (5 providers)
• ✅ Factory pattern (extensible)
• ✅ Automatic fallback chain
• ✅ Cost optimization
• ✅ Retry strategies
• ✅ Complete monitoring
• ✅ Testing suite

---

💻 Complete Architecture

#!/usr/bin/env python3
"""
Unified AI Client - Final Project
Abstracts multiple LLM providers
"""

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:
    """Standardized response."""
    content: str
    model: str
    provider: str
    tokens: int
    latency: float
    cost: float

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

class LLMProvider(ABC):
    """Abstract interface for providers."""
    
    @abstractmethod
    def chat(self, messages: List[Dict], **kwargs) -> LLMResponse:
        """Sends chat and returns a standardized response."""
        pass
    
    @abstractmethod
    def get_name(self) -> str:
        """Provider name."""
        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
        
        # Cost (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:
    """Unified client with multi-provider support."""
    
    def __init__(self, fallback_chain: List[str] = None):
        """
        Args:
            fallback_chain: List of providers in order (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]:
        """Initializes available providers."""
        providers = {}
        
        # OpenAI (if API key exists)
        if os.getenv("OPENAI_API_KEY"):
            providers["openai"] = OpenAIProvider(
                api_key=os.getenv("OPENAI_API_KEY")
            )
        
        # OpenRouter (if API key exists)
        if os.getenv("OPENROUTER_API_KEY"):
            providers["openrouter"] = OpenRouterProvider(
                api_key=os.getenv("OPENROUTER_API_KEY")
            )
        
        # Ollama (try to 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:
        """
        Sends chat with automatic fallback.
        
        Args:
            prompt: User message
            **kwargs: Additional parameters (temperature, max_tokens, etc.)
        
        Returns:
            LLMResponse with the result
        """
        
        self.metrics["total_requests"] += 1
        
        messages = [{"role": "user", "content": prompt}]
        
        # Try each provider in order
        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:
        """Returns metrics."""
        return self.metrics
    
    def print_stats(self):
        """Prints metrics."""
        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():
    """Unified client demo."""
    
    print("\n🤖 UNIFIED AI CLIENT\n")
    
    # Initialize with fallback chain
    client = UnifiedAIClient(
        fallback_chain=["openai", "openrouter", "ollama"]
    )
    
    # Test queries
    prompts = [
        "What is Python?",
        "Explain what a server is",
        "Give me a code example"
    ]
    
    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()

---

🚀 Usage

1. Setup .env:

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

---

2. Run:

python unified_client.py

Expected output:

🤖 UNIFIED AI CLIENT

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

Prompt: What is Python?
Response: Python is an interpreted programming language...
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
============================================================

---

✅ Self-Assessment Rubric

Architecture (30 pts):

• (10) Abstract interface (LLMProvider)
• (10) Factory pattern implementation
• (10) Extensible (easy to add providers)

Functionality (40 pts):

• (10) Multi-provider support (3+)
• (10) Fallback chain works
• (10) Complete metrics tracking
• (10) Robust error handling

Production-ready (30 pts):

• (10) Comprehensive logging
• (10) Testing suite (pytest)
• (10) Clear documentation

Total: ___/100

---

🎯 Optional Extensions

1. Add a new provider:

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 based on the 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)

---

🎉 Congratulations!

You completed the entire "LLM Access Strategies" guide!

What you mastered:

Strategic:

• ✅ Decision framework (5 dimensions)
• ✅ Decision matrix with 7+ cases
• ✅ Quantitative trade-offs

Cloud APIs:

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

Local LLMs:

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

Integration:

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

Completed projects (8):

1. Decision Framework tool
2. OpenAI API chatbot
3. LM Studio chatbot (local)
4. Ollama + Docker chatbot
5. Smart OpenRouter chatbot
6. Modal serverless API
7. Comparative decision tool
8. Unified AI Client (this one)

---

📚 Final Resources

Official docs:

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

Community:

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

---

🚀 Next Steps

1. Go deeper on a provider:

• If cloud: advanced OpenAI/OpenRouter
• If local: Ollama clustering
• If serverless: Modal + Ray

2. Implement in a real project:

• Production chatbot
• Client API
• Internal tools

3. Stay up to date:

• Pricing changes (check every 3 months)
• New models (Llama 3, GPT-5, etc.)
• New providers

---

🎓 Certification Completed

You have successfully completed:

✅ 8 Modules (64 capsules)
✅ 8 Integrating projects
✅ 5 Providers mastered
✅ 15-20 hours of practical content

You're ready to implement LLMs in production with confidence.

---

Congratulations on completing the entire bootcamp! 🎉🚀