AI 可观测性与可靠性工程

from dataclasses import dataclass, field
from typing import Any, Optional
from datetime import datetime
import json
import hashlib

@dataclass
class AIObservation:
    """AI 系统的单次观测记录"""
    timestamp: datetime
    observation_type: str  # "performance" | "quality" | "data" | "cost"
    component: str  # 组件名称，如 "llm-inference" | "rag-retrieval" | "agent-loop"
    
    # 通用字段
    request_id: str = ""
    user_id: str = ""
    
    # 性能指标
    latency_ms: float = 0.0
    tokens_input: int = 0
    tokens_output: int = 0
    token_cost: float = 0.0
    
    # 质量指标
    quality_score: Optional[float] = None
    hallucination_detected: bool = False
    safety_score: Optional[float] = None
    
    # 数据指标
    context_quality: Optional[float] = None
    retrieval_recall: Optional[float] = None
    input_drift_score: Optional[float] = None
    
    # 元数据
    model_name: str = ""
    prompt_hash: str = ""
    metadata: dict = field(default_factory=dict)
    
    def compute_prompt_hash(self, prompt: str) -> str:
        """对 prompt 做哈希，用于追踪 prompt 变更的影响"""
        return hashlib.sha256(prompt.encode()).hexdigest()[:16]
    
    def to_json(self) -> str:
        return json.dumps({
            "timestamp": self.timestamp.isoformat(),
            "type": self.observation_type,
            "component": self.component,
            "request_id": self.request_id,
            "latency_ms": self.latency_ms,
            "tokens": {"input": self.tokens_input, "output": self.tokens_output},
            "quality": self.quality_score,
            "cost": self.token_cost,
            "model": self.model_name
        })

# 示例：记录一次 LLM 推理的完整观测
observation = AIObservation(
    timestamp=datetime.utcnow(),
    observation_type="performance",
    component="llm-inference",
    request_id="req_001",
    latency_ms=234.5,
    tokens_input=1280,
    tokens_output=567,
    token_cost=0.0032,
    quality_score=0.87,
    model_name="qwen3.6-27b",
    metadata={"temperature": 0.7, "max_tokens": 1024}
)

from typing import List, Dict, Tuple
import numpy as np
from enum import Enum

class QualityDimension(Enum):
    ACCURACY = "accuracy"        # 答案是否正确
    COMPLETENESS = "completeness"  # 回答是否完整
    CONSISTENCY = "consistency"    # 多次运行是否一致
    SAFETY = "safety"              # 是否安全无害
    RELEVANCE = "relevance"        # 是否与问题相关

@dataclass
class QualityEvaluation:
    dimensions: Dict[QualityDimension, float]
    overall_score: float
    flags: List[str] = field(default_factory=list)

def evaluate_llm_response(
    question: str,
    response: str,
    reference_answer: str = None,
    n_runs: int = 3
) -> QualityEvaluation:
    """多维度评估 LLM 输出质量"""
    scores = {}
    flags = []
    
    # 1. 一致性检测：多次运行同一 prompt，检查输出变化
    if n_runs > 1:
        variations = []
        for _ in range(n_runs):
            variations.append(call_llm(question))
        
        # 计算输出间的语义相似度
        similarities = compute_pairwise_similarities(variations)
        scores[QualityDimension.CONSISTENCY] = np.mean(similarities)
        
        if scores[QualityDimension.CONSISTENCY] < 0.7:
            flags.append("输出一致性偏低，建议降低 temperature")
    
    # 2. 准确性评估（如果有参考答案）
    if reference_answer:
        scores[QualityDimension.ACCURACY] = compute_similarity(
            response, reference_answer
        )
    
    # 3. 安全性检测
    safety_score = check_safety(response)
    scores[QualityDimension.SAFETY] = safety_score
    
    if safety_score < 0.5:
        flags.append("⚠️ 检测到潜在安全风险")
    
    # 4. 相关性检测
    relevance = compute_semantic_similarity(question, response)
    scores[QualityDimension.RELEVANCE] = relevance
    
    # 计算综合得分（加权平均）
    weights = {
        QualityDimension.ACCURACY: 0.35,
        QualityDimension.SAFETY: 0.30,
        QualityDimension.CONSISTENCY: 0.20,
        QualityDimension.RELEVANCE: 0.15,
    }
    overall = sum(
        weights.get(dim, 0.0) * score 
        for dim, score in scores.items()
    ) / sum(weights.get(dim, 0.0) for dim in scores)
    
    return QualityEvaluation(
        dimensions=scores,
        overall_score=overall,
        flags=flags
    )

