┌─────────────────────────────────────────────┐
│              应用层（Applications）          │
│  故障自愈 │ 智能告警 │ 容量规划 │ 知识问答   │
├─────────────────────────────────────────────┤
│              能力层（Capabilities）          │
│  异常检测 │ 根因分析 │ 预测分析 │ 决策引擎   │
├─────────────────────────────────────────────┤
│              数据层（Data Layer）            │
│  指标数据 │ 日志数据 │ 链路追踪 │ 事件数据   │
├─────────────────────────────────────────────┤
│              基础设施层（Infrastructure）     │
│  Kubernetes │ 微服务 │ 数据库 │ 中间件       │
└─────────────────────────────────────────────┘

Prometheus → 数据采集 → 特征工程 → 模型预测 → 异常判定 → 告警输出
     ↑                                                         │
     └────────────── 反馈优化 ←────────────────────────────────┘

import pandas as pd
import numpy as np
from prometheus_api_client import PrometheusConnect
from sklearn.ensemble import IsolationForest
from sklearn.preprocessing import StandardScaler
import joblib
import time

class AnomalyDetector:
    """基于孤立森林的多维异常检测器"""
    
    def __init__(self, prometheus_url, model_path=None):
        self.prom = PrometheusConnect(url=prometheus_url)
        self.scaler = StandardScaler()
        self.model = IsolationForest(
            n_estimators=100,
            contamination=0.01,  # 异常比例1%
            random_state=42
        )
        self.is_trained = False
        
        if model_path and os.path.exists(model_path):
            self.load_model(model_path)
    
    def collect_metrics(self, start_time, end_time, step='1m'):
        """采集关键指标数据"""
        queries = {
            'cpu_usage': '100 - (avg by (instance) (rate(node_cpu_seconds_total{mode="idle"}[5m])) * 100)',
            'memory_usage': '(1 - (node_memory_MemAvailable_bytes / node_memory_MemTotal_bytes)) * 100',
            'disk_io': 'rate(node_disk_read_bytes_total[5m]) + rate(node_disk_write_bytes_total[5m])',
            'network_io': 'rate(node_network_receive_bytes_total[5m]) + rate(node_network_transmit_bytes_total[5m])',
        }
        
        data = {}
        for name, query in queries.items():
            result = self.prom.custom_query_range(
                query=query,
                start_time=start_time,
                end_time=end_time,
                step=step
            )
            
            # 处理多实例数据
            for series in result:
                instance = series['metric'].get('instance', 'unknown')
                key = f"{instance}_{name}"
                values = [float(v[1]) for v in series['values']]
                data[key] = values
        
        return pd.DataFrame(data)
    
    def extract_features(self, df):
        """提取特征"""
        features = pd.DataFrame()
        
        for col in df.columns:
            # 基础统计特征
            features[f'{col}_mean'] = df[col].rolling(window=10).mean()
            features[f'{col}_std'] = df[col].rolling(window=10).std()
            features[f'{col}_rate'] = df[col].diff()
            
            # 分位数特征
            features[f'{col}_p95'] = df[col].rolling(window=30).quantile(0.95)
            features[f'{col}_p99'] = df[col].rolling(window=30).quantile(0.99)
        
        return features.dropna()
    
    def train(self, start_time, end_time):
        """训练模型"""
        print("正在采集训练数据...")
        df = self.collect_metrics(start_time, end_time)
        print(f"采集到 {len(df)} 条数据")
        
        print("正在提取特征...")
        features = self.extract_features(df)
        print(f"提取到 {features.shape[1]} 个特征")
        
        print("正在标准化数据...")
        scaled_features = self.scaler.fit_transform(features)
        
        print("正在训练模型...")
        self.model.fit(scaled_features)
        self.is_trained = True
        print("模型训练完成")
    
    def detect(self, lookback='30m'):
        """实时异常检测"""
        if not self.is_trained:
            raise Exception("模型未训练，请先调用train()方法")
        
        end_time = pd.Timestamp.now()
        start_time = end_time - pd.Timedelta(lookback)
        
        df = self.collect_metrics(start_time, end_time, step='1m')
        features = self.extract_features(df)
        
        if len(features) == 0:
            return []
        
        scaled_features = self.scaler.transform(features)
        predictions = self.model.predict(scaled_features)
        scores = -self.model.score_samples(scaled_features)
        
        anomalies = []
        for i, (pred, score) in enumerate(zip(predictions, scores)):
            if pred == -1:  # 异常
                timestamp = features.index[i]
                anomalies.append({
                    'timestamp': timestamp,
                    'anomaly_score': score,
                    'metrics': df.iloc[i].to_dict(),
                })
        
        return anomalies
    
    def save_model(self, path):
        """保存模型"""
        joblib.dump({
            'model': self.model,
            'scaler': self.scaler,
        }, path)
        print(f"模型已保存到 {path}")
    
    def load_model(self, path):
        """加载模型"""
        data = joblib.load(path)
        self.model = data['model']
        self.scaler = data['scaler']
        self.is_trained = True
        print(f"模型已从 {path} 加载")


# 使用示例
if __name__ == "__main__":
    detector = AnomalyDetector("http://prometheus:9090")
    
    # 训练（使用过去7天的数据）
    end_time = pd.Timestamp.now()
    start_time = end_time - pd.Timedelta(days=7)
    detector.train(start_time, end_time)
    
    # 保存模型
    detector.save_model("anomaly_model.pkl")
    
    # 实时检测
    while True:
        anomalies = detector.detect(lookback='30m')
        if anomalies:
            print(f"检测到 {len(anomalies)} 个异常点！")
            for anomaly in anomalies[:3]:  # 只打印前3个
                print(f"  时间: {anomaly['timestamp']}")
                print(f"  异常分数: {anomaly['anomaly_score']:.4f}")
        else:
            print("当前无异常")
        
        time.sleep(60)  # 每分钟检测一次

