Production AI System Architecture

# Async AI processing with queues
from celery import Celery
from langchain.llms import OpenAI
import redis

# Celery for async tasks
app = Celery('ai_worker', broker='redis://localhost:6379')

@app.task
def process_ai_request(user_id, prompt, request_id):
    """Process AI request asynchronously."""
    try:
        llm = OpenAI(temperature=0.7)
        result = llm.predict(prompt)
        
        # Store result
        store_result(request_id, result, user_id)
        
        # Notify user (WebSocket, webhook, etc.)
        notify_user(user_id, request_id, "completed")
        
        return result
    except Exception as e:
        notify_user(user_id, request_id, "failed", str(e))
        raise

# Request batching
from langchain.llms import OpenAI
import asyncio

async def batch_process_requests(requests):
    """Batch multiple requests to reduce API calls."""
    llm = OpenAI(temperature=0.7)
    
    # Group similar requests
    batches = group_similar_requests(requests)
    
    results = []
    for batch in batches:
        # Process batch
        batch_results = await asyncio.gather(*[
            process_single_request(llm, req) for req in batch
        ])
        results.extend(batch_results)
    
    return results

# Response caching
from functools import lru_cache
import hashlib
import json

def cache_key(prompt, model, temperature):
    """Generate cache key from request parameters."""
    data = json.dumps({
        "prompt": prompt,
        "model": model,
        "temperature": temperature
    })
    return hashlib.md5(data.encode()).hexdigest()

@lru_cache(maxsize=1000)
def cached_llm_call(cache_key, prompt):
    """Cached LLM call."""
    llm = OpenAI(temperature=0.7)
    return llm.predict(prompt)

# Usage
key = cache_key("What is AI?", "gpt-3.5-turbo", 0.7)
result = cached_llm_call(key, "What is AI?")  # Cached on second call

Async processing, batching, and caching are essential for scalable AI systems. These patterns reduce costs, improve performance, and handle traffic spikes.

# Retry with exponential backoff
import time
from functools import wraps

def retry_with_backoff(max_retries=3, initial_delay=1, backoff_factor=2):
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            delay = initial_delay
            for attempt in range(max_retries):
                try:
                    return func(*args, **kwargs)
                except Exception as e:
                    if attempt == max_retries - 1:
                        raise
                    time.sleep(delay)
                    delay *= backoff_factor
            return None
        return wrapper
    return decorator

@retry_with_backoff(max_retries=3)
def call_ai_api(prompt):
    """AI API call with automatic retry."""
    llm = OpenAI(temperature=0.7)
    return llm.predict(prompt)

# Circuit breaker pattern
class CircuitBreaker:
    def __init__(self, failure_threshold=5, timeout=60):
        self.failure_count = 0
        self.failure_threshold = failure_threshold
        self.timeout = timeout
        self.last_failure_time = None
        self.state = "closed"  # closed, open, half-open
    
    def call(self, func, *args, **kwargs):
        if self.state == "open":
            if time.time() - self.last_failure_time > self.timeout:
                self.state = "half-open"
            else:
                raise Exception("Circuit breaker is open")
        
        try:
            result = func(*args, **kwargs)
            if self.state == "half-open":
                self.state = "closed"
                self.failure_count = 0
            return result
        except Exception as e:
            self.failure_count += 1
            self.last_failure_time = time.time()
            if self.failure_count >= self.failure_threshold:
                self.state = "open"
            raise

# Fallback mechanism
def get_ai_response_with_fallback(prompt):
    """AI response with fallback to simpler model."""
    try:
        # Try primary model
        llm = OpenAI(model_name="gpt-4", temperature=0.7)
        return llm.predict(prompt)
    except Exception as e:
        # Fallback to cheaper model
        try:
            llm = OpenAI(model_name="gpt-3.5-turbo", temperature=0.7)
            return llm.predict(prompt)
        except Exception as e2:
            # Final fallback
            return "I'm sorry, I'm having trouble processing your request right now."

Robust error handling with retries, circuit breakers, and fallbacks ensures production systems remain available even when components fail.

# Comprehensive monitoring
import time
import logging
from functools import wraps
from prometheus_client import Counter, Histogram, Gauge