def check_safety(text: str) -> float:
    """简化的安全性检测（实际应使用专用模型）"""
    risk_patterns = [
        # 添加你的风险检测模式
    ]
    risk_score = sum(1 for p in risk_patterns if p in text.lower())
    return max(0.0, 1.0 - risk_score * 0.1)

import asyncio
from typing import List, Optional
from datetime import datetime
from enum import Enum
from dataclasses import dataclass

class FaultSeverity(Enum):
    LOW = "low"
    MEDIUM = "medium"
    HIGH = "high"
    CRITICAL = "critical"

@dataclass
class FaultReport:
    """故障报告"""
    fault_id: str
    timestamp: datetime
    severity: FaultSeverity
    component: str
    description: str
    detected_by: str  # "metric" | "anomaly" | "manual"
    metrics_snapshot: dict = None
    
    def estimate_impact(self) -> float:
        """估算故障影响范围（0-1）"""
        severity_weight = {
            FaultSeverity.LOW: 0.1,
            FaultSeverity.MEDIUM: 0.3,
            FaultSeverity.HIGH: 0.7,
            FaultSeverity.CRITICAL: 1.0,
        }
        return severity_weight[self.severity]

class SelfHealingAgent:
    """AI 故障自愈 Agent"""
    
    def __init__(self, llm_client, action_executor):
        self.llm = llm_client
        self.executor = action_executor
        self.knowledge_base = []  # 历史故障经验
        self.active_remediations = set()  # 正在执行的修复
    
    async def detect_and_remediate(self, observations: List) -> List:
        """检测故障并自动修复"""
        faults = await self._detect_faults(observations)
        remediations = []
        
        for fault in faults:
            # 1. 根因分析
            rca = await self._root_cause_analysis(fault)
            
            # 2. 查找历史经验
            past_solution = self._find_similar_fault(fault)
            
            # 3. 制定修复方案
            plan = await self._create_remediation_plan(
                fault, rca, past_solution
            )
            
            # 4. 执行修复
            if plan.confidence > 0.8:
                result = await self._execute_plan(plan)
                remediations.append(result)
            else:
                # 低置信度，通知人工
                await self._notify_human(fault, plan)
        
        # 5. 记录经验
        self._record_experience(faults, remediations)
        return remediations
    
    async def _detect_faults(
        self, observations: List
    ) -> List[FaultReport]:
        """从观测数据中检测故障"""
        faults = []
        
        for obs in observations:
            # 规则检测
            if obs.latency_ms > 5000:
                faults.append(FaultReport(
                    fault_id=f"fault_{obs.request_id}",
                    timestamp=obs.timestamp,
                    severity=FaultSeverity.HIGH,
                    component=obs.component,
                    description=f"推理延迟过高: {obs.latency_ms}ms",
                    detected_by="metric"
                ))
            
            # 质量检测
            if obs.quality_score is not None and obs.quality_score < 0.5:
                faults.append(FaultReport(
                    fault_id=f"fault_q_{obs.request_id}",
                    timestamp=obs.timestamp,
                    severity=FaultSeverity.MEDIUM,
                    component=obs.component,
                    description=f"输出质量下降: 质量得分 {obs.quality_score}",
                    detected_by="anomaly",
                    metrics_snapshot={"quality_score": obs.quality_score}
                ))
        
        return faults
    
    async def _root_cause_analysis(
        self, fault: FaultReport
    ) -> dict:
        """使用 LLM 进行智能根因分析"""
        prompt = f"""
你是一位 AI SRE 专家。请分析以下故障报告，
判断最可能的根因并给出置信度。

故障: {fault.description}
组件: {fault.component}
严重程度: {fault.severity.value}
指标快照: {fault.metrics_snapshot}

请按以下格式输出:
- 根因: [一句话描述]
- 置信度: [0-1]
- 建议修复: [具体操作]
"""
        analysis = await self.llm.generate(prompt)
        return self._parse_analysis(analysis)

    async def _create_remediation_plan(
        self, fault: FaultReport, rca: dict, past: Optional[dict]
    ) -> dict:
        """制定修复方案"""
        plan = {
            "fault_id": fault.fault_id,
            "actions": [],
            "confidence": rca.get("confidence", 0.5),
            "estimated_recovery_time": "unknown",
        }
        
        # 如果有历史相似故障，优先使用历史方案
        if past:
            plan["actions"] = past.get("successful_actions", [])
            plan["confidence"] = max(plan["confidence"], 0.7)
        
        # 基于根因分析添加操作
        root_cause = rca.get("root_cause", "")
        if "memory" in root_cause.lower():
            plan["actions"].append({
                "type": "restart",
                "target": fault.component,
                "reason": "内存泄漏嫌疑"
            })
        elif "rate_limit" in root_cause.lower():
            plan["actions"].append({
                "type": "backoff",
                "target": "api_client",
                "parameters": {"delay_ms": 1000, "max_retries": 5}
            })
        
        return plan
    
    async def _execute_plan(self, plan: dict) -> dict:
        """执行修复计划"""
        results = []
        for action in plan["actions"]:
            result = await self.executor.execute(action)
            results.append(result)
        
        return {
            "plan_id": plan["fault_id"],
            "actions_executed": len(results),
            "success_rate": sum(1 for r in results if r.success) / len(results)
        }

