Update llm/metrics.py

llm/metrics.py

@@ -1,280 +1,119 @@
-"""Coleta e análise de métricas de uso de LLM."""
+"""
+llm/metrics.py
+Coleta e análise de métricas de uso de LLM
+"""
 
-from typing import Dict, List, Optional, Tuple
-from dataclasses import dataclass, field, asdict
-from datetime import datetime, timedelta
-from collections import defaultdict
-import json
 import logging
+from typing import Dict, Any
+from datetime import datetime
+from collections import defaultdict
 
 logger = logging.getLogger(__name__)
 
-
-@dataclass
-class RequestMetric:
-    """Métrica de uma requisição individual."""
-    timestamp: datetime
-    provider: str
-    model: str
-    tokens_input: int
-    tokens_output: int
-    cost_usd: float
-    latency_ms: float
-    success: bool
-    error_msg: Optional[str] = None
-    user_id: Optional[str] = None
-    request_id: Optional[str] = None
-
-
-@dataclass
-class ProviderStats:
-    """Estatísticas de um provedor."""
-    provider: str
-    total_requests: int = 0
-    successful_requests: int = 0
-    failed_requests: int = 0
-    total_tokens: int = 0
-    total_cost: float = 0.0
-    average_latency_ms: float = 0.0
-    min_latency_ms: float = float('inf')
-    max_latency_ms: float = 0.0
-    average_tokens: float = 0.0
-
-
-class MetricsCollector:
-    """Coletor centralizado de métricas."""
+class LLMMetrics:
+    """
+    Coleta métricas de processamento LLM.
+    
+    Rastreia:
+    - Taxa de sucesso/falha por provedor
+    - Tempo de processamento
+    - Tokens utilizados
+    - Custo estimado
+    """
     
-    def __init__(self, window_size: int = 10000):
-        """
-        Inicializa o coletor.
+    def __init__(self):
+        self.metricas = defaultdict(lambda: {
+            "tentativas": 0,
+            "sucessos": 0,
+            "falhas": 0,
+            "tempo_total_segundos": 0.0,
+            "tokens_total": 0,
+            "custo_total_usd": 0.0,
+        })
         
-        Args:
-            window_size: Número máximo de métricas em memória
-        """
-        self.window_size = window_size
-        self.metrics: List[RequestMetric] = []
-        self.aggregates: Dict[str, Dict] = defaultdict(dict)
+        # Preços aproximados por provedor (USD por 1M tokens)
+        self.precos = {
+            "groq": 0.000100,      # Mais barato
+            "openai": 0.015000,     # GPT-4 input
+            "anthropic": 0.010000,  # Claude
+        }
     