# Metrics
api_calls = Counter('ai_api_calls_total', 'Total API calls', ['model', 'status'])
api_latency = Histogram('ai_api_latency_seconds', 'API call latency')
token_usage = Counter('ai_tokens_total', 'Total tokens used', ['type'])
active_requests = Gauge('ai_active_requests', 'Active requests')

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

def monitor_ai_call(func):
    """Decorator to monitor AI API calls."""
    @wraps(func)
    def wrapper(*args, **kwargs):
        start_time = time.time()
        active_requests.inc()
        
        try:
            result = func(*args, **kwargs)
            
            # Log success
            latency = time.time() - start_time
            api_latency.observe(latency)
            api_calls.labels(model=kwargs.get('model', 'unknown'), status='success').inc()
            
            logger.info("AI call succeeded", extra={
                "latency": latency,
                "model": kwargs.get('model'),
                "tokens": result.get('usage', {}).get('total_tokens', 0)
            })
            
            # Track tokens
            if 'usage' in result:
                token_usage.labels(type='prompt').inc(result['usage'].get('prompt_tokens', 0))
                token_usage.labels(type='completion').inc(result['usage'].get('completion_tokens', 0))
            
            return result
        except Exception as e:
            # Log failure
            api_calls.labels(model=kwargs.get('model', 'unknown'), status='error').inc()
            logger.error("AI call failed", extra={
                "error": str(e),
                "model": kwargs.get('model')
            })
            raise
        finally:
            active_requests.dec()
    
    return wrapper

@monitor_ai_call
def call_ai_api(prompt, model="gpt-3.5-turbo"):
    """Monitored AI API call."""
    llm = OpenAI(model_name=model)
    return llm.predict(prompt)

# Health check endpoint
from flask import Flask, jsonify

app = Flask(__name__)

@app.route('/health')
def health_check():
    """Health check endpoint."""
    checks = {
        "api": check_api_connectivity(),
        "database": check_database(),
        "cache": check_cache(),
        "status": "healthy" if all([checks["api"], checks["database"], checks["cache"]]) else "unhealthy"
    }
    
    status_code = 200 if checks["status"] == "healthy" else 503
    return jsonify(checks), status_code

Comprehensive monitoring tracks performance, costs, errors, and system health. This enables proactive issue detection and optimization.

# Cost tracking and optimization
class CostTracker:
    def __init__(self, budget_per_day=100):
        self.budget_per_day = budget_per_day
        self.daily_spend = 0
        self.model_costs = {
            "gpt-4": 0.03 / 1000,  # per 1K tokens
            "gpt-3.5-turbo": 0.002 / 1000
        }
    
    def estimate_cost(self, model, prompt_tokens, completion_tokens):
        """Estimate cost for API call."""
        cost_per_1k = self.model_costs.get(model, 0.002)
        total_tokens = prompt_tokens + completion_tokens
        return (total_tokens / 1000) * cost_per_1k
    
    def should_use_cheaper_model(self, prompt):
        """Decide if cheaper model is sufficient."""
        # Simple heuristic: use GPT-3.5 for short, simple prompts
        if len(prompt) < 500:
            return True
        return False
    
    def track_spend(self, cost):
        """Track spending and check budget."""
        self.daily_spend += cost
        if self.daily_spend > self.budget_per_day:
            raise Exception("Daily budget exceeded")
        return self.daily_spend

# Smart model selection
def get_ai_response_optimized(prompt):
    """Get AI response with cost optimization."""
    tracker = CostTracker()
    
    # Choose model based on complexity
    if tracker.should_use_cheaper_model(prompt):
        model = "gpt-3.5-turbo"
    else:
        model = "gpt-4"
    
    # Check cache first
    cached = get_from_cache(prompt, model)
    if cached:
        return cached
    
    # Make API call
    result = call_ai_api(prompt, model=model)
    
    # Estimate and track cost
    cost = tracker.estimate_cost(
        model,
        result['usage']['prompt_tokens'],
        result['usage']['completion_tokens']
    )
    tracker.track_spend(cost)
    
    # Cache result
    cache_result(prompt, model, result)
    
    return result

Cost optimization is essential for sustainable AI systems. Track spending, use appropriate models, cache aggressively, and set budgets.