Deploying an AI agent to production is exciting, but without proper infrastructure, you risk runaway costs, system crashes from traffic spikes, and zero visibility into what's happening inside your agent. I've personally watched a startup burn through $12,000 in API credits in a single weekend because they had no rate limiting and no monitoring dashboard. That pain inspired this guide.

In this comprehensive tutorial, you'll learn how to build a production-ready AI agent deployment using HolySheep AI as your API gateway, implementing three critical pillars: real-time monitoring, intelligent rate limiting, and granular cost control. By the end, you'll have a system that alerts you before problems occur, protects you from traffic abuse, and shows you exactly where every cent goes.

Table of Contents

Prerequisites: What You Need Before Starting

Before diving into the technical implementation, let's make sure you have everything you need. Don't worry if you're a complete beginner — I've designed this guide to be accessible to anyone, even if you've never written a line of code before.

What You'll Need

Screenshot hint: Open your HolySheep dashboard after registration. You should see a section labeled "API Keys" where you can generate your first key. Keep this key secret — treat it like a password.

Understanding the Three Pillars of Production AI Deployment

When you deploy an AI agent to handle real users, three things become non-negotiable:

Pillar 1: Monitoring — The Eyes of Your System

You cannot manage what you cannot measure. Monitoring means tracking:

Without monitoring, you're essentially flying blind. Problems emerge silently until they become crises.

Pillar 2: Rate Limiting — The Immune System

Rate limiting protects your system from:

Think of rate limiting as the bouncer at a club — it ensures fair access and prevents chaos.

Pillar 3: Cost Control — The Wallet Guardian

AI APIs charge per token, and costs can spiral quickly. Cost control means:

Step 1: Setting Up Your HolySheep API Connection with Monitoring

Let's start by connecting to HolySheep's API with built-in monitoring. HolySheep provides sub-50ms latency and supports all major AI models including GPT-4.1, Claude Sonnet 4.5, Gemini 2.5 Flash, and DeepSeek V3.2.

Installing the Required Libraries

# Create a new project directory
mkdir ai-agent-production
cd ai-agent-production


Create a virtual environment (keeps your project dependencies isolated)

python -m venv venv


Activate the virtual environment


On Windows:

venv\Scripts\activate

On Mac/Linux:

source venv/bin/activate


Install required libraries

pip install requests python-dotenv prometheus-client flask

Creating Your First Monitored API Connection

import requests
import time
import json
from datetime import datetime
from collections import defaultdict

