observability.py

# File: llm_observability.py
import sqlite3
import json
from datetime import datetime
from typing import Dict, Any, List, Optional, Callable
import logging
import functools

logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)

def log_execution(func: Callable) -> Callable:
    @functools.wraps(func)
    def wrapper(*args: Any, **kwargs: Any) -> Any:
        logger.info(f"Executing {func.__name__}")
        try:
            result = func(*args, **kwargs)
            logger.info(f"{func.__name__} completed successfully")
            return result
        except Exception as e:
            logger.error(f"Error in {func.__name__}: {e}")
            raise
    return wrapper


class LLMObservabilityManager:
    def __init__(self, db_path: str = "/data/llm_observability_v2.db"):
        self.db_path = db_path
        self.create_table()

    def create_table(self):
        with sqlite3.connect(self.db_path) as conn:
            cursor = conn.cursor()
            cursor.execute('''
                CREATE TABLE IF NOT EXISTS llm_observations (
                    id INTEGER PRIMARY KEY AUTOINCREMENT,
                    conversation_id TEXT,
                    created_at DATETIME,
                    status TEXT,
                    request TEXT,
                    response TEXT,
                    model TEXT,
                    prompt_tokens INTEGER,
                    completion_tokens INTEGER,
                    total_tokens INTEGER,
                    cost FLOAT,
                    latency FLOAT,
                    user TEXT
                )
            ''')

    def insert_observation(self, response: str, conversation_id: str, status: str, request: str, model: str, prompt_tokens: int,completion_tokens: int, total_tokens: int, cost: float, latency: float, user: str):
        created_at = datetime.now()
        
        with sqlite3.connect(self.db_path) as conn:
            cursor = conn.cursor()
            cursor.execute('''
                INSERT INTO llm_observations 
                (conversation_id, created_at, status, request, response, model, prompt_tokens, completion_tokens,total_tokens, cost, latency, user)
                VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
            ''', (
                conversation_id,
                created_at,
                status,
                request,
                response,
                model,
                prompt_tokens,
                completion_tokens,
                total_tokens,
                cost,
                latency,
                user
            ))

    def get_observations(self, conversation_id: Optional[str] = None) -> List[Dict[str, Any]]:
        with sqlite3.connect(self.db_path) as conn:
            cursor = conn.cursor()
            if conversation_id:
                cursor.execute('SELECT * FROM llm_observations WHERE conversation_id = ? ORDER BY created_at', (conversation_id,))
            else:
                cursor.execute('SELECT * FROM llm_observations ORDER BY created_at')
            rows = cursor.fetchall()

            column_names = [description[0] for description in cursor.description]
            return [dict(zip(column_names, row)) for row in rows]

    def get_all_observations(self) -> List[Dict[str, Any]]:
        return self.get_observations()
    
    def get_all_unique_conversation_observations(self, limit: Optional[int] = None) -> List[Dict[str, Any]]:
        with sqlite3.connect(self.db_path) as conn:
            cursor = conn.cursor()
            # Get the latest observation for each unique conversation_id
            query = '''
                SELECT * FROM llm_observations o1
                WHERE created_at = (
                    SELECT MAX(created_at) 
                    FROM llm_observations o2 
                    WHERE o2.conversation_id = o1.conversation_id
                )
                ORDER BY created_at DESC
            '''
            if limit is not None:
                query += f' LIMIT {limit}'
                
            cursor.execute(query)
            rows = cursor.fetchall()
            
            column_names = [description[0] for description in cursor.description]
            return [dict(zip(column_names, row)) for row in rows]

    def get_dashboard_statistics(self, days: Optional[int] = None, time_series_interval: str = 'day') -> Dict[str, Any]:
        """
        Get statistical metrics for LLM usage dashboard with time series data.
        
        Args:
            days (int, optional): Number of days to look back. If None, returns all-time statistics
            time_series_interval (str): Interval for time series data ('hour', 'day', 'week', 'month')
            
        Returns:
            Dict containing dashboard statistics and time series data
        """
        def safe_round(value: Any, decimals: int = 2) -> float:
            """Safely round a value, returning 0 if the value is None or invalid."""
            try:
                return round(float(value), decimals) if value is not None else 0.0
            except (TypeError, ValueError):
                return 0.0
    
        def safe_divide(numerator: Any, denominator: Any, decimals: int = 2) -> float:
            """Safely divide two numbers, handling None and zero division."""
            try:
                if not denominator or denominator is None:
                    return 0.0
                return round(float(numerator or 0) / float(denominator), decimals)
            except (TypeError, ValueError):
                return 0.0
    
        try:
            with sqlite3.connect(self.db_path) as conn:
                cursor = conn.cursor()
                
                # Build time filter
                time_filter = ""
                if days is not None:
                    time_filter = f"WHERE created_at >= datetime('now', '-{days} days')"
                
                # Get general statistics
                cursor.execute(f"""
                    SELECT 
                        COUNT(*) as total_requests,
                        COUNT(DISTINCT conversation_id) as unique_conversations,
                        COUNT(DISTINCT user) as unique_users,
                        SUM(total_tokens) as total_tokens,
                        SUM(cost) as total_cost,
                        AVG(latency) as avg_latency,
                        SUM(CASE WHEN status = 'error' THEN 1 ELSE 0 END) as error_count
                    FROM llm_observations
                    {time_filter}
                """)
                row = cursor.fetchone()
                if not row:
                    return self._get_empty_statistics()
                    
                general_stats = dict(zip([col[0] for col in cursor.description], row))
                
                # Get model distribution
                cursor.execute(f"""
                    SELECT model, COUNT(*) as count
                    FROM llm_observations
                    {time_filter}
                    GROUP BY model
                    ORDER BY count DESC
                """)
                model_distribution = {row[0]: row[1] for row in cursor.fetchall()} if cursor.fetchall() else {}
                
                # Get average tokens per request
                cursor.execute(f"""
                    SELECT 
                        AVG(prompt_tokens) as avg_prompt_tokens,
                        AVG(completion_tokens) as avg_completion_tokens
                    FROM llm_observations
                    {time_filter}
                """)
                token_averages = dict(zip([col[0] for col in cursor.description], cursor.fetchone()))
                
                # Get top users by request count
                cursor.execute(f"""
                    SELECT user, COUNT(*) as request_count, 
                           SUM(total_tokens) as total_tokens,
                           SUM(cost) as total_cost
                    FROM llm_observations
                    {time_filter}
                    GROUP BY user
                    ORDER BY request_count DESC
                    LIMIT 5
                """)
                top_users = [
                    {
                        "user": row[0],
                        "request_count": row[1],
                        "total_tokens": row[2] or 0,
                        "total_cost": safe_round(row[3])
                    }
                    for row in cursor.fetchall()
                ]
    
                # Get time series data
                time_series_format = {
                    'hour': "%Y-%m-%d %H:00:00",
                    'day': "%Y-%m-%d",
                    'week': "%Y-%W",
                    'month': "%Y-%m"
                }
                
                format_string = time_series_format.get(time_series_interval, "%Y-%m-%d")
                
                cursor.execute(f"""
                    SELECT 
                        strftime('{format_string}', created_at) as time_bucket,
                        COUNT(*) as request_count,
                        SUM(total_tokens) as total_tokens,
                        SUM(cost) as total_cost,
                        AVG(latency) as avg_latency,
                        COUNT(DISTINCT user) as unique_users,
                        SUM(CASE WHEN status = 'error' THEN 1 ELSE 0 END) as error_count
                    FROM llm_observations
                    {time_filter}
                    GROUP BY time_bucket
                    ORDER BY time_bucket
                """)
                
                time_series = [
                    {
                        "timestamp": row[0],
                        "request_count": row[1] or 0,
                        "total_tokens": row[2] or 0,
                        "total_cost": safe_round(row[3]),
                        "avg_latency": safe_round(row[4]),
                        "unique_users": row[5] or 0,
                        "error_count": row[6] or 0
                    }
                    for row in cursor.fetchall()
                ]
    
                # Calculate trends safely
                trends = self._calculate_trends(time_series)
    
                return {
                    "general_stats": {
                        "total_requests": general_stats["total_requests"] or 0,
                        "unique_conversations": general_stats["unique_conversations"] or 0,
                        "unique_users": general_stats["unique_users"] or 0,
                        "total_tokens": general_stats["total_tokens"] or 0,
                        "total_cost": safe_round(general_stats["total_cost"]),
                        "avg_latency": safe_round(general_stats["avg_latency"]),
                        "error_rate": safe_round(
                            safe_divide(general_stats["error_count"], general_stats["total_requests"]) * 100
                        )
                    },
                    "model_distribution": model_distribution,
                    "token_metrics": {
                        "avg_prompt_tokens": safe_round(token_averages["avg_prompt_tokens"]),
                        "avg_completion_tokens": safe_round(token_averages["avg_completion_tokens"])
                    },
                    "top_users": top_users,
                    "time_series": time_series,
                    "trends": trends
                }
        except sqlite3.Error as e:
            logger.error(f"Database error in get_dashboard_statistics: {e}")
            return self._get_empty_statistics()
        except Exception as e:
            logger.error(f"Error in get_dashboard_statistics: {e}")
            return self._get_empty_statistics()
    
    def _get_empty_statistics(self) -> Dict[str, Any]:
        """Return an empty statistics structure when no data is available."""
        return {
            "general_stats": {
                "total_requests": 0,
                "unique_conversations": 0,
                "unique_users": 0,
                "total_tokens": 0,
                "total_cost": 0.0,
                "avg_latency": 0.0,
                "error_rate": 0.0
            },
            "model_distribution": {},
            "token_metrics": {
                "avg_prompt_tokens": 0.0,
                "avg_completion_tokens": 0.0
            },
            "top_users": [],
            "time_series": [],
            "trends": {
                "request_trend": 0.0,
                "cost_trend": 0.0,
                "token_trend": 0.0
            }
        }
    
    def _calculate_trends(self, time_series: List[Dict[str, Any]]) -> Dict[str, float]:
        """Calculate trends safely from time series data."""
        if len(time_series) >= 2:
            current = time_series[-1]
            previous = time_series[-2]
            return {
                "request_trend": self._calculate_percentage_change(
                    previous["request_count"], current["request_count"]),
                "cost_trend": self._calculate_percentage_change(
                    previous["total_cost"], current["total_cost"]),
                "token_trend": self._calculate_percentage_change(
                    previous["total_tokens"], current["total_tokens"])
            }
        return {
            "request_trend": 0.0,
            "cost_trend": 0.0,
            "token_trend": 0.0
        }
    
    def _calculate_percentage_change(self, old_value: Any, new_value: Any) -> float:
        """Calculate percentage change between two values safely."""
        try:
            old_value = float(old_value or 0)
            new_value = float(new_value or 0)
            if old_value == 0:
                return 100.0 if new_value > 0 else 0.0
            return round(((new_value - old_value) / old_value) * 100, 2)
        except (TypeError, ValueError):
            return 0.0