from dataclasses import dataclass, field
from typing import List, Callable, Dict
from enum import Enum
import time

class NotificationChannel(Enum):
    SLACK = "slack"
    EMAIL = "email"
    SMS = "sms"
    PHONE = "phone"
    WEBHOOK = "webhook"

@dataclass
class AlertRule:
    """告警规则定义"""
    name: str
    condition: Callable  # 返回 bool 的函数
    severity: FaultSeverity
    channels: List[NotificationChannel]
    cooldown_seconds: int = 300  # 冷却时间
    auto_escalate: bool = False
    escalation_after_seconds: int = 900
    
    _last_fired: float = 0
    
    def should_fire(self) -> bool:
        """判断是否应该触发告警"""
        if not self.condition():
            return False
        if time.time() - self._last_fired < self.cooldown_seconds:
            return False
        return True
    
    def fire(self) -> None:
        self._last_fired = time.time()

class AlertRouter:
    """智能告警路由"""
    
    def __init__(self):
        self.rules: List[AlertRule] = []
        self.notification_handlers: Dict[NotificationChannel, Callable] = {}
        self.alert_history: List = []
        self.suppression_windows: Dict[str, float] = {}
    
    def add_rule(self, rule: AlertRule) -> None:
        self.rules.append(rule)
    
    def register_handler(
        self, channel: NotificationChannel, handler: Callable
    ) -> None:
        self.notification_handlers[channel] = handler
    
    def evaluate_and_alert(self, context: dict) -> List:
        """评估所有规则并发送告警"""
        fired_alerts = []
        
        for rule in self.rules:
            if rule.should_fire():
                # 检查抑制窗口（避免告警风暴）
                if self._is_suppressed(rule.name):
                    continue
                
                rule.fire()
                
                # 确定通知渠道（根据严重级别）
                channels = self._determine_channels(rule)
                
                for channel in channels:
                    handler = self.notification_handlers.get(channel)
                    if handler:
                        handler({
                            "rule": rule.name,
                            "severity": rule.severity.value,
                            "context": context,
                            "timestamp": time.time()
                        })
                
                fired_alerts.append(rule.name)
                self.alert_history.append({
                    "rule": rule.name,
                    "timestamp": time.time(),
                    "context": context
                })
        
        return fired_alerts
    
    def _determine_channels(self, rule: AlertRule) -> List:
        """根据严重级别确定通知渠道"""
        severity_channels = {
            FaultSeverity.LOW: [NotificationChannel.SLACK],
            FaultSeverity.MEDIUM: [
                NotificationChannel.SLACK,
                NotificationChannel.EMAIL
            ],
            FaultSeverity.HIGH: [
                NotificationChannel.SLACK,
                NotificationChannel.EMAIL,
                NotificationChannel.SMS
            ],
            FaultSeverity.CRITICAL: [
                NotificationChannel.SLACK,
                NotificationChannel.EMAIL,
                NotificationChannel.SMS,
                NotificationChannel.PHONE
            ],
        }
        return list(set(rule.channels) | set(
            severity_channels.get(rule.severity, [])
        ))
    
    def _is_suppressed(self, rule_name: str) -> bool:
        """检查是否在抑制窗口内（避免告警风暴）"""
        last_alert = self._get_last_alert_time(rule_name)
        if last_alert and time.time() - last_alert < 60:
            return True
        return False
    
    def _get_last_alert_time(self, rule_name: str) -> Optional[float]:
        recent = [
            a for a in self.alert_history 
            if a["rule"] == rule_name
        ]
        return recent[-1]["timestamp"] if recent else None