class HolySheepMonitor:
    """A simple monitoring wrapper for HolySheep AI API calls"""
    
    def __init__(self, api_key: str):
        self.api_key = api_key
        self.base_url = "https://api.holysheep.ai/v1"
        self.headers = {
            "Authorization": f"Bearer {api_key}",
            "Content-Type": "application/json"
        }
        
        # Metrics storage
        self.metrics = {
            "total_requests": 0,
            "successful_requests": 0,
            "failed_requests": 0,
            "total_tokens_used": 0,
            "total_cost": 0.0,
            "response_times": [],
            "errors_by_type": defaultdict(int),
            "requests_per_minute": [],
            "last_request_time": None
        }
    
    def call_model(self, model: str, messages: list, max_tokens: int = 1000):
        """Make a monitored API call to the specified model"""
        
        start_time = time.time()
        self.metrics["total_requests"] += 1
        self.metrics["last_request_time"] = datetime.now()
        
        try:
            # Prepare the request payload
            payload = {
                "model": model,
                "messages": messages,
                "max_tokens": max_tokens
            }
            
            # Make the API call
            response = requests.post(
                f"{self.base_url}/chat/completions",
                headers=self.headers,
                json=payload,
                timeout=30
            )
            
            # Calculate response time
            response_time = (time.time() - start_time) * 1000  # Convert to ms
            self.metrics["response_times"].append(response_time)
            
            # Check for successful response
            if response.status_code == 200:
                self.metrics["successful_requests"] += 1
                data = response.json()
                
                # Extract token usage (if available in response)
                if "usage" in data:
                    tokens_used = data["usage"].get("total_tokens", 0)
                    self.metrics["total_tokens_used"] += tokens_used
                    
                    # Calculate cost (2026 pricing in USD per million tokens)
                    model_prices = {
                        "gpt-4.1": {"output": 8.00},
                        "claude-sonnet-4.5": {"output": 15.00},
                        "gemini-2.5-flash": {"output": 2.50},
                        "deepseek-v3.2": {"output": 0.42}
                    }
                    
                    price_info = model_prices.get(model, {"output": 8.00})
                    cost = (tokens_used / 1_000_000) * price_info["output"]
                    self.metrics["total_cost"] += cost
                
                return {
                    "success": True,
                    "data": data,
                    "response_time_ms": response_time
                }
            else:
                # Handle error responses
                self.metrics["failed_requests"] += 1
                error_type = f"HTTP_{response.status_code}"
                self.metrics["errors_by_type"][error_type] += 1
                
                return {
                    "success": False,
                    "error": f"HTTP Error: {response.status_code}",
                    "response": response.text,
                    "response_time_ms": response_time
                }
                
        except requests.exceptions.Timeout:
            self.metrics["failed_requests"] += 1
            self.metrics["errors_by_type"]["Timeout"] += 1
            return {
                "success": False,
                "error": "Request timed out after 30 seconds"
            }
            
        except requests.exceptions.RequestException as e:
            self.metrics["failed_requests"] += 1
            self.metrics["errors_by_type"]["NetworkError"] += 1
            return {
                "success": False,
                "error": f"Network error: {str(e)}"
            }
    
    def get_metrics_summary(self):
        """Return a summary of all collected metrics"""
        
        avg_response_time = (
            sum(self.metrics["response_times"]) / len(self.metrics["response_times"])
            if self.metrics["response_times"] else 0
        )
        
        return {
            "total_requests": self.metrics["total_requests"],
            "success_rate": (
                self.metrics["successful_requests"] / self.metrics["total_requests"] * 100
                if self.metrics["total_requests"] > 0 else 0
            ),
            "total_tokens_used": self.metrics["total_tokens_used"],
            "total_cost_usd": round(self.metrics["total_cost"], 4),
            "average_response_time_ms": round(avg_response_time, 2),
            "error_breakdown": dict(self.metrics["errors_by_type"]),
            "last_request": self.metrics["last_request_time"].isoformat() 
                if self.metrics["last_request_time"] else None
        }



Example usage

if __name__ == "__main__":
    # Initialize the monitor with your API key
    monitor = HolySheepMonitor(api_key="YOUR_HOLYSHEEP_API_KEY")
    
    # Make a test request
    messages = [
        {"role": "system", "content": "You are a helpful assistant."},
        {"role": "user", "content": "Hello! What is the capital of France?"}
    ]
    
    result = monitor.call_model("deepseek-v3.2", messages)
    print(json.dumps(result, indent=2))
    
    # Get metrics summary
    print("\n--- Metrics Summary ---")
    print(json.dumps(monitor.get_metrics_summary(), indent=2))

Screenshot hint: Run this script and watch the output. You should see your response time in milliseconds, token usage, and cost in USD. HolySheep's <50ms latency means you should see response times under 100ms for most requests.

Understanding the Monitoring Output

After running the script, you'll see metrics like:

Step 2: Implementing Rate Limiting That Scales

Rate limiting is crucial for production systems. Without it, a single misbehaving client or a traffic spike can take down your entire service. Let's implement a robust rate limiter that works at multiple levels.

Building a Token Bucket Rate Limiter

import time
import threading
from collections import defaultdict
from dataclasses import dataclass, field
from typing import Dict, Optional
import json

