Shahzad Bhatti Welcome to my ramblings and rants!

# Create instance with L4 GPU
gcloud compute instances create vllm-test \
  --zone=us-central1-a \
  --machine-type=g2-standard-8 \
  --accelerator=type=nvidia-l4,count=1 \
  --image-family=ubuntu-2004-lts \
  --image-project=ubuntu-os-cloud \
  --boot-disk-size=200GB \
  --boot-disk-type=pd-ssd \
  --maintenance-policy=TERMINATE

# SSH into instance
gcloud compute ssh vllm-test --zone=us-central1-a

# Update system
sudo apt update && sudo apt upgrade -y

# Install CUDA 11.8
wget https://developer.download.nvidia.com/compute/cuda/11.8.0/local_installers/cuda_11.8.0_520.61.05_linux.run
sudo sh cuda_11.8.0_520.61.05_linux.run --silent --toolkit

# Add to PATH
echo 'export PATH=/usr/local/cuda-11.8/bin:$PATH' >> ~/.bashrc
echo 'export LD_LIBRARY_PATH=/usr/local/cuda-11.8/lib64:$LD_LIBRARY_PATH' >> ~/.bashrc
source ~/.bashrc

# Verify
nvidia-smi  # Should show your L4 GPU
nvcc --version  # Should show CUDA 11.8

# Install Python 3.10
sudo apt install -y python3.10 python3.10-venv python3-pip

# Clone the repository
git clone https://github.com/bhatti/vllm-tutorial.git
cd vllm-tutorial

# Create virtual environment
python3 -m venv venv
source venv/bin/activate

# Install dependencies
pip install --upgrade pip
pip install -r requirements.txt

# Test vLLM installation
python -c "import vllm; print(f'vLLM version: {vllm.__version__}')"

# Quick functionality test
python examples/01_basic_vllm.py

Loading model microsoft/phi-2...
Model loaded in 8.3 seconds

Generating response...
Generated 50 tokens in 987ms
Throughput: 41.5 tokens/sec

? vLLM is working!

   git clone https://github.com/bhatti/vllm-tutorial.git
   cd vllm-tutorial

   # Follow setup instructions in README
   python examples/01_basic_vllm.py

   # See the actual results from GCP L4 testing
   cat benchmarks/results/01_throughput_results.json

def classify_complexity(self, prompt: str) -> str:
    """
    Classify prompt complexity to select appropriate model

    Rules:
    - Simple: Definitions, quick facts, <50 words
    - Medium: Summaries, comparisons, 50-150 words
    - Complex: Deep analysis, multi-step reasoning, >150 words
    """
    word_count = len(prompt.split())

    # Keywords indicating complexity
    complex_keywords = [
        "analyze", "compare", "evaluate", "assess risk",
        "recommend", "predict", "forecast", "implications"
    ]

    medium_keywords = [
        "summarize", "explain", "describe", "list",
        "what are", "how does", "differences"
    ]

    has_complex = any(kw in prompt.lower() for kw in complex_keywords)
    has_medium = any(kw in prompt.lower() for kw in medium_keywords)

    # Classification logic
    if word_count > 150 or has_complex:
        return "complex"
    elif word_count > 50 or has_medium:
        return "medium"
    else:
        return "simple"

Query: "What is EBITDA?"
Classified as: simple ? Routed to: Phi-2
Cost: $0.00002038
Latency: 4,843ms (first request, includes model loading)
Quality: ? Perfect (simple definition)

Query: "Summarize Apple's Q4 2024 earnings highlights"
Classified as: medium ? Routed to: Mistral-7B
Cost: $0.00000865
Latency: 4,827ms
Quality: ? Good summary

Query: "Analyze Microsoft's 10-K risk factors and assess their potential impact on future earnings"
Classified as: complex ? Routed to: Llama-3-8B
Cost: $0.00001382
Latency: 4,836ms
Quality: ? Detailed analysis

from typing import Dict, Optional
from dataclasses import dataclass
from vllm import LLM, SamplingParams

@dataclass
class ModelConfig:
    """Configuration for a model tier"""
    name: str
    complexity: str  # "simple", "medium", "complex"
    cost_per_1k_tokens: float
    max_tokens: int

class IntelligentRouter:
    """
    Production-ready intelligent router with:
    - Complexity-based routing
    - Budget enforcement
    - Cost tracking
    - Fallback handling
    """

    def __init__(self, daily_budget_usd: float = 100.0):
        self.daily_budget_usd = daily_budget_usd
        self.total_cost_today = 0.0

        # Model configurations
        self.models = {
            "phi-2": ModelConfig(
                name="microsoft/phi-2",
                complexity="simple",
                cost_per_1k_tokens=0.0001,
                max_tokens=1024,
            ),
            "mistral-7b": ModelConfig(
                name="mistralai/Mistral-7B-Instruct-v0.2",
                complexity="medium",
                cost_per_1k_tokens=0.0002,
                max_tokens=2048,
            ),
            "llama-3-8b": ModelConfig(
                name="meta-llama/Meta-Llama-3-8B",
                complexity="complex",
                cost_per_1k_tokens=0.0003,
                max_tokens=4096,
            ),
        }

        # Initialize LLM (in production, these would be separate instances)
        self.llm = LLM(
            model=self.models["phi-2"].name,
            trust_remote_code=True,
            gpu_memory_utilization=0.9,
        )

    def route_request(self, prompt: str, max_tokens: int = 200) -> Dict:
        """
        Route request to appropriate model

        Returns:
            Dict with 'response', 'model_used', 'cost', 'latency_ms'
        """
        # Step 1: Classify complexity
        complexity = self.classify_complexity(prompt)

        # Step 2: Select model
        model_id = self._select_model(complexity)
        model_config = self.models[model_id]

        # Step 3: Check budget
        estimated_cost = self._estimate_cost(model_config, prompt, max_tokens)
        if self.total_cost_today + estimated_cost > self.daily_budget_usd:
            # Budget exceeded - fallback to cheapest model
            model_id = "phi-2"
            model_config = self.models[model_id]

        # Step 4: Generate response
        sampling_params = SamplingParams(
            temperature=0.7,
            top_p=0.9,
            max_tokens=max_tokens,
        )

        start_time = time.time()
        outputs = self.llm.generate([prompt], sampling_params)
        latency_ms = (time.time() - start_time) * 1000

        # Step 5: Track cost
        tokens_generated = len(outputs[0].outputs[0].token_ids)
        actual_cost = self._calculate_cost(model_config, prompt, tokens_generated)
        self.total_cost_today += actual_cost

        return {
            "response": outputs[0].outputs[0].text,
            "model_used": model_id,
            "cost_usd": actual_cost,
            "latency_ms": latency_ms,
            "tokens_generated": tokens_generated,
        }

    def _select_model(self, complexity: str) -> str:
        """Select model based on complexity"""
        for model_id, config in self.models.items():
            if config.complexity == complexity:
                return model_id
        return "phi-2"  # Default fallback

    def _estimate_cost(self, config: ModelConfig, prompt: str, max_tokens: int) -> float:
        """Estimate cost before generation"""
        input_tokens = len(prompt) / 4  # Rough estimate
        total_tokens = input_tokens + max_tokens
        return (total_tokens / 1000) * config.cost_per_1k_tokens

    def _calculate_cost(self, config: ModelConfig, prompt: str, tokens_generated: int) -> float:
        """Calculate actual cost after generation"""
        input_tokens = len(prompt) / 4
        total_tokens = input_tokens + tokens_generated
        return (total_tokens / 1000) * config.cost_per_1k_tokens