from opentelemetry import trace, metrics
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.metrics import MeterProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import (
    OTLPSpanExporter
)
from opentelemetry.semconv.trace import SpanAttributes
import time
from contextlib import contextmanager

# 初始化 Tracer
trace.set_tracer_provider(TracerProvider())
tracer = trace.get_tracer("ai-master.llm")

# 初始化 Metrics
meter = metrics.get_meter("ai-master.llm")

# 定义自定义指标
llm_latency = meter.create_histogram(
    "llm.latency",
    unit="ms",
    description="LLM 推理延迟"
)
token_counter = meter.create_counter(
    "llm.tokens",
    unit="tokens",
    description="Token 消耗计数"
)
quality_gauge = meter.create_up_down_counter(
    "llm.quality_score",
    unit="score",
    description="输出质量评分"
)

@contextmanager
def observe_llm_call(
    model_name: str,
    prompt: str,
    request_id: str = None
):
    """LLM 调用的观测上下文管理器"""
    request_id = request_id or generate_request_id()
    
    with tracer.start_as_current_span(
        "llm.generate",
        attributes={
            "llm.system": model_name,
            "llm.request_id": request_id,
            "llm.prompt_length": len(prompt),
            "gen_ai.operation.name": "chat",
        }
    ) as span:
        start_time = time.time()
        
        try:
            # 执行业务逻辑（由调用方传入）
            yield {"request_id": request_id, "span": span}
            
            # 记录成功
            latency_ms = (time.time() - start_time) * 1000
            llm_latency.record(latency_ms, {"model": model_name})
            span.set_attribute("llm.latency_ms", latency_ms)
            span.set_status(trace.StatusCode.OK)
            
        except Exception as e:
            # 记录错误
            latency_ms = (time.time() - start_time) * 1000
            llm_latency.record(latency_ms, {"model": model_name, "error": "true"})
            span.record_exception(e)
            span.set_status(trace.StatusCode.ERROR, str(e))
            raise