@dataclass
class TokenBucket:
    """Token bucket algorithm for rate limiting"""
    
    capacity: int  # Maximum tokens in the bucket
    refill_rate: float  # Tokens added per second
    tokens: float = field(init=False)
    last_refill: float = field(init=False)
    
    def __post_init__(self):
        self.tokens = float(self.capacity)
        self.last_refill = time.time()
    
    def consume(self, tokens: int = 1) -> bool:
        """Attempt to consume tokens. Returns True if allowed, False if rate limited."""
        
        # Refill tokens based on time elapsed
        now = time.time()
        elapsed = now - self.last_refill
        self.tokens = min(
            self.capacity,
            self.tokens + (elapsed * self.refill_rate)
        )
        self.last_refill = now
        
        # Check if we have enough tokens
        if self.tokens >= tokens:
            self.tokens -= tokens
            return True
        return False
    
    def get_wait_time(self, tokens: int = 1) -> float:
        """Calculate how many seconds to wait before tokens are available"""
        
        if self.tokens >= tokens:
            return 0.0
        return (tokens - self.tokens) / self.refill_rate


class ProductionRateLimiter:
    """Multi-level rate limiter for production AI agents"""
    
    def __init__(self):
        # Per-user rate limits
        self.user_buckets: Dict[str, TokenBucket] = defaultdict(
            lambda: TokenBucket(capacity=100, refill_rate=10)  # 100 req burst, 10 req/sec
        )
        
        # Global rate limits
        self.global_bucket = TokenBucket(capacity=10000, refill_rate=1000)  # 10K req burst, 1K req/sec
        
        # Budget tracking
        self.daily_budgets: Dict[str, float] = defaultdict(lambda: 100.0)  # $100 default
        self.daily_spending: Dict[str, float] = defaultdict(float)
        self.last_budget_reset = time.time()
        
        # Concurrency control
        self.max_concurrent_requests = 50
        self.active_requests = 0
        self.semaphore = threading.Semaphore(self.max_concurrent_requests)
        
        # Thread safety
        self.lock = threading.Lock()
        
        # Logging
        self.blocked_requests = []
    
    def check_rate_limit(
        self,
        user_id: str,
        required_tokens: int = 1,
        estimated_cost: float = 0.0
    ) -> Dict:
        """
        Comprehensive rate limit check.
        
        Returns:
            {
                "allowed": bool,
                "reason": str or None,
                "wait_time_ms": float,
                "current_bucket_tokens": int
            }
        """
        
        with self.lock:
            # Reset daily budgets if needed (every 24 hours)
            if time.time() - self.last_budget_reset > 86400:
                self.daily_spending.clear()
                self.last_budget_reset = time.time()
            
            # Check 1: Budget limit
            if self.daily_spending[user_id] + estimated_cost > self.daily_budgets[user_id]:
                return {
                    "allowed": False,
                    "reason": f"Daily budget exceeded. Limit: ${self.daily_budgets[user_id]:.2f}, "
                             f"Spent: ${self.daily_spending[user_id]:.2f}",
                    "wait_time_ms": 0,
                    "current_bucket_tokens": 0
                }
            
            # Check 2: Global rate limit
            if not self.global_bucket.consume(required_tokens):
                wait_time = self.global_bucket.get_wait_time(required_tokens) * 1000
                self.log_blocked(user_id, "global_rate_limit", wait_time)
                return {
                    "allowed": False,
                    "reason": "Global rate limit exceeded. Please try again later.",
                    "wait_time_ms": wait_time,
                    "current_bucket_tokens": int(self.global_bucket.tokens)
                }
            
            # Check 3: Per-user rate limit
            user_bucket = self.user_buckets[user_id]
            if not user_bucket.consume(required_tokens):
                wait_time = user_bucket.get_wait_time(required_tokens) * 1000
                self.log_blocked(user_id, "user_rate_limit", wait_time)
                return {
                    "allowed": False,
                    "reason": "User rate limit exceeded. Please slow down your requests.",
                    "wait_time_ms": wait_time,
                    "current_bucket_tokens": int(user_bucket.tokens)
                }
            
            # Check 4: Concurrency limit
            if self.active_requests >= self.max_concurrent_requests:
                self.log_blocked(user_id, "concurrency_limit", 1000)
                return {
                    "allowed": False,
                    "reason": "Server is busy. Maximum concurrent requests reached.",
                    "wait_time_ms": 1000,
                    "current_bucket_tokens": int(user_bucket.tokens)
                }
            
            # All checks passed
            self.active_requests += 1
            return {
                "allowed": True,
                "reason": None,
                "wait_time_ms": 0,
                "current_bucket_tokens": int(user_bucket.tokens)
            }
    
    def release_request(self, user_id: str, actual_cost: float):
        """Call this after a request completes to update spending"""
        
        with self.lock:
            self.active_requests = max(0, self.active_requests - 1)
            self.daily_spending[user_id] += actual_cost
    
    def log_blocked(self, user_id: str, reason: str, wait_time_ms: float):
        """Log blocked requests for analysis"""
        
        self.blocked_requests.append({
            "timestamp": time.time(),
            "user_id": user_id,
            "reason": reason,
            "wait_time_ms": wait_time_ms
        })
        
        # Keep only last 1000 blocked requests
        if len(self.blocked_requests) > 1000:
            self.blocked_requests = self.blocked_requests[-1000:]
    
    def set_user_budget(self, user_id: str, daily_limit: float):
        """Set a custom daily budget for a specific user"""
        
        with self.lock:
            self.daily_budgets[user_id] = daily_limit
    
    def get_rate_limit_status(self, user_id: str) -> Dict:
        """Get current rate limit status for a user"""
        
        with self.lock:
            user_bucket = self.user_buckets[user_id]
            return {
                "user_tokens_remaining": int(user_bucket.tokens),
                "user_bucket_capacity": user_bucket.capacity,
                "user_refill_rate": user_bucket.refill_rate,
                "global_tokens_remaining": int(self.global_bucket.tokens),
                "active_requests": self.active_requests,
                "daily_budget_remaining": (
                    self.daily_budgets[user_id] - self.daily_spending[user_id]
                ),
                "blocked_requests_count": len(
                    [b for b in self.blocked_requests if b["user_id"] == user_id]
                )
            }