# Initialize router with daily budget
router = IntelligentRouter(daily_budget_usd=100.0)

# Route a simple query
result = router.route_request("What is gross margin?")
print(f"Model used: {result['model_used']}")  # phi-2
print(f"Cost: ${result['cost_usd']:.6f}")     # $0.000020

# Route a complex query
result = router.route_request(
    "Analyze Tesla's competitive positioning in the EV market "
    "and provide investment recommendations based on recent trends"
)
print(f"Model used: {result['model_used']}")  # llama-3-8b
print(f"Cost: ${result['cost_usd']:.6f}")     # $0.000138

from dataclasses import dataclass
from datetime import datetime

@dataclass
class BudgetConfig:
    """Budget configuration with multiple enforcement levels"""
    max_cost_per_request: float = 0.50        # Level 1: prevent accidents
    daily_budget_usd: float = 100.0           # Level 2: daily cap
    monthly_budget_usd: float = 3000.0        # Level 3: monthly cap
    warning_threshold_pct: float = 0.80       # Warn at 80%

class BudgetEnforcer:
    """Hard budget enforcement - prevents spending, not just monitors"""
    
    def __init__(self, config: BudgetConfig):
        self.config = config
        self.daily_spend = 0.0
        self.monthly_spend = 0.0
        # ... implementation
    
    def check_budget(self, estimated_cost: float) -> Dict:
        """Check BEFORE generating - this is the key difference"""
        
        # Level 1: Per-request limit
        if estimated_cost > self.config.max_cost_per_request:
            return {"action": "reject", "reason": "Request too expensive"}
        
        # Level 2: Daily budget
        if self.daily_spend + estimated_cost > self.config.daily_budget_usd:
            return {"action": "downgrade", "reason": "Daily limit approaching"}
        
        # Level 3: Monthly budget
        if self.monthly_spend + estimated_cost > self.config.monthly_budget_usd:
            return {"action": "downgrade", "reason": "Monthly limit approaching"}
        
        return {"action": "allow"}

[Fixed System Prompt - 500 tokens]
You are a financial analyst AI assistant specializing in:
- Earnings report analysis
- SEC filing interpretation
- Market sentiment analysis
...

[User Query - 50 tokens]
What is EBITDA?

[Response - 100 tokens]
EBITDA stands for...

from vllm import LLM

# WITHOUT prefix caching
llm = LLM(
    model="microsoft/phi-2",
    trust_remote_code=True,
)

# WITH prefix caching (80% cost reduction!)
llm = LLM(
    model="microsoft/phi-2",
    trust_remote_code=True,
    enable_prefix_caching=True,  # <-- That's it!
)

Request 1: $0.00010188 (full cost)
Request 2: $0.00010188 (full cost)
Request 3: $0.00010188 (full cost)
...
Total: $0.00112068 (11 × $0.00010188)

Request 1: $0.00002038 (full cost - establishes cache)
Request 2: $0.00000414 (80% cheaper - uses cache!)
Request 3: $0.00000409 (80% cheaper)
Request 4: $0.00000865 (80% cheaper)
...
Total: $0.00010031

Savings: $0.00102037 (91% reduction!)
Cache hit rate: 90.9% (10/11 requests)

# Benchmark code
class QuantizationBenchmark:
    def benchmark_quantization(self, quantization: str):
        """
        Test quantization scheme

        Args:
            quantization: "none" (FP16), "fp8", "awq", or "gptq"
        """
        llm_kwargs = {
            "model": "microsoft/phi-2",
            "trust_remote_code": True,
            "gpu_memory_utilization": 0.9,
            "max_model_len": 1024,
        }

        # Add quantization if specified
        if quantization != "none":
            llm_kwargs["quantization"] = quantization

        # Load model
        start = time.time()
        llm = LLM(**llm_kwargs)
        load_time = time.time() - start

        # Measure memory
        gpu_memory = torch.cuda.memory_allocated() / 1e9  # GB

        # Benchmark generation
        prompt = "Explain quantum computing in simple terms"
        sampling_params = SamplingParams(max_tokens=100)

        start = time.time()
        outputs = llm.generate([prompt], sampling_params)
        latency_ms = (time.time() - start) * 1000

        return {
            "quantization": quantization,
            "memory_gb": gpu_memory,
            "load_time_sec": load_time,
            "latency_ms": latency_ms,
            "tokens_per_sec": 100 / (latency_ms / 1000),
        }

from vllm import LLM

# Option 1: Use pre-quantized AWQ model
llm = LLM(
    model="TheBloke/Mistral-7B-Instruct-v0.2-AWQ",  # Pre-quantized!
    quantization="awq",
    trust_remote_code=True,
    gpu_memory_utilization=0.9,
)

# That's it! 3.7x smaller, ready to use

# Mistral-7B in FP16
llm_fp16 = LLM(model="mistralai/Mistral-7B-Instruct-v0.2")
# Memory: ~16GB VRAM
# Fits on: A100-40GB, L4 (barely)

# Mistral-7B in AWQ
llm_awq = LLM(
    model="TheBloke/Mistral-7B-Instruct-v0.2-AWQ",
    quantization="awq"
)
# Memory: ~4.3GB VRAM
# Fits on: T4-16GB, L4 (comfortably), even RTX 3090