服务A异常
  ├── 依赖服务B异常 → 依赖服务D异常（根因）
  └── 依赖服务C异常 → 依赖服务D异常（同上）

import json
from datetime import datetime
from typing import List, Dict
import anthropic

class RootCauseAnalyzer:
    """基于大模型的智能根因分析器"""
    
    def __init__(self, api_key: str):
        self.client = anthropic.Client(api_key=api_key)
        self.model = "claude-3.7-sonnet"
    
    def collect_context(self, alert: Dict) -> Dict:
        """收集故障上下文信息"""
        context = {
            "alert": alert,
            "timestamp": datetime.now().isoformat(),
            "related_metrics": self._get_related_metrics(alert),
            "related_logs": self._get_related_logs(alert),
            "related_traces": self._get_related_traces(alert),
            "recent_changes": self._get_recent_changes(alert),
            "service_dependencies": self._get_service_dependencies(alert),
            "historical_incidents": self._get_historical_incidents(alert),
        }
        return context
    
    def _get_related_metrics(self, alert: Dict) -> List[Dict]:
        """获取相关指标数据"""
        # 实现：从Prometheus/VictoriaMetrics查询相关指标
        # 返回故障前后的关键指标曲线数据
        return []
    
    def _get_related_logs(self, alert: Dict) -> List[Dict]:
        """获取相关日志"""
        # 实现：从Loki/ES查询相关错误日志
        return []
    
    def _get_related_traces(self, alert: Dict) -> List[Dict]:
        """获取相关链路追踪"""
        # 实现：从Jaeger查询慢请求和错误请求的trace
        return []
    
    def _get_recent_changes(self, alert: Dict) -> List[Dict]:
        """获取最近的变更记录"""
        # 实现：从发布系统、配置中心查询最近变更
        return []
    
    def _get_service_dependencies(self, alert: Dict) -> Dict:
        """获取服务依赖关系"""
        # 实现：从服务治理平台获取依赖拓扑
        return {}
    
    def _get_historical_incidents(self, alert: Dict) -> List[Dict]:
        """获取历史类似故障"""
        # 实现：从事务管理系统查询历史故障记录
        return []
    
    def analyze(self, alert: Dict) -> Dict:
        """执行根因分析"""
        print("正在收集故障上下文...")
        context = self.collect_context(alert)
        
        prompt = self._build_prompt(context)
        
        print("正在调用AI进行根因分析...")
        response = self.client.messages.create(
            model=self.model,
            max_tokens=4000,
            system=self._get_system_prompt(),
            messages=[{"role": "user", "content": prompt}]
        )
        
        result = self._parse_response(response.content[0].text)
        result["raw_analysis"] = response.content[0].text
        
        return result
    
    def _get_system_prompt(self) -> str:
        return """你是一位经验丰富的SRE专家，擅长复杂系统的故障根因分析。
请根据提供的故障上下文信息，进行系统性的分析，找出最可能的根因。

分析要求：
1. 先梳理故障现象和影响范围
2. 列出可能的原因假设
3. 根据数据逐一验证假设
4. 给出最可能的根因结论
5. 提供修复建议和预防措施
6. 对分析过程中的不确定之处要明确标注

输出格式：JSON，包含以下字段
- root_cause: 根因描述
- confidence: 置信度（0-1）
- evidence: 支持结论的证据列表
- alternative_causes: 其他可能原因列表
- remediation_steps: 修复建议步骤
- prevention_measures: 预防措施建议
"""
    
    def _build_prompt(self, context: Dict) -> str:
        return f"""请分析以下故障的根因：

## 告警信息
{json.dumps(context['alert'], ensure_ascii=False, indent=2)}

## 相关指标数据
{json.dumps(context['related_metrics'][:10], ensure_ascii=False, indent=2)}

## 相关错误日志
{json.dumps(context['related_logs'][:20], ensure_ascii=False, indent=2)}

## 相关链路追踪
{json.dumps(context['related_traces'][:5], ensure_ascii=False, indent=2)}

## 最近变更记录
{json.dumps(context['recent_changes'], ensure_ascii=False, indent=2)}

## 服务依赖关系
{json.dumps(context['service_dependencies'], ensure_ascii=False, indent=2)}

## 历史类似故障
{json.dumps(context['historical_incidents'][:3], ensure_ascii=False, indent=2)}

请进行详细的根因分析。"""
    
    def _parse_response(self, response_text: str) -> Dict:
        """解析AI响应"""
        # 尝试从响应中提取JSON
        try:
            # 查找JSON代码块
            import re
            json_match = re.search(r'

', response_text, re.DOTALL)
            if json_match:
                return json.loads(json_match.group(1))
            
            # 直接尝试解析
            return json.loads(response_text)
        except:
            # 解析失败，返回原始文本
            return {
                "root_cause": "解析失败，请查看原始分析",
                "confidence": 0,
                "evidence": [],
                "alternative_causes": [],
                "remediation_steps": [],
                "prevention_measures": [],
            }


# 使用示例
if __name__ == "__main__":
    analyzer = RootCauseAnalyzer("your-api-key")
    
    alert = {
        "alertname": "HighErrorRate",
        "service": "order-service",
        "severity": "critical",
        "description": "订单服务错误率超过5%",
        "started_at": "2026-08-25T10:30:00Z",
    }
    
    result = analyzer.analyze(alert)
    
    print(f"
根因分析结果：")
    print(f"根因：{result['root_cause']}")
    print(f"置信度：{result['confidence']}")
    print(f"修复建议：")
    for i, step in enumerate(result.get('remediation_steps', []), 1):
        print(f"  {i}. {step}")