Example usage

if __name__ == "__main__":
    limiter = ProductionRateLimiter()
    
    # Simulate requests from different users
    test_users = ["user_001", "user_002", "user_003"]
    
    for user in test_users:
        # Check if request is allowed (estimated cost: $0.001)
        result = limiter.check_rate_limit(
            user_id=user,
            required_tokens=1,
            estimated_cost=0.001
        )
        
        if result["allowed"]:
            print(f"✅ {user}: Request allowed. Tokens left: {result['current_bucket_tokens']}")
            # Simulate request completion
            limiter.release_request(user, actual_cost=0.0008)
        else:
            print(f"❌ {user}: Request blocked - {result['reason']}")
            print(f"   Wait time: {result['wait_time_ms']}ms")
    
    # Check status for a user
    print(f"\n--- Status for user_001 ---")
    print(json.dumps(limiter.get_rate_limit_status("user_001"), indent=2))

Screenshot hint: Run this code and observe the output. Try making rapid requests from the same user and watch the rate limiter kick in after the burst limit is reached.

Rate Limiting Configuration Options

The rate limiter supports flexible configuration:

Step 3: Cost Control Mechanisms That Actually Work

Monitoring tells you what happened. Rate limiting prevents disasters. But cost control actively manages your spending. Let's implement a comprehensive cost management system.

Building an Intelligent Cost Controller

import time
import threading
from datetime import datetime, timedelta
from collections import defaultdict
from dataclasses import dataclass
from typing import Dict, List, Optional, Callable
import json
import smtplib
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart

@dataclass
class BudgetAlert:
    """Represents a budget alert configuration"""
    
    threshold_percent: float  # e.g., 50.0 for 50%
    message: str
    callback: Optional[Callable] = None
    triggered: bool = False


class CostController:
    """Intelligent cost control for AI agent deployments"""
    
    def __init__(self, daily_budget: float = 100.0, monthly_budget: float = 3000.0):
        # Budget configuration
        self.daily_budget = daily_budget
        self.monthly_budget = monthly_budget
        
        # Spending tracking
        self.spending_by_user: Dict[str, float] = defaultdict(float)
        self.spending_by_model: Dict[str, float] = defaultdict(float)
        self.spending_by_day: Dict[str, float] = defaultdict(float)
        self.spending_by_endpoint: Dict[str, float] = defaultdict(float)
        
        # Real-time cost calculation
        self.model_prices_per_million = {
            "gpt-4.1": 8.00,
            "claude-sonnet-4.5": 15.00,
            "gemini-2.5-flash": 2.50,
            "deepseek-v3.2": 0.42
        }
        
        # Alerts
        self.alerts: List[BudgetAlert] = [
            BudgetAlert(50.0, "⚠️ 50% of daily budget used"),
            BudgetAlert(75.0, "🚨 75% of daily budget used"),
            BudgetAlert(90.0, "🔴 90% of daily budget used - Approaching limit"),
            BudgetAlert(100.0, "💸 Daily budget exhausted"),
        ]
        
        # Alert callbacks
        self.alert_callbacks: List[Callable] = []
        
        # Circuit breaker
        self.circuit_broken = False
        self.circuit_break_threshold = 0.95  # Break at 95% of daily budget
        self.circuit_recovery_time = 3600  # 1 hour
        
        # Thread safety
        self.lock = threading.Lock()
        
        # Caching for cost optimization
        self.response_cache: Dict[str, tuple] = {}  # (hash, (response, expiry_time))
        self.cache_ttl = 3600  # 1 hour default TTL
        self.cache_hits = 0
        self.cache_misses = 0
        
        # Budget reset tracking
        self.last_day_reset = datetime.now().date()
        self.last_month_reset = datetime.now().replace(day=1, hour=0, minute=0, second=0)
    
    def calculate_cost(self, model: str, tokens_used: int, is_cache_hit: bool = False) -> float:
        """Calculate cost for a given request"""
        
        price_per_token = self.model_prices_per_million.get(model, 8.00) / 1_000_000
        
        # Apply discount for cache hits (75% cheaper)
        if is_cache_hit:
            price_per_token *= 0.25
        
        return tokens_used * price_per_token
    
    def process_request(
        self,
        user_id: str,
        model: str,
        input_tokens: int,
        output_tokens: int,
        endpoint: str = "chat",
        cache_key: Optional[str] = None
    ) -> Dict:
        """
        Process a request with cost tracking and optimization.
        
        Returns:
            {
                "allowed": bool,
                "cost": float,
                "cache_hit": bool,
                "reason": str or None,
                "tokens_used": int
            }
        """
        
        with self.lock:
            # Check if circuit breaker is active
            if self.circuit_broken:
                return {
                    "allowed": False,
                    "cost": 0.0,
                    "cache_hit": False,
                    "reason": "Circuit breaker active - budget limit reached",
                    "tokens_used": 0
                }
            
            # Check budget reset
            self._check_budget_reset()
            
            # Check for cache hit first
            cache_hit = False
            if cache_key and cache_key in self.response_cache:
                cached_response, expiry = self.response_cache[cache_key]
                if time.time() < expiry:
                    cache_hit = True
                    self.cache_hits += 1
                    return {
                        "allowed": True,
                        "cost": self.calculate_cost(model, input_tokens + output_tokens, True),
                        "cache_hit": True,
                        "reason": None,
                        "tokens_used": 0  # No new tokens used
                    }
            
            self.cache_misses += 1
            
            # Calculate cost
            total_tokens = input_tokens + output_tokens
            cost = self.calculate_cost(model, total_tokens, False)
            
            # Check if this would exceed daily budget
            today = datetime.now().strftime("%Y-%m-%d")
            projected_daily_spending = self.spending_by_day[today] + cost
            
            if projected_daily_spending > self.daily_budget:
                # Check if we're approaching the circuit break threshold
                current_usage = self.spending_by_day[today] / self.daily_budget
                if current_usage >= self.circuit_break_threshold:
                    self.circuit_broken = True
                    threading.Timer(self.circuit_recovery_time, self._reset_circuit_breaker).start()
                
                return {
                    "allowed": False,
                    "cost": 0.0,
                    "cache_hit": False,
                    "reason": f"Request would exceed daily budget. "
                             f"Current: ${self.spending_by_day[today]:.2f}, "
                             f"Budget: ${self.daily_budget:.2f}",
                    "tokens_used": 0
                }
            
            # Track spending
            self.spending_by_user[user_id] += cost
            self.spending_by_model[model] += cost
            self.spending_by_day[today] += cost
            self.spending_by_endpoint[endpoint] += cost
            
            # Check alerts
            self._check_alerts()
            
            # Cache the response if a cache key was provided
            if cache_key:
                self.response_cache[cache_key] = (True, time.time() + self.cache_ttl)
            
            return {
                "allowed": True,
                "cost": cost,
                "cache_hit": False,
                "reason": None,
                "tokens_used": total_tokens
            }
    
    def cache_response(self, cache_key: str, response: any, ttl: Optional[int] = None):
        """Manually cache a response"""
        
        with self.lock:
            expiry = time.time() + (ttl or self.cache_ttl)
            self.response_cache[cache_key] = (response, expiry)
    
    def get_response(self, cache_key: str) -> Optional[any]:
        """Retrieve a cached response if it exists and is not expired"""
        
        with self.lock:
            if cache_key in self.response_cache:
                response, expiry = self.response_cache[cache_key]
                if time.time() < expiry:
                    return response
                else:
                    del self.response_cache[cache_key]
            return None
    
    def _check_budget_reset(self):
        """Check and reset budgets if needed"""
        
        today = datetime.now().date()
        if today > self.last_day_reset:
            self.spending_by_day.clear()
            self.last_day_reset = today
        
        current_month = datetime.now().replace(day=1, hour=0, minute=0, second=0)
        if current_month > self.last_month_reset:
            self.spending_by_month.clear()
            self.last_month_reset = current_month
    
    def _check_alerts(self):
        """Check if any alert thresholds have been crossed"""
        
        today = datetime.now().strftime("%Y-%m-%d")
        current_spending = self.spending_by_day.get(today, 0.0)
        usage_percent = (current_spending / self.daily_budget) * 100
        
        for alert in self.alerts:
            if not alert.triggered and usage_percent >= alert.threshold_percent:
                alert.triggered = True
                for callback in self.alert_callbacks:
                    try:
                        callback(alert.message, usage_percent)
                    except Exception as e:
                        print(f"Alert callback error: {e}")
    
    def _reset_circuit_breaker(self):
        """Reset the circuit breaker after recovery time"""
        
        with self.lock:
            self.circuit_broken = False
            for alert in self.alerts:
                alert.triggered = False
    
    def register_alert_callback(self, callback: Callable):
        """Register a callback to be called when alerts trigger"""
        
        self.alert_callbacks.append(callback)
    
    def get_cost_breakdown(self) -> Dict:
        """Get detailed cost breakdown"""
        
        with self.lock:
            today = datetime.now().strftime("%Y-%m-%d")
            return {
                "today": {
                    "spending": self.spending_by_day.get(today, 0.0),
                    "budget": self.daily_budget,
                    "remaining": self.daily_budget - self.spending_by_day.get(today, 0.0),
                    "usage_percent": (
                        self.spending_by_day.get(today, 0.0) / self.daily_budget * 100
                        if self.daily_budget > 0 else 0
                    )
                },
                "by_user": dict(self.spending_by_user),
                "by_model": dict(self.spending_by_model),
                "by_endpoint": dict(self.spending_by_endpoint),
                "cache_stats": {
                    "hits": self.cache_hits,
                    "misses": self.cache_misses,
                    "hit_rate": (
                        self.cache_hits / (self.cache_hits + self.cache_misses) * 100
                        if (self.cache_hits + self.cache_misses) > 0 else 0
                    )
                },
                "circuit_breaker_active": self.circuit_broken
            }
    
    def generate_savings_report(self) -> Dict:
        """Generate a report showing potential savings"""
        
        with self.lock:
            total_spending = sum(self.spending_by_model.values())
            estimated_without_cache = total_spending / (1 - 0.75)  # Assuming 75% cache savings
            actual_cache_savings = estimated_without_cache - total_spending
            
            model_comparison = {}
            for model, spending in self.spending_by_model.items():
                # Compare with HolySheep pricing (which saves 85%+ vs ¥7.3)
                holy_sheep_spending = spending
                original_cost = spending * 7.3  # Assuming original ¥7.3 rate
                savings = original_cost - holy_sheep_spending
                
                model_comparison[model] = {
                    "holy_sheep_cost": round(holy_sheep_spending, 4),
                    "original_estimate": round(original_cost, 4),
                    "savings_percent": round((savings / original_cost) * 100, 2) if original_cost > 0 else 0
                }
            
            return {
                "total_spending_with_holy_sheep": round(total_spending, 4),
                "estimated_original_cost": round(estimated_without_cache, 4),
                "total_savings": round(actual_cache_savings + (total_spending * 6.3), 4),
                "model_breakdown": model_comparison,
                "cache_savings": round(actual_cache_savings, 4)
            }



