Monitoring va Observability

1from dataclasses import dataclass, field
2from collections import defaultdict
3import time
4from typing import Dict, List
5
6@dataclass
7class LLMMetrics:
8    total_requests: int = 0
9    total_tokens: int = 0
10    total_cost: float = 0.0
11    latencies: List[float] = field(default_factory=list)
12    errors: int = 0
13    model_usage: Dict[str, int] = field(default_factory=lambda: defaultdict(int))
14    
15    def record_request(self, model, tokens, latency, cost):
16        self.total_requests += 1
17        self.total_tokens += tokens
18        self.total_cost += cost
19        self.latencies.append(latency)
20        self.model_usage[model] += 1
21    
22    def record_error(self):
23        self.errors += 1
24    
25    @property
26    def avg_latency(self):
27        return sum(self.latencies) / len(self.latencies) if self.latencies else 0
28    
29    @property
30    def error_rate(self):
31        return self.errors / self.total_requests if self.total_requests else 0
32    
33    def summary(self):
34        return {
35            "total_requests": self.total_requests,
36            "total_tokens": self.total_tokens,
37            "total_cost": f"${self.total_cost:.4f}",
38            "avg_latency": f"{self.avg_latency:.2f}s",
39            "error_rate": f"{self.error_rate:.1%}",
40            "model_usage": dict(self.model_usage)
41        }
42
43metrics = LLMMetrics()

1from langchain_core.callbacks import BaseCallbackHandler
2import time
3
4class MonitoringCallback(BaseCallbackHandler):
5    def __init__(self, metrics: LLMMetrics):
6        self.metrics = metrics
7        self.start_time = None
8    
9    def on_llm_start(self, serialized, prompts, **kwargs):
10        self.start_time = time.time()
11    
12    def on_llm_end(self, response, **kwargs):
13        latency = time.time() - self.start_time
14        
15        if response.llm_output:
16            usage = response.llm_output.get("token_usage", {})
17            model = response.llm_output.get("model_name", "unknown")
18            tokens = usage.get("total_tokens", 0)
19            cost = calculate_cost(model, usage)
20            
21            self.metrics.record_request(model, tokens, latency, cost)
22    
23    def on_llm_error(self, error, **kwargs):
24        self.metrics.record_error()
25
26def calculate_cost(model, usage):
27    pricing = {
28        "gpt-4o-mini": {"input": 0.15, "output": 0.60},
29        "gpt-4o": {"input": 2.50, "output": 10.00},
30    }
31    rates = pricing.get(model, {"input": 0, "output": 0})
32    input_cost = (usage.get("prompt_tokens", 0) / 1_000_000) * rates["input"]
33    output_cost = (usage.get("completion_tokens", 0) / 1_000_000) * rates["output"]
34    return input_cost + output_cost

1from fastapi import FastAPI
2
3app = FastAPI()
4
5@app.get("/metrics")
6async def get_metrics():
7    return metrics.summary()
8
9@app.get("/health")
10async def health():
11    return {
12        "status": "healthy",
13        "uptime": time.time() - app.state.start_time,
14        "total_requests": metrics.total_requests,
15        "error_rate": metrics.error_rate
16    }

1import logging
2import json
3from datetime import datetime
4
5class JSONFormatter(logging.Formatter):
6    def format(self, record):
7        log_data = {
8            "timestamp": datetime.utcnow().isoformat(),
9            "level": record.levelname,
10            "message": record.getMessage(),
11            "module": record.module,
12        }
13        if hasattr(record, "extra_data"):
14            log_data.update(record.extra_data)
15        return json.dumps(log_data)
16
17logger = logging.getLogger("ai-api")
18handler = logging.StreamHandler()
19handler.setFormatter(JSONFormatter())
20logger.addHandler(handler)
21logger.setLevel(logging.INFO)
22
23# Usage
24logger.info("LLM request completed", extra={
25    "extra_data": {
26        "model": "gpt-4o-mini",
27        "tokens": 150,
28        "latency_ms": 450,
29        "session_id": "abc123"
30    }
31})

1async def check_alerts():
2    if metrics.error_rate > 0.05:  # 5% error rate
3        await send_alert("High error rate", f"Error rate: {metrics.error_rate:.1%}")
4    
5    if metrics.avg_latency > 5.0:  # 5 second avg
6        await send_alert("High latency", f"Avg latency: {metrics.avg_latency:.1f}s")
7    
8    if metrics.total_cost > 100:  # $100 daily limit
9        await send_alert("Cost alert", f"Daily cost: ${metrics.total_cost:.2f}")