# Savings: 72% memory reduction!

from prometheus_client import Counter, Histogram, Gauge, generate_latest
from typing import Dict
import time

class VLLMMetrics:
    """
    Production metrics for vLLM serving

    Tracks:
    - Request counts (total, success, failure)
    - Latency distributions (P50, P95, P99)
    - Token throughput
    - Cost tracking
    - Model distribution
    """

    def __init__(self):
        # Request counters
        self.requests_total = Counter(
            'vllm_requests_total',
            'Total number of requests',
            ['model', 'status']
        )

        # Latency histogram
        self.latency = Histogram(
            'vllm_latency_ms',
            'Request latency in milliseconds',
            ['model'],
            buckets=[10, 50, 100, 250, 500, 1000, 2500, 5000, 10000]
        )

        # Token metrics
        self.tokens_generated = Counter(
            'vllm_tokens_generated_total',
            'Total tokens generated',
            ['model']
        )

        self.tokens_per_second = Gauge(
            'vllm_tokens_per_second',
            'Current tokens per second',
            ['model']
        )

        # Cost tracking
        self.cost_usd = Counter(
            'vllm_cost_usd_total',
            'Total cost in USD',
            ['model']
        )

        self.daily_cost = Gauge(
            'vllm_daily_cost_usd',
            'Cost today in USD'
        )

        # GPU memory
        self.gpu_memory_used = Gauge(
            'vllm_gpu_memory_used_gb',
            'GPU memory used in GB'
        )

        self.gpu_memory_total = Gauge(
            'vllm_gpu_memory_total_gb',
            'Total GPU memory in GB'
        )

        # Cache metrics
        self.cache_hit_rate = Gauge(
            'vllm_cache_hit_rate',
            'Prefix cache hit rate'
        )

    def record_request(
        self,
        model: str,
        latency_ms: float,
        tokens: int,
        cost_usd: float,
        success: bool,
        cached: bool = False
    ):
        """Record request metrics"""

        # Update counters
        status = "success" if success else "failure"
        self.requests_total.labels(model=model, status=status).inc()

        if success:
            # Latency
            self.latency.labels(model=model).observe(latency_ms)

            # Tokens
            self.tokens_generated.labels(model=model).inc(tokens)
            tokens_per_sec = tokens / (latency_ms / 1000)
            self.tokens_per_second.labels(model=model).set(tokens_per_sec)

            # Cost
            self.cost_usd.labels(model=model).inc(cost_usd)

    def update_gpu_memory(self):
        """Update GPU memory metrics"""
        if torch.cuda.is_available():
            used_gb = torch.cuda.memory_allocated() / 1e9
            total_gb = torch.cuda.get_device_properties(0).total_memory / 1e9

            self.gpu_memory_used.set(used_gb)
            self.gpu_memory_total.set(total_gb)

    def export_metrics(self) -> str:
        """Export Prometheus metrics"""
        return generate_latest().decode('utf-8')

# Usage in FastAPI
from fastapi import FastAPI

app = FastAPI()
metrics = VLLMMetrics()

@app.post("/generate")
async def generate(request: GenerateRequest):
    start = time.time()

    try:
        # Generate response
        result = llm.generate(request.prompt)

        # Record success metrics
        latency_ms = (time.time() - start) * 1000
        metrics.record_request(
            model="phi-2",
            latency_ms=latency_ms,
            tokens=len(result.tokens),
            cost_usd=calculate_cost(result),
            success=True,
        )

        return result

    except Exception as e:
        # Record failure
        latency_ms = (time.time() - start) * 1000
        metrics.record_request(
            model="phi-2",
            latency_ms=latency_ms,
            tokens=0,
            cost_usd=0,
            success=False,
        )
        raise

@app.get("/metrics")
async def get_metrics():
    """Prometheus scrape endpoint"""
    metrics.update_gpu_memory()
    return Response(
        content=metrics.export_metrics(),
        media_type="text/plain"
    )

from langfuse import Langfuse
import os

# Initialize Langfuse
langfuse = Langfuse(
    public_key=os.getenv("LANGFUSE_PUBLIC_KEY"),
    secret_key=os.getenv("LANGFUSE_SECRET_KEY"),
    host=os.getenv("LANGFUSE_HOST", "http://localhost:3001"),
)

def generate_with_observability(prompt: str, user_id: str, metadata: Dict = None):
    """Generate response with full Langfuse tracing"""

    # Create trace
    trace = langfuse.trace(
        name="financial_analysis",
        user_id=user_id,
        metadata=metadata or {},
    )

    # Start generation span
    generation = trace.generation(
        name="vllm_generate",
        model="microsoft/phi-2",
        input=prompt,
        metadata={
            "quantization": "awq",
            "max_tokens": 200,
        }
    )

    # Generate
    start = time.time()
    result = llm.generate(prompt)
    latency_ms = (time.time() - start) * 1000

    # Calculate cost
    tokens_in = len(prompt) / 4
    tokens_out = len(result.tokens)
    cost_usd = ((tokens_in + tokens_out) / 1000) * 0.0001

    # End span with metrics
    generation.end(
        output=result.text,
        usage={
            "input_tokens": int(tokens_in),
            "output_tokens": tokens_out,
            "total_tokens": int(tokens_in + tokens_out),
        },
        metadata={
            "latency_ms": latency_ms,
            "cost_usd": cost_usd,
            "model": "phi-2",
        }
    )

    return result

# Usage
result = generate_with_observability(
    prompt="Analyze Apple's Q4 earnings",
    user_id="analyst_001",
    metadata={
        "team": "equity_research",
        "department": "finance",
    }
)

from typing import Callable
from dataclasses import dataclass
import time

@dataclass
class RetryConfig:
    """Retry configuration"""
    max_retries: int = 3
    initial_delay: float = 1.0
    max_delay: float = 60.0
    exponential_base: float = 2.0