# 使用示例
def generate_with_observability(
    model: str,
    prompt: str,
    max_tokens: int = 1024
) -> str:
    with observe_llm_call(model, prompt) as ctx:
        request_id = ctx["request_id"]
        
        # 实际 LLM 调用
        response = call_llm_api(
            model=model,
            prompt=prompt,
            max_tokens=max_tokens
        )
        
        # 记录 Token 消耗
        token_counter.add(
            response.usage.total_tokens,
            {"model": model, "type": "total"}
        )
        token_counter.add(
            response.usage.prompt_tokens,
            {"model": model, "type": "input"}
        )
        token_counter.add(
            response.usage.completion_tokens,
            {"model": model, "type": "output"}
        )
        
        # 记录质量评分
        quality = evaluate_response(prompt, response.text)
        quality_gauge.add(
            quality.overall_score - 0.5,
            {"model": model, "request_id": request_id}
        )
        
        return response.text

from dataclasses import dataclass
from typing import List, Dict, Optional
from datetime import datetime, timedelta
import numpy as np

@dataclass
class SLO:
    """Service Level Objective 定义"""
    name: str
    metric_name: str
    target: float  # 目标值
    window: timedelta  # 评估窗口
    threshold: float  # 告警阈值（低于此值告警）
    
    def calculate_burn_rate(
        self, 
        observations: List,
        current_time: datetime
    ) -> float:
        """计算错误预算消耗率"""
        window_start = current_time - self.window
        window_obs = [
            o for o in observations 
            if o.timestamp >= window_start
        ]
        
        if not window_obs:
            return 0.0
        
        # 计算当前指标值
        values = [getattr(o, self.metric_name) for o in window_obs 
                  if hasattr(o, self.metric_name)]
        
        if not values:
            return 0.0
        
        current_value = np.mean(values)
        
        # 燃烧率 = (目标 - 当前值) / (目标 - 0)
        if self.target == 0:
            return 0.0
        
        burn_rate = (self.target - current_value) / self.target
        return max(0.0, burn_rate)
    
    def should_alert(self, burn_rate: float) -> bool:
        """判断是否应该触发告警"""
        return burn_rate > self.threshold

# 定义 AI 系统的 SLO
ai_slos = [
    SLO(
        name="推理延迟 P99",
        metric_name="latency_ms",
        target=2000.0,  # 目标 P99 < 2000ms
        window=timedelta(hours=1),
        threshold=0.3,  # 燃烧率 > 30% 告警
    ),
    SLO(
        name="输出质量",
        metric_name="quality_score",
        target=0.8,  # 目标质量 > 0.8
        window=timedelta(hours=4),
        threshold=0.2,
    ),
    SLO(
        name="服务可用性",
        metric_name="availability",
        target=0.999,  # 目标 99.9% 可用
        window=timedelta(days=30),
        threshold=0.1,
    ),
    SLO(
        name="Token 成本预算",
        metric_name="cost_per_hour",
        target=100.0,  # 每小时成本 < $100
        window=timedelta(hours=24),
        threshold=0.5,
    ),
]

def evaluate_slos(
    observations: List,
    current_time: datetime
) -> List[dict]:
    """评估所有 SLO 的状态"""
    results = []
    
    for slo in ai_slos:
        burn_rate = slo.calculate_burn_rate(observations, current_time)
        
        results.append({
            "slo": slo.name,
            "burn_rate": round(burn_rate, 4),
            "target": slo.target,
            "alerting": slo.should_alert(burn_rate),
            "error_budget_remaining": max(0, 1 - burn_rate),
        })
    
    return results

def generate_slo_report(results: List[dict]) -> str:
    """生成 SLO 报告"""
    report = ["📊 AI 系统 SLO 报告", "=" * 40]
    
    for r in results:
        status = "🔴" if r["alerting"] else "🟢"
        report.append(
            f"{status} {r['slo']}: "
            f"燃烧率 {r['burn_rate']:.1%}, "
            f"错误预算剩余 {r['error_budget_remaining']:.1%}"
        )
    
    alerting = [r for r in results if r["alerting"]]
    if alerting:
        report.append(f"\n⚠️  {len(alerting)} 个 SLO 正在告警！")
    
    return "\n".join(report)