-    def record_request(self, 
-                      provider: str,
-                      model: str,
-                      tokens_input: int,
-                      tokens_output: int,
-                      cost_usd: float,
-                      latency_ms: float,
-                      success: bool,
-                      error_msg: Optional[str] = None,
-                      user_id: Optional[str] = None,
-                      request_id: Optional[str] = None):
-        """Registra métrica de requisição."""
-        metric = RequestMetric(
-            timestamp=datetime.now(),
-            provider=provider,
-            model=model,
-            tokens_input=tokens_input,
-            tokens_output=tokens_output,
-            cost_usd=cost_usd,
-            latency_ms=latency_ms,
-            success=success,
-            error_msg=error_msg,
-            user_id=user_id,
-            request_id=request_id,
+    def registrar_sucesso(
+        self,
+        provider: str,
+        tempo_segundos: float,
+        tokens: int,
+    ) -> None:
+        """Registra processamento bem-sucedido"""
+        metrica = self.metricas[provider]
+        metrica["tentativas"] += 1
+        metrica["sucessos"] += 1
+        metrica["tempo_total_segundos"] += tempo_segundos
+        metrica["tokens_total"] += tokens
+        
+        # Calcular custo
+        custo = (tokens / 1_000_000) * self.precos.get(provider, 0.0)
+        metrica["custo_total_usd"] += custo
+        
+        logger.info(
+            f"Sucesso {provider}: {tempo_segundos:.2f}s, "
+            f"{tokens} tokens, ${custo:.6f}"
         )
-        
-        self.metrics.append(metric)
-        
-        # Manter janela de tamanho máximo
-        if len(self.metrics) > self.window_size:
-            self.metrics.pop(0)
-        
-        logger.debug(f"Métrica registrada: {provider}/{model}")
     
-    def get_provider_stats(self, provider: Optional[str] = None) -> Dict[str, ProviderStats]:
-        """
-        Retorna estatísticas por provedor.
-        
-        Args:
-            provider: Se especificado, retorna apenas para este provedor
-        
-        Returns:
-            Dict de estatísticas
-        """
-        stats = defaultdict(lambda: ProviderStats(provider=""))
-        
-        for metric in self.metrics:
-            if provider and metric.provider != provider:
-                continue
-            
-            stat = stats[metric.provider]
-            if not stat.provider:
-                stat.provider = metric.provider
-            
-            stat.total_requests += 1
-            if metric.success:
-                stat.successful_requests += 1
-            else:
-                stat.failed_requests += 1
-            
-            stat.total_tokens += metric.tokens_input + metric.tokens_output
-            stat.total_cost += metric.cost_usd
-            stat.min_latency_ms = min(stat.min_latency_ms, metric.latency_ms)
-            stat.max_latency_ms = max(stat.max_latency_ms, metric.latency_ms)
-        
-        # Calcular médias
-        for stat in stats.values():
-            if stat.total_requests > 0:
-                stat.average_latency_ms = sum(
-                    m.latency_ms for m in self.metrics 
-                    if m.provider == stat.provider
-                ) / stat.total_requests
-                stat.average_tokens = stat.total_tokens / stat.total_requests
-        
-        return dict(stats)
+    def registrar_falha(self, provider: str) -> None:
+        """Registra processamento falhado"""
+        metrica = self.metricas[provider]
+        metrica["tentativas"] += 1
+        metrica["falhas"] += 1
+        logger.warning(f"Falha registrada para {provider}")
     
-    def get_model_stats(self) -> Dict[str, Dict[str, any]]:
-        """Retorna estatísticas por modelo."""
-        model_stats = defaultdict(lambda: {
-            'requests': 0,
-            'success': 0,
-            'cost': 0.0,
-            'tokens': 0,
-            'latency': 0.0,
-        })
+    def obter_estatisticas(self) -> Dict[str, Dict[str, Any]]:
+        """Retorna estatísticas consolidadas por provedor"""
+        stats = {}
         
-        for metric in self.metrics:
-            key = f"{metric.provider}/{metric.model}"
-            stats = model_stats[key]
+        for provider, metrica in self.metricas.items():
+            tentativas = metrica["tentativas"]
+            sucessos = metrica["sucessos"]
             
-            stats['requests'] += 1
-            if metric.success:
-                stats['success'] += 1
-            stats['cost'] += metric.cost_usd
-            stats['tokens'] += metric.tokens_input + metric.tokens_output
-            stats['latency'] += metric.latency_ms
-        
-        # Calcular médias
-        for stats in model_stats.values():
-            if stats['requests'] > 0:
-                stats['success_rate'] = stats['success'] / stats['requests']
-                stats['avg_cost'] = stats['cost'] / stats['requests']
-                stats['avg_tokens'] = stats['tokens'] / stats['requests']
-                stats['avg_latency'] = stats['latency'] / stats['requests']
-        
-        return dict(model_stats)
-    
-    def get_time_series(self, 
-                       minutes: int = 60,
-                       interval_seconds: int = 60) -> Dict[str, List[Tuple[datetime, float]]]:
-        """
-        Retorna série temporal de custos.
-        
-        Args:
-            minutes: Janela de tempo em minutos
-            interval_seconds: Intervalo de agregação
-        
-        Returns:
-            Dict com séries temporais por provedor
-        """
-        cutoff = datetime.now() - timedelta(minutes=minutes)
-        recent_metrics = [m for m in self.metrics if m.timestamp >= cutoff]
-        
-        time_buckets = defaultdict(lambda: defaultdict(float))
-        
-        for metric in recent_metrics:
-            bucket_time = (metric.timestamp.replace(second=0, microsecond=0) -
-                          timedelta(seconds=metric.timestamp.second % interval_seconds))
-            time_buckets[metric.provider][bucket_time] += metric.cost_usd
-        
-        result = {}
-        for provider, buckets in time_buckets.items():
-            result[provider] = sorted(buckets.items())
-        
-        return result
-    
-    def get_summary(self) -> Dict[str, any]:
-        """Retorna resumo geral de métricas."""
-        if not self.metrics:
-            return {
-                'total_requests': 0,
-                'total_cost': 0.0,
-                'total_tokens': 0,
+            stats[provider] = {
+                "tentativas": tentativas,
+                "sucessos": sucessos,
+                "falhas": metrica["falhas"],
+                "sucesso_rate": (sucessos / tentativas * 100) if tentativas > 0 else 0,
+                "tempo_medio_segundos": (
+                    metrica["tempo_total_segundos"] / sucessos
+                    if sucessos > 0 else 0
+                ),
+                "tokens_total": metrica["tokens_total"],
+                "custo_total_usd": round(metrica["custo_total_usd"], 6),
             }
         
-        successful = sum(1 for m in self.metrics if m.success)
-        
-        return {
-            'total_requests': len(self.metrics),
-            'successful_requests': successful,
-            'failed_requests': len(self.metrics) - successful,
-            'success_rate': successful / len(self.metrics),
-            'total_cost': sum(m.cost_usd for m in self.metrics),
-            'total_tokens': sum(m.tokens_input + m.tokens_output for m in self.metrics),
-            'average_cost': sum(m.cost_usd for m in self.metrics) / len(self.metrics),
-            'average_latency_ms': sum(m.latency_ms for m in self.metrics) / len(self.metrics),
-            'min_latency_ms': min(m.latency_ms for m in self.metrics),
-            'max_latency_ms': max(m.latency_ms for m in self.metrics),
-            'providers': list(set(m.provider for m in self.metrics)),
-            'models': list(set(m.model for m in self.metrics)),
-        }
+        return stats
     
-    def get_user_stats(self, user_id: str) -> Optional[Dict[str, any]]:
-        """Retorna estatísticas para um usuário específico."""
-        user_metrics = [m for m in self.metrics if m.user_id == user_id]
-        
-        if not user_metrics:
-            return None
-        
-        return {
-            'user_id': user_id,
-            'total_requests': len(user_metrics),
-            'total_cost': sum(m.cost_usd for m in user_metrics),
-            'total_tokens': sum(m.tokens_input + m.tokens_output for m in user_metrics),
-            'providers_used': list(set(m.provider for m in user_metrics)),
-            'models_used': list(set(m.model for m in user_metrics)),
-        }
-    
-    def get_cost_by_provider(self) -> Dict[str, float]:
-        """Retorna custo total por provedor."""
-        costs = defaultdict(float)
-        for metric in self.metrics:
-            costs[metric.provider] += metric.cost_usd
-        return dict(costs)
+    def relatorio_consolidado(self) -> str:
+        """Gera relatório consolidado em texto"""
+        stats = self.obter_estatisticas()
+        
+        relatorio = "\n" + "=" * 80 + "\n"
+        relatorio += "RELATÓRIO DE MÉTRICAS LLM\n"
+        relatorio += f"Data: {datetime.now().isoformat()}\n"
+        relatorio += "=" * 80 + "\n\n"
+        
+        for provider, metrica in stats.items():
+            relatorio += f"PROVEDOR: {provider.upper()}\n"
+            relatorio += f"  Tentativas: {metrica['tentativas']}\n"
+            relatorio += f"  Sucessos: {metrica['sucessos']}\n"
+            relatorio += f"  Falhas: {metrica['falhas']}\n"
+            relatorio += f"  Taxa de sucesso: {metrica['sucesso_rate']:.1f}%\n"
+            relatorio += f"  Tempo médio: {metrica['tempo_medio_segundos']:.2f}s\n"
+            relatorio += f"  Tokens totais: {metrica['tokens_total']}\n"
+            relatorio += f"  Custo total: ${metrica['custo_total_usd']:.6f}\n"
+            relatorio += "\n"
+        
+        relatorio += "=" * 80 + "\n"
+        return relatorio
     
-    def export_metrics(self, filepath: str):
-        """Exporta métricas para JSON."""
-        try:
-            data = {
-                'export_timestamp': datetime.now().isoformat(),
-                'metrics': [asdict(m) for m in self.metrics],
-                'summary': self.get_summary(),
-                'provider_stats': {
-                    k: asdict(v) for k, v in self.get_provider_stats().items()
-                },
-                'model_stats': self.get_model_stats(),
-            }
-            
-            with open(filepath, 'w') as f:
-                json.dump(data, f, indent=2, default=str)
-            
-            logger.info(f"Métricas exportadas para {filepath}")
-        
-        except Exception as e:
-            logger.error(f"Erro ao exportar métricas: {e}")
-            raise
-    
-    def clear(self):
-        """Limpa todas as métricas."""
-        self.metrics.clear()
-        logger.info("Métricas limpas")
-
-
-# Instância global
-_metrics_collector: Optional[MetricsCollector] = None
-
-
-def get_metrics_collector() -> MetricsCollector:
-    """Retorna a instância global do coletor de métricas."""
-    global _metrics_collector
-    if _metrics_collector is None:
-        _metrics_collector = MetricsCollector()
-    return _metrics_collector
+    def resetar_metricas(self) -> None:
+        """Reseta todas as métricas"""
+        self.metricas.clear()
+        logger.info("Métricas foram resetadas")