def retry_with_backoff(config: RetryConfig = RetryConfig()):
    """
    Decorator: Retry with exponential backoff

    Example:
        @retry_with_backoff()
        def generate_text(prompt):
            return llm.generate(prompt)
    """
    def decorator(func: Callable) -> Callable:
        def wrapper(*args, **kwargs):
            delay = config.initial_delay

            for attempt in range(config.max_retries):
                try:
                    return func(*args, **kwargs)

                except Exception as e:
                    if attempt == config.max_retries - 1:
                        raise  # Last attempt, re-raise

                    error_type = classify_error(e)

                    # Don't retry on invalid input
                    if error_type == ErrorType.INVALID_INPUT:
                        raise

                    print(f"??  Attempt {attempt + 1} failed: {error_type.value}")
                    print(f"   Retrying in {delay:.1f}s...")
                    time.sleep(delay)

                    # Exponential backoff
                    delay = min(delay * config.exponential_base, config.max_delay)

            raise RuntimeError(f"Failed after {config.max_retries} retries")

        return wrapper
    return decorator

# Usage
@retry_with_backoff(RetryConfig(max_retries=3, initial_delay=1.0))
def generate_with_retry(prompt: str):
    """Generate with automatic retry on failure"""
    return llm.generate(prompt)

# This will retry up to 3 times with exponential backoff
result = generate_with_retry("Analyze earnings report")

from datetime import datetime, timedelta
from enum import Enum

class CircuitState(Enum):
    CLOSED = "closed"      # Normal operation
    OPEN = "open"          # Failing, reject requests
    HALF_OPEN = "half_open"  # Testing recovery

class CircuitBreaker:
    """
    Circuit breaker for fault tolerance

    Prevents cascading failures by stopping calls to
    failing services
    """

    def __init__(
        self,
        failure_threshold: int = 5,
        timeout: int = 60,
        expected_exception: type = Exception
    ):
        self.failure_threshold = failure_threshold
        self.timeout = timeout
        self.expected_exception = expected_exception

        self.failure_count = 0
        self.last_failure_time = None
        self.state = CircuitState.CLOSED

    def call(self, func: Callable, *args, **kwargs):
        """Execute function with circuit breaker protection"""

        if self.state == CircuitState.OPEN:
            # Check if timeout elapsed
            if datetime.now() - self.last_failure_time > timedelta(seconds=self.timeout):
                self.state = CircuitState.HALF_OPEN
                print("? Circuit breaker: HALF_OPEN (testing recovery)")
            else:
                raise RuntimeError("Circuit breaker OPEN - service unavailable")

        try:
            result = func(*args, **kwargs)

            # Success - reset if recovering
            if self.state == CircuitState.HALF_OPEN:
                self.state = CircuitState.CLOSED
                self.failure_count = 0
                print("? Circuit breaker: CLOSED (service recovered)")

            return result

        except self.expected_exception as e:
            self.failure_count += 1
            self.last_failure_time = datetime.now()

            if self.failure_count >= self.failure_threshold:
                self.state = CircuitState.OPEN
                print(f"? Circuit breaker: OPEN (threshold {self.failure_threshold} reached)")

            raise

# Usage
circuit_breaker = CircuitBreaker(failure_threshold=5, timeout=60)

def generate_protected(prompt: str):
    """Generate with circuit breaker protection"""
    return circuit_breaker.call(llm.generate, prompt)

# If llm.generate fails 5 times, circuit breaker opens
# Requests fail fast for 60 seconds
# Then one test request (half-open)
# If successful, normal operation resumes

import time

class RateLimiter:
    """
    Token bucket rate limiter

    Limits requests per second to prevent overload
    """

    def __init__(self, max_requests: int, time_window: float = 1.0):
        self.max_requests = max_requests
        self.time_window = time_window
        self.tokens = max_requests
        self.last_update = time.time()

    def acquire(self, tokens: int = 1) -> bool:
        """Try to acquire tokens, return True if allowed"""

        now = time.time()
        elapsed = now - self.last_update

        # Refill tokens based on elapsed time
        self.tokens = min(
            self.max_requests,
            self.tokens + (elapsed / self.time_window) * self.max_requests
        )
        self.last_update = now

        if self.tokens >= tokens:
            self.tokens -= tokens
            return True
        else:
            return False

    def wait_for_token(self, tokens: int = 1):
        """Wait until token is available"""
        while not self.acquire(tokens):
            time.sleep(0.1)

# Usage
rate_limiter = RateLimiter(max_requests=100, time_window=1.0)

@app.post("/generate")
async def generate(request: GenerateRequest):
    # Check rate limit
    if not rate_limiter.acquire():
        raise HTTPException(
            status_code=429,
            detail="Rate limit exceeded (100 req/sec)"
        )

    # Process request
    result = llm.generate(request.prompt)
    return result

def generate_with_fallback(prompt: str) -> str:
    """
    Try multiple strategies before failing

    Strategy 1: Primary model (Llama-3-8B)
    Strategy 2: Cached response (if available)
    Strategy 3: Simpler model (Phi-2)
    Strategy 4: Template response
    """

    # Try primary model
    try:
        return llm_primary.generate(prompt)

    except Exception as e:
        print(f"??  Primary model failed: {e}")

        # Fallback 1: Check cache
        cached_response = cache.get(prompt)
        if cached_response:
            print("? Returning cached response")
            return cached_response

        # Fallback 2: Try simpler model
        try:
            print("? Falling back to Phi-2")
            return llm_simple.generate(prompt)

        except Exception as e2:
            print(f"??  Fallback model also failed: {e2}")

            # Fallback 3: Template response
            return (
                "I apologize, but I'm unable to process your request right now. "
                "Please try again in a few minutes, or contact support if the issue persists."
            )

# User never sees "Internal Server Error"
# They always get SOME response

import signal

class TimeoutError(Exception):
    pass

def timeout_handler(signum, frame):
    raise TimeoutError("Request timed out")

def generate_with_timeout(prompt: str, timeout_seconds: int = 30):
    """Generate with timeout"""

    # Set timeout
    signal.signal(signal.SIGALRM, timeout_handler)
    signal.alarm(timeout_seconds)

    try:
        result = llm.generate(prompt)

        # Cancel timeout
        signal.alarm(0)
        return result

    except TimeoutError:
        print(f"? Request timed out after {timeout_seconds}s")
        return "Request timed out. Please try a shorter prompt."

# Or using asyncio
import asyncio

async def generate_with_timeout_async(prompt: str, timeout_seconds: int = 30):
    """Generate with async timeout"""

    try:
        result = await asyncio.wait_for(
            llm.generate_async(prompt),
            timeout=timeout_seconds
        )
        return result

    except asyncio.TimeoutError:
        return "Request timed out. Please try a shorter prompt."

from fastapi import FastAPI, HTTPException
from circuitbreaker import CircuitBreaker, CircuitBreakerError

