AI系统可观测性搭建

from prometheus_client import Counter, Histogram, Gauge

# 请求指标
REQUEST_TOTAL = Counter('ai_requests_total', 'Total AI requests', ['model', 'status'])
REQUEST_LATENCY = Histogram('ai_request_duration_seconds', 'Request duration', ['model'])
ACTIVE_REQUESTS = Gauge('ai_active_requests', 'Active requests')

# Token指标
TOKEN_INPUT = Counter('ai_tokens_input_total', 'Input tokens', ['model'])
TOKEN_OUTPUT = Counter('ai_tokens_output_total', 'Output tokens', ['model'])
TOKEN_COST = Counter('ai_token_cost_usd', 'Token cost in USD', ['model'])

# 质量指标
HALLUCINATION_RATE = Gauge('ai_hallucination_rate', 'Hallucination rate', ['model'])
TOOL_CALL_SUCCESS = Counter('ai_tool_calls_total', 'Tool calls', ['tool', 'status'])

# 缓存指标
CACHE_HIT_RATE = Gauge('ai_cache_hit_rate', 'Cache hit rate')

class ObservabilityMiddleware:
    def __init__(self, app):
        self.app = app
    
    async def __call__(self, request):
        ACTIVE_REQUESTS.inc()
        start = time.time()
        model = request.json.get("model", "unknown")
        
        try:
            response = await self.app(request)
            duration = time.time() - start
            
            REQUEST_TOTAL.labels(model=model, status="success").inc()
            REQUEST_LATENCY.labels(model=model).observe(duration)
            
            if "usage" in response:
                TOKEN_INPUT.labels(model=model).inc(response["usage"]["prompt_tokens"])
                TOKEN_OUTPUT.labels(model=model).inc(response["usage"]["completion_tokens"])
                cost = self.calculate_cost(model, response["usage"])
                TOKEN_COST.labels(model=model).inc(cost)
            
            return response
        except Exception as e:
            REQUEST_TOTAL.labels(model=model, status="error").inc()
            raise
        finally:
            ACTIVE_REQUESTS.dec()
    
    def calculate_cost(self, model, usage):
        pricing = {"gpt-4": 0.03, "qwen3-32b": 0.002, "claude-3": 0.015}
        rate = pricing.get(model, 0.01)
        return (usage["prompt_tokens"] + usage["completion_tokens"]) / 1000 * rate

from opentelemetry import trace

tracer = trace.get_tracer(__name__)

class TracedLLMCall:
    def __init__(self, llm_client):
        self.client = llm_client
    
    async def chat(self, messages, **kwargs):
        with tracer.start_as_current_span("llm_chat") as span:
            span.set_attribute("llm.model", kwargs.get("model", "unknown"))
            span.set_attribute("llm.messages_count", len(messages))
            span.set_attribute("llm.temperature", kwargs.get("temperature", 0.7))
            
            start = time.time()
            response = await self.client.chat(messages, **kwargs)
            duration = time.time() - start
            
            span.set_attribute("llm.duration_ms", duration * 1000)
            span.set_attribute("llm.prompt_tokens", response["usage"]["prompt_tokens"])
            span.set_attribute("llm.completion_tokens", response["usage"]["completion_tokens"])
            
            return response

class QualityMonitor:
    def __init__(self, sample_rate=0.05):
        self.sample_rate = sample_rate  # 采样5%的请求做质量评估
    
    async def evaluate_response(self, query, response, context=None):
        """异步评估响应质量"""
        import random
        if random.random() > self.sample_rate:
            return None
        
        metrics = {}
        
        # 幻觉检测
        metrics["hallucination"] = await self.detect_hallucination(response, context)
        
        # 相关性
        metrics["relevance"] = await self.score_relevance(query, response)
        
        # 毒性检测
        metrics["toxicity"] = await self.detect_toxicity(response)
        
        # 记录到Prometheus
        if metrics["hallucination"]:
            HALLUCINATION_RATE.inc()
        else:
            HALLUCINATION_RATE.dec(0.01)
        
        return metrics

# Prometheus告警规则
groups:
  - name: ai_alerts
    rules:
      - alert: HighErrorRate
        expr: rate(ai_requests_total{status="error"}[5m]) / rate(ai_requests_total[5m]) > 0.05
        for: 5m
        annotations:
          summary: "AI error rate > 5%"
      
      - alert: HighLatency
        expr: histogram_quantile(0.95, ai_request_duration_seconds_bucket) > 30
        for: 10m
        annotations:
          summary: "P95 latency > 30s"
      
      - alert: HighCost
        expr: rate(ai_token_cost_usd[1h]) > 100
        for: 30m
        annotations:
          summary: "Hourly cost > $100"
      
      - alert: ModelDegradation
        expr: ai_hallucination_rate > 0.15
        for: 1h
        annotations:
          summary: "Hallucination rate > 15%"

import structlog

logger = structlog.get_logger()

def log_llm_call(model, messages, response, duration, user_id):
    logger.info("llm_call",
        model=model,
        user_id=user_id,
        input_tokens=response["usage"]["prompt_tokens"],
        output_tokens=response["usage"]["completion_tokens"],
        duration_ms=duration * 1000,
        message_count=len(messages),
        # 脱敏记录（不记录完整内容）
        input_preview=messages[-1]["content"][:100] + "..."
    )