Example alert callback

def on_alert_triggered(message: str, usage_percent: float):
    print(f"ALERT: {message} (Usage: {usage_percent:.1f}%)")
    # In production, you would send an email, Slack message, etc.



Example usage

if __name__ == "__main__":
    controller = CostController(daily_budget=50.0)  # $50 daily limit
    controller.register_alert_callback(on_alert_triggered)
    
    # Simulate requests
    requests = [
        {"user": "user_001", "model": "deepseek-v3.2", "input": 500, "output": 150},
        {"user": "user_002", "model": "gemini-2.5-flash", "input": 800, "output": 200},
        {"user": "user_001", "model": "deepseek-v3.2", "input": 500, "output": 150},  # Duplicate - cache hit
    ]
    
    for req in requests:
        result = controller.process_request(
            user_id=req["user"],
            model=req["model"],
            input_tokens=req["input"],
            output_tokens=req["output"]
        )
        
        if result["allowed"]:
            print(f"✅ {req['user']} - {req['model']}: "
                  f"Cost ${result['cost']:.4f} "
                  f"{'(CACHE HIT)' if result['cache_hit'] else ''}")
        else:
            print(f"❌ {req['user']}: {result['reason']}")
    
    print("\n--- Cost Breakdown ---")
    print(json.dumps(controller.get_cost_breakdown(), indent=2))
    
    print("\n--- Savings Report ---")
    print(json.dumps(controller.generate_savings_report(), indent=2))

Screenshot hint: Run this script and watch the cost tracking. Notice how cache hits dramatically reduce costs. The savings report shows exactly how much you're saving compared to traditional API pricing.

Step 4: Bringing It All Together — The Unified Dashboard

Now let's create a unified system that combines monitoring, rate limiting, and cost control into a single, easy-to-use package.

The Complete Production Agent Framework

Related Resources
Related Articles