app = FastAPI()

# Initialize components
circuit_breaker = CircuitBreaker(failure_threshold=5, timeout=60)
rate_limiter = RateLimiter(max_requests=100, time_window=1.0)
cache = ResponseCache(ttl=3600)

@app.post("/generate")
@retry_with_backoff(max_retries=3)
async def generate(request: GenerateRequest):
    """
    Generate with full error handling:
    - Rate limiting
    - Circuit breaker
    - Retry with backoff
    - Timeout
    - Fallback strategies
    - Caching
    """

    # Rate limiting
    if not rate_limiter.acquire():
        raise HTTPException(status_code=429, detail="Rate limit exceeded")

    # Check cache first
    cached = cache.get(request.prompt)
    if cached:
        return {"text": cached, "cached": True}

    try:
        # Circuit breaker protection
        result = circuit_breaker.call(
            generate_with_timeout,
            request.prompt,
            timeout_seconds=30
        )

        # Cache successful response
        cache.set(request.prompt, result)

        return {"text": result, "status": "success"}

    except CircuitBreakerError:
        # Circuit breaker open - return fallback
        return {
            "text": "Service temporarily unavailable. Using cached response.",
            "status": "degraded",
            "fallback": True
        }

    except TimeoutError:
        raise HTTPException(status_code=504, detail="Request timed out")

    except Exception as e:
        # Log error
        logger.error(f"Generation failed: {e}")

        # Return graceful error
        return {
            "text": "I apologize, but I'm unable to process your request.",
            "status": "error",
            "fallback": True
        }

Request 1 (uncached): $0.00002038
Requests 2-11 (cached): $0.00000414 average

Total: $0.00010031 for 11 requests
Average: $0.0000091 per request

GCP L4 GPU: $0.45/hour = $328/month

Total monthly cost:
- Infrastructure: $328
- API costs: $2.73
- Total: $330.73/month for 10,000 requests/day

OpenAI GPT-4:
- Input: $0.03 per 1K tokens
- Output: $0.06 per 1K tokens
- Average request: 100 tokens in + 100 tokens out = $0.009
- 10,000 requests/day = $90/day = $2,700/month

Savings: $2,369/month (88% cheaper!)

Cost per request: $0.009
Daily: $450
Monthly: $13,500
Annual: $162,000

Infrastructure: $328/month
API costs (with optimizations): $13.65/month
Monthly total: $341.65
Annual: $4,100

Savings: $157,900/year (97%)

Setup time: 2 days engineering ($2,000)
Maintenance: 4 hours/month ($200/month)

Year 1:
  Savings: $157,900
  Costs: $2,000 setup + $2,400 maintenance = $4,400
  Net: $153,500 saved

ROI: 3,500% in year 1

OpenAI GPT-4: $1,350,000/year
vLLM solution: $41,000/year

Savings: $1,309,000/year (97%)

# src/services/startup_service.py
async def preload_daily_data():
    """Background job that generates brief and caches in database."""

    # 1. Fetch all data in parallel (LangGraph orchestration)
    data = await langgraph_orchestrator.fetch_all_sources()

    # 2. Generate AI brief (ReAct agent with RAG)
    brief = await brief_generator.generate(
        emails=data['emails'],
        calendar=data['calendar'],
        news=data['news'],
        weather=data['weather']
    )

    # 3. Cache everything in SQLite
    await db.set_cache('daily_brief_data', brief.to_dict(), ttl=3600)  # 1 hour TTL
    await db.set_cache('news_data', data['news'], ttl=3600)
    await db.set_cache('emails_data', data['emails'], ttl=3600)

    logger.info("? All data preloaded and cached")

# src/ui/components/daily_brief.py
def render_daily_brief_section():
    """UI just reads from cache - no AI processing!"""

    # Fast read from database (milliseconds, not seconds)
    if 'data' in st.session_state and st.session_state.data.get('daily_brief'):
        brief_data = st.session_state.data['daily_brief']
        _display_persisted_brief(brief_data)  # Instant!
    else:
        st.info("Run `make preload` to generate your first brief.")

# On macOS (what I use)
brew install ollama

# Start the service
ollama serve

# Pull the models I chose
ollama pull qwen2.5:7b         # Main LLM - fast on my M3 Mac
ollama pull nomic-embed-text   # For RAG embeddings

# Quick test
PYTHONPATH=. python -c "
import asyncio
from src.services.ollama_service import get_ollama_service

async def test():
    ollama = get_ollama_service()
    result = await ollama.generate('Explain RAG in one sentence.')
    print(result)

asyncio.run(test())
"

# Output I got:
# RAG (Retrieval-Augmented Generation) enhances LLM responses by retrieving
# relevant information from a knowledge base before generating answers.

# ? My first attempt (brittle)
async def get_daily_data():
    news = await fetch_hackernews()
    weather = await fetch_weather()
    # Later I wanted to add RSS feeds... had to modify this function
    # Then I wanted Slack... modified again
    # This was getting messy fast!

# src/connectors/hackernews.py
class HackerNewsConnector:
    """Fetches top stories from HackerNews API."""

    async def execute_async(self, max_stories: int = 10):
        """The main method MCP will call."""
        # 1. Fetch top story IDs
        response = await self.client.get(
            "https://hacker-news.firebaseio.com/v0/topstories.json"
        )
        story_ids = response.json()[:max_stories]

        # 2. Fetch each story (I fetch these in parallel for speed)
        articles = []
        for story_id in story_ids:
            story = await self._fetch_story(story_id)
            articles.append(self._convert_to_article(story))

        return articles

# src/services/mcp_server.py
class MCPServer:
    """The tool registry that AI agents query."""

    def _register_tools(self):
        # Register HackerNews
        self.tools["fetch_hackernews"] = MCPTool(
            name="fetch_hackernews",
            description="Fetch top tech stories from HackerNews with scores and comments",
            parameters={
                "max_stories": {
                    "type": "integer",
                    "description": "How many stories to fetch (1-30)",
                    "default": 10
                }
            },
            executor=HackerNewsConnector()
        )

# I tested if the AI could discover my tools
PYTHONPATH=. python -c "
from src.services.mcp_server import get_mcp_server

mcp = get_mcp_server()
print('Available tools:')
for tool in mcp.list_tools():
    print(f'  ? {tool[\"name\"]}: {tool[\"description\"]}')
"

# Output I got:
# Available tools:
#   ? fetch_hackernews: Fetch top tech stories from HackerNews...
#   ? get_current_weather: Get current weather conditions...
#   ? fetch_rss_feeds: Fetch articles from configured RSS feeds...

# src/services/rag_service.py
class RAGService:
    """Semantic memory using ChromaDB."""

    def __init__(self):
        # Initialize ChromaDB (stores on disk)
        self.client = chromadb.Client(Settings(
            persist_directory="./data/chroma_data"
        ))

        # Create collection for my articles
        self.collection = self.client.get_or_create_collection(
            name="daily_minutes"
        )

        # Ollama for creating embeddings
        self.ollama = get_ollama_service()

    async def add_document(self, content: str, metadata: dict):
        """Store a document with its vector embedding."""

        # 1. Convert text to vector (this is the magic!)
        embedding = await self.ollama.create_embeddings(content)

        # 2. Store in ChromaDB with metadata
        self.collection.add(
            documents=[content],
            embeddings=[embedding],
            metadatas=[metadata],
            ids=[hashlib.md5(content.encode()).hexdigest()]
        )

    async def search(self, query: str, max_results: int = 5):
        """Semantic search - find by meaning!"""

        # 1. Convert query to vector
        query_embedding = await self.ollama.create_embeddings(query)

        # 2. Find similar documents (cosine similarity)
        results = self.collection.query(
            query_embeddings=[query_embedding],
            n_results=max_results
        )

        return results

# I stored an article about EU AI regulations
await rag.add_document(
    content="European Union announces comprehensive AI safety regulations "
            "focusing on transparency, accountability, and privacy protection.",
    metadata={"type": "article", "topic": "ai_safety"}
)

# Later, I searched using different words
results = await rag.search("privacy rules for artificial intelligence")

# 1. News articles (for historical context)
await rag.add_article(article)

# 2. Action items from emails
await rag.add_todo(
    "Complete security training by Nov 15",
    source="email",
    priority="high"
)

# 3. Meeting context
await rag.add_document(
    "Q4 Planning Meeting - need to prepare budget estimates",
    metadata={"type": "meeting", "date": "2025-02-01"}
)

# 4. User preferences (this feeds into RLHF later!)
await rag.add_document(
    "User marked 'AI safety' topics as important",
    metadata={"type": "preference", "category": "ai_safety"}
)

# ? First attempt - no thinking!
async def generate_brief():
    news = await fetch_all_news()  # Fetches everything
    summary = await llm.generate(f"Summarize: {news}")
    return summary

# src/agents/react_agent.py
class ReActAgent:
    """Agent that thinks before acting."""

    async def run(self, goal: str):
        """Execute goal using ReAct loop."""
        steps = []
        observations = []

        for step_num in range(1, self.max_steps + 1):
            # 1. THOUGHT: Ask AI what to do next
            thought = await self._generate_thought(goal, steps, observations)

            # Check if we're done
            if "FINAL ANSWER" in thought:
                return self._extract_answer(thought)

            # 2. ACTION: Parse what action to take
            action = self._parse_action(thought)
            # Example: {"action": "call_tool", "tool": "fetch_hackernews"}

            # 3. EXECUTE: Run the action via MCP
            observation = await self._execute_action(action)
            observations.append(observation)

            # Record this step for debugging
            steps.append({
                "thought": thought,
                "action": action,
                "observation": observation
            })

        return {"steps": steps, "answer": "Max steps reached"}

async def _generate_thought(self, goal, steps, observations):
    """Generate next reasoning step."""

    prompt = f"""Goal: {goal}

Previous steps:
{self._format_steps(steps)}

Available actions:
- query_rag(query): Search my semantic memory
- call_tool(name, params): Execute an MCP tool
- FINAL ANSWER: When you have everything needed

Think step-by-step. What should I do next?

Format your response as:
THOUGHT: <your reasoning>
ACTION: <action to take>
"""

    return await self.ollama.generate(prompt, temperature=0.7)

? Goal: Generate my daily brief

Step 1:
  ? THOUGHT: I need to gather news, check weather, and see user preferences
  ? ACTION: call_tool("fetch_hackernews", max_stories=10)
  ?? OBSERVATION: Fetched 10 articles about AI, privacy, and tech

Step 2:
  ? THOUGHT: Got news. User preferences would help prioritize.
  ? ACTION: query_rag("user interests and preferences")
  ?? OBSERVATION: User cares about AI safety, security, privacy

Step 3:
  ? THOUGHT: Should filter articles to user's interests
  ? ACTION: call_tool("get_current_weather", location="Seattle")
  ?? OBSERVATION: 70°F, Partly cloudy

Step 4:
  ? THOUGHT: I have news (filtered by user interests), weather. Ready to generate.
  ? ACTION: FINAL ANSWER
  ? Generated personalized brief highlighting AI safety articles

? "Newsletter: 10 CSS Tips" ? Importance: 0.5
? "URGENT: Production outage!" ? Importance: 0.5

# src/models/email.py
class ImportanceScoringMixin:
    """Learn from user feedback."""

    importance_score: float = 0.5  # AI's base score
    boost_labels: Set[str] = set()  # Words user marked important
    filter_labels: Set[str] = set()  # Words user wants to skip

    def apply_rlhf_boost(self, content_text: str) -> float:
        """Adjust score based on learned preferences."""
        adjusted = self.importance_score
        content_lower = content_text.lower()

        # Boost if content matches important keywords
        for label in self.boost_labels:
            if label.lower() in content_lower:
                adjusted += 0.1  # Bump up priority!

        # Penalize if content matches skip keywords
        for label in self.filter_labels:
            if label.lower() in content_lower:
                adjusted -= 0.2  # Push down priority!

        # Keep in valid range [0, 1]
        return max(0.0, min(1.0, adjusted))

# User sees an email
for email in emails:
    col1, col2, col3 = st.columns([8, 1, 1])

    with col1:
        st.write(f"**{email.subject}**")
        st.info(email.snippet)

    with col2:
        if st.button("?", key=f"important_{email.id}"):
            # Extract what made this important
            keywords = await extract_keywords(email.subject + email.body)
            # Add to boost labels
            user_profile.boost_labels.update(keywords)
            st.success(f"? Learned: You care about {', '.join(keywords)}")

    with col3:
        if st.button("?", key=f"skip_{email.id}"):
            # Learn to deprioritize these
            keywords = await extract_keywords(email.subject)
            user_profile.filter_labels.update(keywords)
            st.success(f"? Will deprioritize: {', '.join(keywords)}")

# ? Sequential execution - SLOW!
news = await fetch_news()      # 5 seconds
emails = await fetch_emails()  # 3 seconds
calendar = await fetch_calendar()  # 2 seconds
weather = await fetch_weather()  # 1 second
# Total: 11 seconds just waiting! ?

# src/services/langgraph_orchestrator.py
from langgraph.graph import StateGraph, END

class LangGraphOrchestrator:
    def _create_workflow(self):
        """Define my workflow graph."""
        workflow = StateGraph(WorkflowState)

        # Add nodes (processing steps)
        workflow.add_node("analyze", self._analyze_request)
        workflow.add_node("fetch_news", self._fetch_news)
        workflow.add_node("fetch_emails", self._fetch_emails)
        workflow.add_node("fetch_calendar", self._fetch_calendar)
        workflow.add_node("search_rag", self._search_context)
        workflow.add_node("generate_summary", self._generate_summary)

        # Define edges (execution flow)
        workflow.set_entry_point("analyze")

        # Parallel fetch (all happen at once!)
        workflow.add_edge("analyze", "fetch_news")
        workflow.add_edge("analyze", "fetch_emails")
        workflow.add_edge("analyze", "fetch_calendar")

        # All converge to RAG search
        workflow.add_edge("fetch_news", "search_rag")
        workflow.add_edge("fetch_emails", "search_rag")
        workflow.add_edge("fetch_calendar", "search_rag")

        # Sequential processing
        workflow.add_edge("search_rag", "generate_summary")
        workflow.add_edge("generate_summary", END)

        return workflow.compile()

async def _fetch_news(self, state: WorkflowState):
    """Fetch news in parallel."""
    try:
        articles = await self.mcp.execute_tool(
            "fetch_hackernews",
            {"max_stories": 10}
        )
        state["news_articles"] = articles
    except Exception as e:
        state["errors"].append(f"News fetch failed: {e}")
        state["news_articles"] = []

    return state

async def _search_context(self, state: WorkflowState):
    """Search RAG for relevant context."""
    query = state["user_request"]
    results = await self.rag.search(query, max_results=5)

    # Build context string
    context = "\n".join([r['content'] for r in results])
    state["context"] = context

    return state

# Execute the complete workflow
result = await orchestrator.run("Generate my daily brief")

# I get back:
{
    "news_articles": [...],      # 10 articles
    "emails": [...],              # 5 unread
    "calendar_events": [...],     # 3 events today
    "context": "...",             # RAG context
    "summary": "...",             # Generated brief
    "processing_time": 5.2        # Seconds (not 11!)
}

# src/services/llm_factory.py
def get_llm_service():
    """Factory pattern - works with any LLM provider."""
    provider = os.getenv("LLM_PROVIDER", "ollama")

    if provider == "ollama":
        return OllamaService(
            base_url=os.getenv("OLLAMA_BASE_URL", "http://localhost:11434"),
            model=os.getenv("OLLAMA_MODEL", "qwen2.5:7b")
        )
    elif provider == "openai":
        return OpenAIService(
            api_key=os.getenv("OPENAI_API_KEY"),
            model=os.getenv("OPENAI_MODEL", "gpt-4-turbo")
        )
    elif provider == "google":
        # Like in my previous Vertex AI article!
        return VertexAIService(
            project_id=os.getenv("GCP_PROJECT_ID"),
            model="gemini-1.5-flash"
        )

    raise ValueError(f"Unknown provider: {provider}")

# All services implement the same interface:
class BaseLLMService:
    async def generate(self, prompt: str, **kwargs) -> str:
        """Generate text from prompt."""
        raise NotImplementedError

    async def create_embeddings(self, text: str) -> List[float]:
        """Create vector embeddings."""
        raise NotImplementedError

? "Today's news features a mix of technology updates and various topics."

# ? Better prompt with strict rules
prompt = f"""Generate a daily brief following these STRICT rules:

PRIORITY ORDER (most important first):
1. Urgent emails or action items
2. Today's calendar events
3. Market/business news
4. Tech news

TLDR FORMAT (exactly 3 bullets, be SPECIFIC):
* Bullet 1: Most urgent email/action (include WHO, WHAT, WHEN)
   Example: "Client escalation from Acme Corp affecting 50K users - response needed by 2pm"

* Bullet 2: Most important calendar event today (include TIME and WHAT TO PREPARE)
   Example: "2pm: Board meeting - prepare Q4 revenue slides"

* Bullet 3: Top market/business news (include NUMBERS/SPECIFICS)
   Example: "Federal Reserve raises rates 0.5% to 5.25% - affects tech hiring"

AVOID THESE PHRASES (they're too vague):
? "mix of updates"
? "various topics"
? "continues to make progress"
? "interesting developments"

USE SPECIFIC DETAILS:
? Names (people, companies)
? Numbers (percentages, dollar amounts, deadlines)
? Times (when something happened or needs to happen)

Content to summarize:
{content}

Generate: TLDR (3 bullets), Summary (5-6 detailed sentences), Key Insights (5 bullets)
"""

? • Critical email... • Meeting at 2pm... • Market news...

# ? Doesn't work
st.info(tldr)

# ? Works!
tldr_lines = [line.strip() for line in tldr.split('\n') if line.strip()]
for bullet in tldr_lines:
    st.markdown(bullet)

? TLDR bullet 1: "OpenAI releases GPT-5" (who cares?)
   TLDR bullet 2: "Crypto market surges" (not relevant to me)
   TLDR bullet 3: "Client escalation requires response" (BURIED!)

# src/services/brief_scheduler.py
async def _generate_daily_brief(emails, calendar, news, weather):
    """Generate prioritized daily brief with structured prompt."""

    # Separate market vs tech news (market is higher priority)
    market_news = [n for n in news if 'market' in n.tags]
    tech_news = [n for n in news if 'market' not in n.tags]

    # Sort emails by RLHF-boosted importance score
    important_emails = sorted(
        emails,
        key=lambda e: e.apply_rlhf_boost(e.subject + e.snippet),
        reverse=True
    )[:5]  # Top 5 only

    # Build structured prompt with clear priority
    prompt = f"""
**SECTION 1: IMPORTANT EMAILS (HIGHEST PRIORITY - use for TLDR bullet #1)**
{format_emails(important_emails)}

**SECTION 2: TODAY'S CALENDAR (SECOND PRIORITY - use for TLDR bullet #2)**
{format_calendar(calendar)}

**SECTION 3: MARKET NEWS (THIRD PRIORITY - use for TLDR bullet #3)**
{format_market_news(market_news)}

**SECTION 4: TECH NEWS (LOWEST PRIORITY - summarize briefly)**
{format_tech_news(tech_news)}

**SECTION 5: WEATHER**
{format_weather(weather)}

Generate a daily brief following this EXACT priority order:
1. Email action items FIRST
2. Calendar events SECOND
3. Market/business news THIRD
4. Tech news LAST (brief mention only)

TLDR must have EXACTLY 3 bullets using content from sections 1, 2, 3 (not section 4).
"""

    return await llm.generate(prompt)

Query: "AI safety regulations"
Result: Article about "safe AI models for healthcare" (wrong context!)

# Store with rich metadata
await rag.add_document(
    content=article.title,
    metadata={
        "type": "article",
        "category": "ai_safety",  # For filtering!
        "tags": ["regulation", "eu", "policy"],
        "date": "2025-01-28",
        "importance": "high"
    }
)

# Search with filters
results = await rag.search(
    "AI regulations",
    filter_metadata={
        "category": "ai_safety",
        "importance": "high"
    }
)

  # Current: Simple keyword match
  if "security" in email:
      score += 0.1

  # Better: Contextual understanding
  if embeddings_similar(email, user.important_emails):
      score += contextual_boost  # Uses semantic similarity!

# src/connectors/slack.py
class SlackConnector:
    """Fetch recent messages from Slack channels."""

    async def execute_async(self, channel: str, max_messages: int = 10):
        # 1. Connect to Slack API
        client = WebClient(token=os.getenv("SLACK_BOT_TOKEN"))

        # 2. Fetch recent messages
        response = await client.conversations_history(
            channel=channel,
            limit=max_messages
        )

        # 3. Convert to standard format
        messages = []
        for msg in response['messages']:
            messages.append(SlackMessage(
                text=msg['text'],
                user=msg['user'],
                channel=channel,
                timestamp=msg['ts']
            ))

        return messages

# src/services/mcp_server.py
def _register_tools(self):
    # ... existing tools ...

    # Add Slack
    self.tools["get_slack_messages"] = MCPTool(
        name="get_slack_messages",
        description="Fetch recent Slack messages from a channel",
        parameters={
            "channel": {"type": "string", "description": "Channel name"},
            "max_messages": {"type": "integer", "default": 10}
        },
        executor=SlackConnector()
    )

# Your ReAct agent will now see:
# "Available tools: fetch_hackernews, get_slack_messages, ..."
# No changes needed to ReAct logic!

# src/services/brief_scheduler.py
prompt = f"""
**IMPORTANT EMAILS**: {emails}
**CALENDAR**: {calendar}
**SLACK HIGHLIGHTS**: {slack_messages}  # New!
**NEWS**: {news}

Generate brief prioritizing: Email > Calendar > Slack > News
"""

# .env.development
LLM_PROVIDER=ollama
LLM_OLLAMA_BASE_URL=http://localhost:11434
LLM_OLLAMA_MODEL=qwen2.5:7b
DATABASE_URL=sqlite:///./data/daily_minutes.db
REDIS_URL=redis://localhost:6379

# .env.production
LLM_PROVIDER=openai  # Just change this line!
OPENAI_API_KEY=sk-...
OPENAI_MODEL=gpt-4-turbo
DATABASE_URL=postgresql://...  # Scalable DB
REDIS_URL=redis://prod-cluster:6379  # Distributed cache

# src/services/llm_factory.py
def get_llm_service():
    """Factory pattern - works with any LLM."""
    provider = os.getenv("LLM_PROVIDER", "ollama")

    if provider == "ollama":
        return OllamaService()
    elif provider == "openai":
        return OpenAIService()
    elif provider == "anthropic":
        return ClaudeService()
    elif provider == "google":
        return VertexAIService()  # Like in my previous article!

    raise ValueError(f"Unknown provider: {provider}")

# Test MCP tool registration
pytest tests/unit/test_mcp_server.py -v

# Test RAG semantic search
pytest tests/unit/test_rag_service.py -v

# Test ReAct reasoning
pytest tests/unit/test_react_agent.py -v

# Test RLHF scoring
pytest tests/unit/test_rlhf_scoring.py -v

# Run all unit tests
pytest tests/unit/ -v
# 516 passed in 45.23s ?

# tests/integration/test_brief_quality.py
async def test_tldr_has_three_bullets():
    """TLDR must have exactly 3 bullets."""
    brief = await db.get_cache('daily_brief_data')
    tldr = brief.get('tldr', '')

    bullets = [line for line in tldr.split('\n') if line.strip().startswith('•')]

    assert len(bullets) == 3, f"Expected 3 bullets, got {len(bullets)}"
    assert "email" in bullets[0].lower() or "urgent" in bullets[0].lower()
    assert "calendar" in bullets[1].lower() or "meeting" in bullets[1].lower()

async def test_no_generic_phrases():
    """Brief should not contain vague phrases."""
    brief = await db.get_cache('daily_brief_data')
    summary = brief.get('summary', '')

    bad_phrases = ["mix of updates", "various topics", "continues to"]
    for phrase in bad_phrases:
        assert phrase not in summary.lower(), f"Found generic phrase: {phrase}"

# 1. Fetch data and generate brief
make preload

# Output I see:
# ? Fetching news from HackerNews... (10 articles)
# ? Fetching weather... (70°F, Sunny)
# ? Analyzing articles with AI... (15 articles)
# ? Generating daily brief... (Done in 18.3s)
# ? Brief saved to database

# 2. Launch UI
streamlit run src/ui/streamlit_app.py

# 3. Check brief quality
# - Is TLDR specific? (not vague)
# - Are priorities correct? (email > calendar > news)
# - Are action items extracted? (from emails)
# - Did RLHF work? (boosted my preferences)

# Clone my repo
git clone https://github.com/bhatti/daily-minutes
cd daily-minutes

# Install dependencies
pip install -r requirements.txt

# Setup Ollama
ollama pull qwen2.5:7b
ollama pull nomic-embed-text

# Generate your first brief
make preload

# Launch dashboard
streamlit run src/ui/streamlit_app.py