57_大模型监控与运维：构建稳定可靠的服务体系

监控体系架构图
┌─────────────────────────────────────────────────────────────────┐
│                       统一监控平台层                            │
│  ┌─────────────┐  ┌─────────────┐  ┌──────────────────────┐     │
│  │ 可视化展示  │  │ 告警管理    │  │ 智能分析引擎         │     │
│  └─────────────┘  └─────────────┘  └──────────────────────┘     │
└───────────────────────┬─────────────────────────────────────────┘
                        │
┌───────────────────────┼─────────────────────────────────────────┐
│                       │                                         │
▼                       ▼                                         ▼
┌─────────────┐   ┌─────────────┐                       ┌─────────┐
│ 基础设施层  │   │ 模型服务层  │                       │ 应用层  │
│ - 硬件监控  │   │ - 推理引擎  │                       │ - API   │
│ - 资源监控  │   │ - 模型状态  │                       │ - 业务  │
│ - 网络监控  │   │ - 缓存系统  │                       │ - 用户  │
└─────────────┘   └─────────────┘                       └─────────┘

# 使用nvidia-smi和DCGM采集GPU指标的示例代码
import subprocess
import json
import time
from prometheus_client import Gauge, start_http_server

# 初始化Prometheus指标
gpu_utilization = Gauge('gpu_utilization_percent', 'GPU utilization percentage', ['gpu_id'])
gpu_memory_used = Gauge('gpu_memory_used_bytes', 'GPU memory used in bytes', ['gpu_id'])
gpu_memory_total = Gauge('gpu_memory_total_bytes', 'GPU memory total in bytes', ['gpu_id'])
gpu_temperature = Gauge('gpu_temperature_celsius', 'GPU temperature in Celsius', ['gpu_id'])
gpu_power_usage = Gauge('gpu_power_usage_watts', 'GPU power usage in watts', ['gpu_id'])

def collect_gpu_metrics():
    try:
        # 使用nvidia-smi获取GPU信息
        result = subprocess.run(
            ['nvidia-smi', '--query-gpu=index,utilization.gpu,memory.used,memory.total,temperature.gpu,power.draw',
             '--format=json'],
            capture_output=True,
            text=True
        )
        gpu_info = json.loads(result.stdout)
        
        for gpu in gpu_info['gpu']:
            gpu_id = gpu['index']
            gpu_utilization.labels(gpu_id=gpu_id).set(float(gpu['utilization.gpu'].replace('%', '')))
            gpu_memory_used.labels(gpu_id=gpu_id).set(int(gpu['memory.used'].replace(' MiB', '')) * 1024 * 1024)
            gpu_memory_total.labels(gpu_id=gpu_id).set(int(gpu['memory.total'].replace(' MiB', '')) * 1024 * 1024)
            gpu_temperature.labels(gpu_id=gpu_id).set(float(gpu['temperature.gpu']))
            gpu_power_usage.labels(gpu_id=gpu_id).set(float(gpu['power.draw'].replace(' W', '')))
            
        print(f"GPU metrics collected successfully")
    except Exception as e:
        print(f"Error collecting GPU metrics: {e}")

if __name__ == '__main__':
    # 启动Prometheus HTTP服务器
    start_http_server(9101)
    print("GPU metrics exporter started on port 9101")
    
    # 定期采集指标
    while True:
        collect_gpu_metrics()
        time.sleep(15)  # 每15秒采集一次

# 集成到推理服务的指标采集中间件
from prometheus_client import Counter, Histogram, Gauge, Summary
import time

# 请求计数器
REQUEST_COUNT = Counter('llm_requests_total', 'Total number of LLM requests', ['model', 'endpoint', 'status'])

# 请求延迟直方图
REQUEST_LATENCY = Histogram('llm_request_latency_seconds', 'LLM request latency in seconds', ['model', 'endpoint'])

# Token处理统计
TOKENS_PROCESSED = Counter('llm_tokens_processed_total', 'Total number of tokens processed', ['model', 'type'])

# 当前活跃请求数
ACTIVE_REQUESTS = Gauge('llm_active_requests', 'Number of active LLM requests', ['model'])

# 缓存命中率
CACHE_HIT_RATE = Summary('llm_cache_hit_rate', 'Cache hit rate', ['model'])

def llm_request_middleware(func):
    def wrapper(model, prompt, **kwargs):
        # 获取模型名称和端点
        model_name = model.name if hasattr(model, 'name') else 'unknown'
        endpoint = kwargs.get('endpoint', 'default')
        
        # 增加活跃请求计数
        ACTIVE_REQUESTS.labels(model=model_name).inc()
        
        # 记录请求开始时间
        start_time = time.time()
        
        try:
            # 处理请求前记录输入tokens
            input_tokens = len(prompt.split())  # 简化计算，实际应使用tokenizer
            TOKENS_PROCESSED.labels(model=model_name, type='input').inc(input_tokens)
            
            # 调用原始函数
            result = func(model, prompt, **kwargs)
            
            # 处理请求后记录输出tokens
            output_tokens = len(result.split())  # 简化计算
            TOKENS_PROCESSED.labels(model=model_name, type='output').inc(output_tokens)
            
            # 记录成功请求
            REQUEST_COUNT.labels(model=model_name, endpoint=endpoint, status='success').inc()
            
            return result
        except Exception as e:
            # 记录失败请求
            REQUEST_COUNT.labels(model=model_name, endpoint=endpoint, status='error').inc()
            raise e
        finally:
            # 记录请求延迟
            latency = time.time() - start_time
            REQUEST_LATENCY.labels(model=model_name, endpoint=endpoint).observe(latency)
            
            # 减少活跃请求计数
            ACTIVE_REQUESTS.labels(model=model_name).dec()
    
    return wrapper

# Prometheus配置示例 (prometheus.yml)
global:
  scrape_interval: 15s
  evaluation_interval: 15s

alerting:
  alertmanagers:
  - static_configs:
    - targets:
      - alertmanager:9093

rule_files:
  - "llm_alerts.yml"

scrape_configs:
  - job_name: 'prometheus'
    static_configs:
    - targets: ['localhost:9090']

  - job_name: 'node_exporter'
    static_configs:
    - targets: ['node-exporter:9100']

  - job_name: 'gpu_exporter'
    static_configs:
    - targets: ['gpu-exporter:9101']

  - job_name: 'llm_inference'
    static_configs:
    - targets: ['llm-service:9091']

  - job_name: 'kubernetes-apiservers'
    kubernetes_sd_configs:
    - role: endpoints
    scheme: https
    tls_config:
      ca_file: /var/run/secrets/kubernetes.io/serviceaccount/ca.crt
    bearer_token_file: /var/run/secrets/kubernetes.io/serviceaccount/token
    relabel_configs:
    - source_labels: [__meta_kubernetes_namespace, __meta_kubernetes_service_name, __meta_kubernetes_endpoint_port_name]
      action: keep
      regex: default;kubernetes;https

# 合规性自动检查脚本示例
import json
import os
import re
from datetime import datetime, timedelta

def check_data_retention_compliance(log_dir, retention_days=90):
    """检查日志数据保留是否符合合规要求"""
    compliance_issues = []
    cutoff_date = datetime.now() - timedelta(days=retention_days)
    
    for root, dirs, files in os.walk(log_dir):
        for file in files:
            if file.endswith('.log'):
                file_path = os.path.join(root, file)
                try:
                    # 检查文件修改时间
                    mtime = datetime.fromtimestamp(os.path.getmtime(file_path))
                    if mtime < cutoff_date:
                        compliance_issues.append({
                            'type': 'data_retention',
                            'file': file_path,
                            'issue': f'File exceeds retention period (modified on {mtime})'
                        })
                    
                    # 检查文件是否包含敏感信息
                    with open(file_path, 'r', errors='ignore') as f:
                        content = f.read()
                        # 简单的敏感信息检测示例
                        email_pattern = r'\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b'
                        phone_pattern = r'\b\d{11}\b'  # 简单的中国手机号检测
                        
                        if re.search(email_pattern, content) or re.search(phone_pattern, content):
                            compliance_issues.append({
                                'type': 'sensitive_data',
                                'file': file_path,
                                'issue': 'File may contain sensitive data (emails or phone numbers)'
                            })
                except Exception as e:
                    compliance_issues.append({
                        'type': 'file_access',
                        'file': file_path,
                        'issue': f'Error accessing file: {str(e)}'
                    })
    
    return compliance_issues

def check_access_control_compliance(access_logs):
    """检查访问控制合规性"""
    compliance_issues = []
    privilege_escalation_patterns = [
        r'permission denied.*admin',
        r'failed login.*admin',
        r'role change.*admin'
    ]
    
    for log_file in access_logs:
        try:
            with open(log_file, 'r', errors='ignore') as f:
                for line_num, line in enumerate(f, 1):
                    for pattern in privilege_escalation_patterns:
                        if re.search(pattern, line, re.IGNORECASE):
                            compliance_issues.append({
                                'type': 'access_control',
                                'file': log_file,
                                'line': line_num,
                                'issue': f'Possible privilege escalation attempt: {line.strip()}'
                            })
        except Exception as e:
            compliance_issues.append({
                'type': 'log_access',
                'file': log_file,
                'issue': f'Error accessing log file: {str(e)}'
            })
    
    return compliance_issues

def generate_compliance_report(issues):
    """生成合规性报告"""
    report = {
        'timestamp': datetime.now().isoformat(),
        'total_issues': len(issues),
        'issues_by_type': {},
        'detailed_issues': issues
    }
    
    # 按类型统计问题
    for issue in issues:
        issue_type = issue['type']
        if issue_type not in report['issues_by_type']:
            report['issues_by_type'][issue_type] = 0
        report['issues_by_type'][issue_type] += 1
    
    return report

if __name__ == '__main__':
    # 检查数据保留合规性
    retention_issues = check_data_retention_compliance('/var/log/llm-service')
    
    # 检查访问控制合规性
    access_logs = ['/var/log/llm-service/access.log', '/var/log/llm-service/auth.log']
    access_issues = check_access_control_compliance(access_logs)
    
    # 合并所有问题
    all_issues = retention_issues + access_issues
    
    # 生成报告
    report = generate_compliance_report(all_issues)
    
    # 输出报告
    print(json.dumps(report, indent=2))
    
    # 保存报告到文件
    with open(f'compliance_report_{datetime.now().strftime("%Y%m%d_%H%M%S")}.json', 'w') as f:
        json.dump(report, f, indent=2)

# 安全事件自动响应示例
import json
import requests
import subprocess
import logging
from datetime import datetime

# 配置日志
logging.basicConfig(
    filename='/var/log/security_response.log',
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger('security_response')

def detect_security_incidents(alert_source):
    """从告警源获取安全事件"""
    try:
        # 示例：从安全监控系统API获取告警
        response = requests.get(
            f"{alert_source}/api/alerts",
            headers={"Authorization": "Bearer SECRET_KEY"},
            params={"severity": "high,critical", "status": "new"}
        )
        response.raise_for_status()
        return response.json()['alerts']
    except Exception as e:
        logger.error(f"Error fetching alerts: {e}")
        return []

def classify_incident(incident):
    """对安全事件进行分类"""
    severity = incident.get('severity', 'low')
    alert_type = incident.get('type', 'unknown')
    
    # 根据事件类型和严重程度确定响应策略
    if severity == 'critical' and alert_type == 'unauthorized_access':
        return {
            'classification': 'P0',
            'response_strategy': 'immediate_containment',
            'auto_response': True
        }
    elif severity == 'high' and alert_type == 'suspicious_activity':
        return {
            'classification': 'P1',
            'response_strategy': 'investigation_and_containment',
            'auto_response': True
        }
    else:
        return {
            'classification': 'P2',
            'response_strategy': 'monitoring',
            'auto_response': False
        }

def contain_incident(incident, classification):
    """实施事件遏制措施"""
    try:
        logger.info(f"Containing incident {incident.get('id')} - {classification.get('classification')}")
        
        # 根据不同类型的事件采取不同的遏制措施
        if incident.get('type') == 'unauthorized_access':
            # 示例：阻止可疑IP地址
            ip_address = incident.get('source_ip')
            if ip_address:
                subprocess.run([
                    'iptables', '-A', 'INPUT', '-s', ip_address, '-j', 'DROP'
                ], check=True)
                logger.info(f"Blocked IP address: {ip_address}")
        
        elif incident.get('type') == 'suspicious_activity':
            # 示例：隔离受影响的容器或服务
            affected_service = incident.get('affected_service')
            if affected_service:
                subprocess.run([
                    'kubectl', 'isolate', 'pod', f'--selector=app={affected_service}'
                ], check=True)
                logger.info(f"Isolated service: {affected_service}")
        
        return True
    except Exception as e:
        logger.error(f"Error containing incident: {e}")
        return False

def notify_team(incident, classification):
    """通知安全团队"""
    try:
        # 示例：发送Slack通知
        webhook_url = "https://hooks.slack.com/services/YOUR/SLACK/WEBHOOK"
        message = {
            "text": f"🚨 Security Incident Alert 🚨",
            "blocks": [
                {
                    "type": "header",
                    "text": {
                        "type": "plain_text",
                        "text": f"Security Incident: {classification.get('classification')}"
                    }
                },
                {
                    "type": "section",
                    "fields": [
                        {
                            "type": "mrkdwn",
                            "text": f"*ID:* {incident.get('id')}"
                        },
                        {
                            "type": "mrkdwn",
                            "text": f"*Type:* {incident.get('type')}"
                        },
                        {
                            "type": "mrkdwn",
                            "text": f"*Severity:* {incident.get('severity')}"
                        },
                        {
                            "type": "mrkdwn",
                            "text": f"*Time:* {datetime.fromtimestamp(incident.get('timestamp')).isoformat()}"
                        }
                    ]
                }
            ]
        }
        
        response = requests.post(webhook_url, json=message)
        response.raise_for_status()
        logger.info(f"Team notified about incident {incident.get('id')}")
        return True
    except Exception as e:
        logger.error(f"Error notifying team: {e}")
        return False

def handle_security_incidents(alert_source):
    """处理安全事件的主函数"""
    incidents = detect_security_incidents(alert_source)
    logger.info(f"Found {len(incidents)} high severity incidents")
    
    for incident in incidents:
        # 分类事件
        classification = classify_incident(incident)
        logger.info(f"Incident {incident.get('id')} classified as {classification.get('classification')}")
        
        # 自动响应（根据分类决定）
        if classification.get('auto_response'):
            containment_result = contain_incident(incident, classification)
            logger.info(f"Containment {'successful' if containment_result else 'failed'} for incident {incident.get('id')}")
        
        # 通知团队
        notify_team(incident, classification)
        
        # 记录事件响应
        with open('/var/log/incident_responses.json', 'a') as f:
            f.write(json.dumps({
                'incident': incident,
                'classification': classification,
                'response_time': datetime.now().isoformat(),
                'containment': classification.get('auto_response')
            }) + '\n')

if __name__ == '__main__':
    # 从配置中获取告警源
    alert_source = "https://security-monitor.example.com"
    handle_security_incidents(alert_source)

# 使用OpenTelemetry进行分布式追踪的示例
from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.exporter.jaeger.thrift import JaegerExporter
from opentelemetry.sdk.resources import SERVICE_NAME, Resource
from opentelemetry.propagate import inject
import requests
import time

# 配置追踪提供者
resource = Resource(attributes={
    SERVICE_NAME: "llm-inference-service"
})

provider = TracerProvider(resource=resource)

# 配置Jaeger导出器
jaeger_exporter = JaegerExporter(
    agent_host_name="jaeger-agent",
    agent_port=6831,
)

# 添加批处理跨度处理器
processor = BatchSpanProcessor(jaeger_exporter)
provider.add_span_processor(processor)

# 设置全局追踪提供者
trace.set_tracer_provider(provider)

# 获取tracer
tracer = trace.get_tracer(__name__)

def llm_inference_handler(request):
    """LLM推理请求处理函数"""
    # 创建根跨度
    with tracer.start_as_current_span("llm_inference_request") as span:
        # 设置请求属性
        span.set_attribute("http.method", request.method)
        span.set_attribute("http.url", request.url)
        span.set_attribute("model.name", request.json.get("model", "unknown"))
        
        try:
            # 预处理阶段
            with tracer.start_as_current_span("preprocess_request") as preprocess_span:
                # 预处理逻辑
                prompt = request.json.get("prompt", "")
                max_tokens = request.json.get("max_tokens", 100)
                preprocess_span.set_attribute("prompt.length", len(prompt))
                preprocess_span.set_attribute("max_tokens", max_tokens)
                
                # 模拟预处理时间
                time.sleep(0.01)
            
            # 缓存检查阶段
            with tracer.start_as_current_span("check_cache") as cache_span:
                # 缓存检查逻辑
                cache_key = hash(prompt)
                cache_hit = False  # 模拟缓存未命中
                cache_span.set_attribute("cache.key", cache_key)
                cache_span.set_attribute("cache.hit", cache_hit)
                
                # 模拟缓存操作时间
                time.sleep(0.005)
            
            # 如果缓存未命中，执行推理
            if not cache_hit:
                with tracer.start_as_current_span("execute_inference") as inference_span:
                    # 调用推理引擎
                    result = execute_model_inference(prompt, max_tokens, inference_span)
                    inference_span.set_attribute("result.length", len(result))
                    
                    # 更新缓存
                    with tracer.start_as_current_span("update_cache") as update_cache_span:
                        # 缓存更新逻辑
                        update_cache_span.set_attribute("cache.key", cache_key)
                        # 模拟缓存更新时间
                        time.sleep(0.005)
            
            # 后处理阶段
            with tracer.start_as_current_span("postprocess_response") as postprocess_span:
                # 后处理逻辑
                # 模拟后处理时间
                time.sleep(0.01)
            
            # 返回结果
            span.set_status(trace.StatusCode.OK)
            return {"result": result, "status": "success"}
            
        except Exception as e:
            # 记录错误
            span.set_status(trace.StatusCode.ERROR, description=str(e))
            span.record_exception(e)
            raise

def execute_model_inference(prompt, max_tokens, parent_span):
    """执行模型推理"""
    # 为模型调用创建子跨度
    with tracer.start_as_current_span("model_inference", context=parent_span.get_span_context()) as span:
        # 模拟调用推理引擎
        # 在实际应用中，这里会调用vLLM、TGI等推理引擎
        
        # 设置模型调用属性
        span.set_attribute("model.inference.prompt_tokens", len(prompt.split()))
        span.set_attribute("model.inference.max_tokens", max_tokens)
        
        # 模拟不同的token生成阶段
        with tracer.start_as_current_span("token_generation") as token_span:
            # 模拟token生成时间
            time.sleep(0.5)  # 模拟生成延迟
            generated_tokens = min(max_tokens, 50)  # 模拟生成了一些tokens
            token_span.set_attribute("model.inference.generated_tokens", generated_tokens)
        
        # 返回模拟结果
        return f"This is a simulated response with {generated_tokens} tokens generated for prompt: {prompt[:20]}..."

# 模拟HTTP请求处理
def simulate_request_handler():
    # 模拟HTTP请求
    class MockRequest:
        method = "POST"
        url = "/api/generate"
        json = {
            "model": "llama-3-70b",
            "prompt": "Write a short story about AI and humans working together",
            "max_tokens": 100
        }
    
    # 处理请求
    result = llm_inference_handler(MockRequest())
    print(f"Request processed: {result}")

if __name__ == "__main__":
    # 模拟处理多个请求
    for i in range(5):
        print(f"Processing request {i+1}")
        simulate_request_handler()
        time.sleep(1)
    
    # 等待一段时间以确保所有span都被导出
    time.sleep(5)

# 基于机器学习的异常检测和根因分析示例
import pandas as pd
import numpy as np
from sklearn.ensemble import IsolationForest
from sklearn.cluster import DBSCAN
from sklearn.preprocessing import StandardScaler
import matplotlib.pyplot as plt
from statsmodels.tsa.seasonal import seasonal_decompose

class LLMAnomalyDetector:
    def __init__(self, window_size=60, contamination=0.01):
        self.window_size = window_size  # 滑动窗口大小
        self.contamination = contamination  # 异常比例估计
        self.isolation_forest = IsolationForest(contamination=contamination, random_state=42)
        self.scaler = StandardScaler()
        self.baseline_stats = None
    
    def load_metrics(self, metrics_file):
        """加载监控指标数据"""
        try:
            # 假设数据格式为CSV，包含时间戳和各种指标
            df = pd.read_csv(metrics_file, parse_dates=['timestamp'])
            df.set_index('timestamp', inplace=True)
            return df
        except Exception as e:
            print(f"Error loading metrics: {e}")
            return None
    
    def establish_baseline(self, df):
        """建立正常状态基线"""
        # 计算每个指标的统计信息
        self.baseline_stats = df.describe()
        print("Baseline statistics established:")
        print(self.baseline_stats)
    
    def detect_anomalies(self, df):
        """检测异常"""
        if df is None or df.empty:
            return None
        
        # 选择关键指标进行异常检测
        features = ['request_latency', 'gpu_utilization', 'memory_usage', 'error_rate']
        if not all(col in df.columns for col in features):
            print("Required features not found in data")
            return df
        
        # 标准化数据
        X = df[features].copy()
        X_scaled = self.scaler.fit_transform(X)
        
        # 使用隔离森林检测异常
        predictions = self.isolation_forest.fit_predict(X_scaled)
        df['anomaly'] = predictions  # -1表示异常，1表示正常
        
        # 找出异常点
        anomalies = df[df['anomaly'] == -1]
        print(f"Detected {len(anomalies)} anomalies")
        
        return df, anomalies
    
    def analyze_time_series(self, df, metric_name, period=1440):
        """进行时间序列分析"""
        if metric_name not in df.columns:
            print(f"Metric {metric_name} not found")
            return None
        
        # 确保数据按时间排序
        df_sorted = df.sort_index()
        
        try:
            # 分解时间序列
            result = seasonal_decompose(df_sorted[metric_name], model='additive', period=period, extrapolate_trend='freq')
            
            # 计算残差的异常
            df_sorted['residual'] = result.resid
            df_sorted['trend'] = result.trend
            df_sorted['seasonal'] = result.seasonal
            
            # 基于残差的异常检测
            residual_std = df_sorted['residual'].std()
            residual_mean = df_sorted['residual'].mean()
            threshold = 3 * residual_std  # 3倍标准差作为阈值
            
            df_sorted['ts_anomaly'] = np.abs(df_sorted['residual'] - residual_mean) > threshold
            
            print(f"Time series analysis completed for {metric_name}")
            print(f"Detected {df_sorted['ts_anomaly'].sum()} time series anomalies")
            
            return df_sorted
        except Exception as e:
            print(f"Error in time series analysis: {e}")
            return None
    
    def correlate_anomalies(self, df_anomalies):
        """关联分析不同指标的异常"""
        if df_anomalies is None or df_anomalies.empty:
            return None
        
        # 找出异常发生的时间点
        anomaly_times = df_anomalies.index
        
        # 计算不同指标异常的时间相关性
        print("Anomaly correlation analysis:")
        print(f"Total anomaly time points: {len(anomaly_times)}")
        
        # 在实际应用中，这里会进行更复杂的相关性分析
        # 例如，使用DBSCAN聚类找出异常模式
        
        return anomaly_times
    
    def generate_root_cause_hypotheses(self, anomalies, metrics_df):
        """基于检测到的异常生成根因假设"""
        if anomalies.empty:
            return []
        
        hypotheses = []
        
        # 分析异常特征
        for idx, anomaly in anomalies.iterrows():
            # 构建假设的示例逻辑
            if anomaly.get('error_rate', 0) > 0.1:
                if anomaly.get('gpu_utilization', 0) > 95:
                    hypotheses.append({
                        'timestamp': idx,
                        'type': 'GPU_OVERLOAD',
                        'confidence': 0.85,
                        'description': 'High error rate correlated with GPU utilization > 95%',
                        'recommended_action': 'Scale up GPU resources or reduce request rate'
                    })
                elif anomaly.get('memory_usage', 0) > 0.9:
                    hypotheses.append({
                        'timestamp': idx,
                        'type': 'MEMORY_PRESSURE',
                        'confidence': 0.8,
                        'description': 'High error rate correlated with memory usage > 90%',
                        'recommended_action': 'Increase memory limits or optimize memory usage'
                    })
            elif anomaly.get('request_latency', 0) > metrics_df['request_latency'].mean() * 3:
                if anomaly.get('gpu_utilization', 0) < 10:
                    hypotheses.append({
                        'timestamp': idx,
                        'type': 'GPU_UNDERUTILIZATION',
                        'confidence': 0.75,
                        'description': 'High latency with low GPU utilization',
                        'recommended_action': 'Check GPU driver, model loading, or request queuing issues'
                    })
        
        # 按时间和置信度排序假设
        hypotheses.sort(key=lambda x: (x['timestamp'], -x['confidence']))
        
        print(f"Generated {len(hypotheses)} root cause hypotheses")
        
        return hypotheses

def main():
    # 创建异常检测器实例
    detector = LLMAnomalyDetector(window_size=60, contamination=0.02)
    
    # 这里应该加载实际的监控数据，现在模拟一些数据
    # 生成模拟数据
    dates = pd.date_range(start='2025-01-01', end='2025-01-02', freq='1min')
    n_samples = len(dates)
    
    # 生成正常数据
    request_latency = np.random.normal(0.5, 0.1, n_samples)
    gpu_utilization = np.random.normal(70, 10, n_samples)
    memory_usage = np.random.normal(0.75, 0.05, n_samples)
    error_rate = np.random.normal(0.01, 0.005, n_samples)
    
    # 插入异常数据点
    # GPU过载异常
    gpu_utilization[100:120] = np.random.normal(98, 1, 20)
    error_rate[100:120] = np.random.normal(0.2, 0.05, 20)
    request_latency[100:120] = np.random.normal(2.0, 0.5, 20)
    
    # 内存压力异常
    memory_usage[300:320] = np.random.normal(0.95, 0.02, 20)
    error_rate[300:320] = np.random.normal(0.15, 0.03, 20)
    
    # GPU未充分利用异常
    gpu_utilization[500:520] = np.random.normal(5, 2, 20)
    request_latency[500:520] = np.random.normal(3.0, 0.8, 20)
    
    # 创建DataFrame
    df = pd.DataFrame({
        'timestamp': dates,
        'request_latency': request_latency,
        'gpu_utilization': gpu_utilization,
        'memory_usage': memory_usage,
        'error_rate': error_rate
    })
    df.set_index('timestamp', inplace=True)
    
    # 建立基线
    detector.establish_baseline(df)
    
    # 检测异常
    df_with_anomalies, anomalies = detector.detect_anomalies(df)
    
    # 时间序列分析
    ts_analyzed = detector.analyze_time_series(df, 'request_latency')
    
    # 关联分析
    anomaly_times = detector.correlate_anomalies(anomalies)
    
    # 生成根因假设
    hypotheses = detector.generate_root_cause_hypotheses(anomalies, df)
    
    # 打印根因分析结果
    print("\nRoot Cause Analysis Results:")
    for i, hypothesis in enumerate(hypotheses[:5]):  # 只显示前5个假设
        print(f"\nHypothesis {i+1}:")
        print(f"  Time: {hypothesis['timestamp']}")
        print(f"  Type: {hypothesis['type']}")
        print(f"  Confidence: {hypothesis['confidence']:.2f}")
        print(f"  Description: {hypothesis['description']}")
        print(f"  Recommended Action: {hypothesis['recommended_action']}")

if __name__ == "__main__":
    main()

自愈系统架构
┌─────────────────────────────────────────────────────────┐
│                     自愈决策引擎                         │
│  ┌─────────────┐  ┌─────────────┐  ┌─────────────────┐  │
│  │ 策略管理器  │  │ 动作协调器  │  │ 效果评估器     │  │
│  └─────────────┘  └─────────────┘  └─────────────────┘  │
└───────────┬─────────────────┬─────────────────────────┘
            │                 │
┌───────────▼────────┐  ┌─────▼─────────────────────────┐
│ 监控系统集成       │  │ 自动执行系统                   │
│ - 告警接入         │  │ - 执行引擎                     │
│ - 状态反馈         │  │ - 动作队列                     │
│ - 事件触发         │  │ - 安全控制                     │
└────────────────────┘  └───────────────────────────────┘

# 基础设施自动恢复示例代码
import time
import subprocess
import logging
from kubernetes import client, config

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("llm_self_healing")

class InfrastructureHealer:
    def __init__(self):
        # 加载Kubernetes配置
        config.load_kube_config()
        self.v1 = client.CoreV1Api()
        self.apps_v1 = client.AppsV1Api()
        # 定义重试参数
        self.max_retries = 3
        self.retry_delay = 60  # 秒
        
    def check_node_health(self, node_name):
        """检查节点健康状态"""
        try:
            node = self.v1.read_node(node_name)
            conditions = {cond.type: cond.status for cond in node.status.conditions}
            # 检查节点是否Ready
            return conditions.get('Ready') == 'True'
        except Exception as e:
            logger.error(f"检查节点{node_name}健康状态失败: {str(e)}")
            return False
    
    def restart_pod(self, namespace, pod_name):
        """重启Pod"""
        try:
            logger.info(f"重启Pod {namespace}/{pod_name}")
            # 删除Pod，Kubernetes会自动重建
            self.v1.delete_namespaced_pod(name=pod_name, namespace=namespace)
            # 等待Pod重启
            time.sleep(10)
            return True
        except Exception as e:
            logger.error(f"重启Pod {namespace}/{pod_name}失败: {str(e)}")
            return False
    
    def scale_deployment(self, namespace, deployment_name, replicas):
        """扩展Deployment副本数"""
        try:
            logger.info(f"调整Deployment {namespace}/{deployment_name}副本数为{replicas}")
            deployment = self.apps_v1.read_namespaced_deployment(name=deployment_name, namespace=namespace)
            deployment.spec.replicas = replicas
            self.apps_v1.replace_namespaced_deployment(name=deployment_name, namespace=namespace, body=deployment)
            return True
        except Exception as e:
            logger.error(f"调整Deployment {namespace}/{deployment_name}副本数失败: {str(e)}")
            return False
    
    def handle_gpu_failure(self, node_name):
        """处理GPU故障"""
        logger.warning(f"检测到节点{node_name}可能存在GPU故障")
        
        # 1. 将节点标记为不可调度
        try:
            self.v1.patch_node(node_name, {"spec": {"unschedulable": True}})
            logger.info(f"已将节点{node_name}标记为不可调度")
        except Exception as e:
            logger.error(f"标记节点{node_name}为不可调度失败: {str(e)}")
        
        # 2. 获取该节点上的所有Pod
        try:
            pods = self.v1.list_namespaced_pod("default", field_selector=f"spec.nodeName={node_name}")
            logger.info(f"节点{node_name}上有{len(pods.items)}个Pod需要迁移")
            
            # 3. 逐个删除Pod，让它们在健康节点上重建
            for pod in pods.items:
                logger.info(f"迁移Pod {pod.metadata.name}")
                self.v1.delete_namespaced_pod(pod.metadata.name, "default")
                time.sleep(2)  # 避免同时删除过多Pod
        except Exception as e:
            logger.error(f"迁移节点{node_name}上的Pod失败: {str(e)}")
        
        # 4. 记录事件并通知维护人员
        logger.error(f"节点{node_name}存在GPU故障，请及时维护")
    
    def auto_recover(self, event_type, details):
        """根据事件类型执行自动恢复"""
        if event_type == "pod_crash_loop":
            # 处理Pod崩溃循环
            namespace = details.get("namespace", "default")
            pod_name = details.get("pod_name")
            restart_count = details.get("restart_count", 0)
            
            if restart_count < self.max_retries:
                logger.info(f"检测到Pod {pod_name}崩溃循环，尝试第{restart_count+1}次重启")
                return self.restart_pod(namespace, pod_name)
            else:
                logger.error(f"Pod {pod_name}已多次重启失败，升级为严重故障")
                # 这里可以触发更高级别的处理，如通知人工干预
                return False
        
        elif event_type == "high_latency":
            # 处理高延迟事件
            namespace = details.get("namespace", "default")
            deployment_name = details.get("deployment_name")
            current_replicas = details.get("current_replicas", 1)
            
            # 扩展副本数以应对高负载
            new_replicas = min(current_replicas * 2, 10)  # 最多扩展到10个副本
            logger.info(f"检测到高延迟，将Deployment {deployment_name}从{current_replicas}扩展到{new_replicas}个副本")
            return self.scale_deployment(namespace, deployment_name, new_replicas)
        
        elif event_type == "gpu_failure":
            # 处理GPU故障
            node_name = details.get("node_name")
            return self.handle_gpu_failure(node_name)
        
        else:
            logger.warning(f"未知的事件类型: {event_type}")
            return False

# 使用示例
if __name__ == "__main__":
    healer = InfrastructureHealer()
    
    # 模拟处理一个Pod崩溃循环事件
    healer.auto_recover("pod_crash_loop", {
        "namespace": "default",
        "pod_name": "llm-inference-78549d567c-x7v4h",
        "restart_count": 1
    })
    
    # 模拟处理高延迟事件
    healer.auto_recover("high_latency", {
        "namespace": "default",
        "deployment_name": "llm-inference",
        "current_replicas": 2
    })

# 动态降级与容错配置示例
import json
import time
import threading
from flask import Flask, request, jsonify

app = Flask(__name__)

class DegradationManager:
    def __init__(self):
        # 初始降级级别
        self.degradation_level = 0  # 0: 正常, 1: 轻度, 2: 中度, 3: 重度, 4: 紧急
        self.last_check_time = time.time()
        self.check_interval = 30  # 秒
        self.metrics = {
            "request_latency": 0,
            "error_rate": 0,
            "gpu_utilization": 0,
            "memory_usage": 0,
            "active_requests": 0
        }
        self.lock = threading.RLock()
        
        # 启动监控线程
        self.monitor_thread = threading.Thread(target=self._monitor_loop)
        self.monitor_thread.daemon = True
        self.monitor_thread.start()
    
    def update_metrics(self, metrics_data):
        """更新监控指标"""
        with self.lock:
            for key, value in metrics_data.items():
                if key in self.metrics:
                    self.metrics[key] = value
    
    def get_degradation_level(self):
        """获取当前降级级别"""
        with self.lock:
            return self.degradation_level
    
    def _monitor_loop(self):
        """监控循环，定期评估系统状态并调整降级级别"""
        while True:
            time.sleep(self.check_interval)
            self._evaluate_degradation()
    
    def _evaluate_degradation(self):
        """根据监控指标评估系统状态并调整降级级别"""
        with self.lock:
            metrics = self.metrics.copy()
        
        # 评估降级级别
        new_level = 0
        
        # 紧急降级条件
        if metrics["error_rate"] > 0.5 or metrics["active_requests"] > 1000:
            new_level = 4
        # 重度降级条件
        elif metrics["request_latency"] > 10 or metrics["gpu_utilization"] > 98 or metrics["memory_usage"] > 0.98:
            new_level = 3
        # 中度降级条件
        elif metrics["request_latency"] > 5 or metrics["gpu_utilization"] > 90 or metrics["memory_usage"] > 0.95:
            new_level = 2
        # 轻度降级条件
        elif metrics["request_latency"] > 2 or metrics["error_rate"] > 0.05:
            new_level = 1
        
        # 更新降级级别
        with self.lock:
            if new_level != self.degradation_level:
                print(f"降级级别变更: {self.degradation_level} -> {new_level}")
                self.degradation_level = new_level
                self._apply_degradation_policy(new_level)
    
    def _apply_degradation_policy(self, level):
        """应用降级策略"""
        policies = {
            0: "正常模式，启用所有功能",
            1: "轻度降级：禁用非核心功能，保留主要服务",
            2: "中度降级：降低模型精度，限制输出长度",
            3: "重度降级：切换到小模型，只提供基本功能",
            4: "紧急降级：只返回状态码和基本错误信息"
        }
        
        print(f"应用降级策略: {policies.get(level, '未知策略')}")
        # 这里可以添加实际的策略应用代码，如更新配置、重启服务等

# 创建降级管理器实例
degradation_manager = DegradationManager()

@app.route('/api/llm/generate', methods=['POST'])
def generate_text():
    degradation_level = degradation_manager.get_degradation_level()
    
    # 紧急降级：只返回基本响应
    if degradation_level >= 4:
        return jsonify({
            "status": "degraded",
            "message": "服务暂时不可用，请稍后再试",
            "degradation_level": degradation_level
        }), 503
    
    # 解析请求
    try:
        data = request.json
        prompt = data.get('prompt', '')
        max_tokens = data.get('max_tokens', 100)
    except:
        return jsonify({"error": "无效的请求格式"}), 400
    
    # 根据降级级别调整参数
    if degradation_level >= 3:  # 重度降级
        # 限制输入长度
        prompt = prompt[:200]
        # 限制输出长度
        max_tokens = min(max_tokens, 50)
        # 可以在这里切换到小模型
        model_to_use = "small_model"
    elif degradation_level >= 2:  # 中度降级
        prompt = prompt[:500]
        max_tokens = min(max_tokens, 100)
        model_to_use = "medium_model"
    else:  # 轻度或正常
        model_to_use = "large_model"
    
    # 模拟模型调用（实际应用中替换为真实的模型调用）
    try:
        # 增加活动请求计数
        degradation_manager.update_metrics({"active_requests": degradation_manager.metrics["active_requests"] + 1})
        
        # 模拟处理时间
        if degradation_level >= 2:
            time.sleep(0.5)  # 小模型更快
        else:
            time.sleep(1.5)  # 大模型较慢
        
        # 生成模拟响应
        response_text = f"这是来自{model_to_use}的响应。降级级别: {degradation_level}"
        
        # 更新延迟指标（模拟）
        if degradation_level == 0:
            latency = 1.2
        elif degradation_level == 1:
            latency = 1.5
        elif degradation_level == 2:
            latency = 2.5
        else:
            latency = 3.8
        degradation_manager.update_metrics({"request_latency": latency})
        
        # 减少活动请求计数
        degradation_manager.update_metrics({"active_requests": degradation_manager.metrics["active_requests"] - 1})
        
        # 根据降级级别决定响应内容
        response = {
            "text": response_text,
            "degradation_level": degradation_level,
            "model_used": model_to_use
        }
        
        # 正常模式下添加更多详细信息
        if degradation_level == 0:
            response.update({
                "usage": {"prompt_tokens": len(prompt), "completion_tokens": len(response_text)},
                "metadata": {"timestamp": time.time(), "server_id": "server-01"}
            })
        
        return jsonify(response)
    except Exception as e:
        # 更新错误率指标（模拟）
        degradation_manager.update_metrics({"error_rate": min(degradation_manager.metrics["error_rate"] + 0.01, 1.0)})
        # 减少活动请求计数
        degradation_manager.update_metrics({"active_requests": max(degradation_manager.metrics["active_requests"] - 1, 0)})
        
        return jsonify({"error": str(e)}), 500

@app.route('/api/system/status', methods=['GET'])
def system_status():
    """获取系统状态"""
    return jsonify({
        "degradation_level": degradation_manager.get_degradation_level(),
        "metrics": degradation_manager.metrics,
        "timestamp": time.time()
    })

if __name__ == '__main__':
    app.run(host='0.0.0.0', port=5000)

# 模型量化与优化示例代码
import torch
import transformers
from transformers import AutoModelForCausalLM, AutoTokenizer
import time
import numpy as np

class ModelOptimizer:
    def __init__(self):
        self.optimization_methods = {
            "fp32": self._fp32_inference,
            "fp16": self._fp16_inference,
            "int8": self._int8_inference,
            "vllm": self._vllm_inference,
            "awq": self._awq_inference
        }
    
    def load_model(self, model_name, optimization=None):
        """加载并优化模型"""
        print(f"加载模型: {model_name}, 优化方式: {optimization}")
        
        # 这里根据不同的优化方式加载模型
        # 实际应用中，需要根据不同的优化库进行调整
        if optimization == "vllm":
            # vLLM优化（示例）
            from vllm import LLM
            model = LLM(model=model_name, quantization=optimization)
            tokenizer = transformers.AutoTokenizer.from_pretrained(model_name)
        elif optimization == "awq":
            # AWQ量化（示例）
            model = AutoModelForCausalLM.from_pretrained(
                model_name, 
                torch_dtype=torch.float16,
                quantization_config={"load_in_awq": True}
            ).cuda()
            tokenizer = AutoTokenizer.from_pretrained(model_name)
        elif optimization == "int8":
            # INT8量化
            model = AutoModelForCausalLM.from_pretrained(
                model_name, 
                torch_dtype=torch.float16,
                load_in_8bit=True
            )
            tokenizer = AutoTokenizer.from_pretrained(model_name)
        elif optimization == "fp16":
            # FP16半精度
            model = AutoModelForCausalLM.from_pretrained(
                model_name, 
                torch_dtype=torch.float16
            ).cuda()
            tokenizer = AutoTokenizer.from_pretrained(model_name)
        else:
            # 默认FP32
            model = AutoModelForCausalLM.from_pretrained(
                model_name, 
                torch_dtype=torch.float32
            ).cuda()
            tokenizer = AutoTokenizer.from_pretrained(model_name)
        
        return model, tokenizer
    
    def benchmark_inference(self, model, tokenizer, prompt, max_new_tokens=100, num_runs=5):
        """基准测试推理性能"""
        # 预热
        for _ in range(2):
            inputs = tokenizer(prompt, return_tensors="pt").to("cuda")
            _ = model.generate(**inputs, max_new_tokens=10)
        
        # 基准测试
        latencies = []
        tokens_generated = []
        
        for i in range(num_runs):
            inputs = tokenizer(prompt, return_tensors="pt").to("cuda")
            prompt_tokens = inputs.input_ids.shape[1]
            
            torch.cuda.synchronize()
            start_time = time.time()
            
            output = model.generate(
                **inputs, 
                max_new_tokens=max_new_tokens,
                do_sample=True,
                temperature=0.7
            )
            
            torch.cuda.synchronize()
            end_time = time.time()
            
            total_tokens = output.shape[1]
            new_tokens = total_tokens - prompt_tokens
            tokens_generated.append(new_tokens)
            
            latency = end_time - start_time
            latencies.append(latency)
            
            print(f"运行 {i+1}: {latency:.4f}秒, 生成{new_tokens}个token, {new_tokens/latency:.2f} tokens/sec")
        
        avg_latency = np.mean(latencies)
        avg_throughput = np.mean(tokens_generated) / avg_latency
        
        print(f"平均延迟: {avg_latency:.4f}秒")
        print(f"平均吞吐量: {avg_throughput:.2f} tokens/sec")
        
        return {
            "avg_latency": avg_latency,
            "avg_throughput": avg_throughput,
            "min_latency": min(latencies),
            "max_latency": max(latencies),
            "total_tokens": sum(tokens_generated)
        }
    
    def compare_optimizations(self, model_name, prompt, max_new_tokens=100):
        """比较不同优化方法的性能"""
        results = {}
        
        # 测试不同的优化方法
        for method in ["fp32", "fp16", "int8", "vllm", "awq"]:
            try:
                print(f"\n测试优化方法: {method}")
                model, tokenizer = self.load_model(model_name, method)
                result = self.benchmark_inference(
                    model, 
                    tokenizer, 
                    prompt, 
                    max_new_tokens
                )
                results[method] = result
                # 释放显存
                del model
                torch.cuda.empty_cache()
            except Exception as e:
                print(f"优化方法 {method} 测试失败: {str(e)}")
                results[method] = {"error": str(e)}
        
        # 打印比较结果
        print("\n性能比较结果:")
        print("-" * 80)
        print(f"{'优化方法':<10} {'平均延迟(秒)':<15} {'平均吞吐量(tokens/sec)':<25} {'性能提升':<15}")
        print("-" * 80)
        
        # 以FP32为基准计算性能提升
        base_latency = results.get("fp32", {}).get("avg_latency", 1)
        
        for method, result in results.items():
            if "error" not in result:
                latency = result["avg_latency"]
                throughput = result["avg_throughput"]
                speedup = base_latency / latency if base_latency > 0 else 0
                print(f"{method:<10} {latency:<15.4f} {throughput:<25.2f} {speedup:<15.2f}x")
        
        print("-" * 80)
        
        return results
    
    # 以下是不同优化方法的具体实现（示例）
    def _fp32_inference(self, model, inputs):
        return model.generate(**inputs)
    
    def _fp16_inference(self, model, inputs):
        return model.generate(**inputs)
    
    def _int8_inference(self, model, inputs):
        return model.generate(**inputs)
    
    def _vllm_inference(self, model, inputs):
        # vLLM特定的推理逻辑
        return model.generate(**inputs)
    
    def _awq_inference(self, model, inputs):
        # AWQ特定的推理逻辑
        return model.generate(**inputs)

# 使用示例
if __name__ == "__main__":
    optimizer = ModelOptimizer()
    
    # 选择一个较小的模型进行测试（实际应用中可以选择更大的模型）
    model_name = "gpt2-xl"  # 可以替换为其他模型
    prompt = "解释量子计算的基本原理，并给出一个简单的例子。"
    
    # 比较不同优化方法的性能
    optimizer.compare_optimizations(model_name, prompt, max_new_tokens=100)

# 成本监控与分析示例代码
import pandas as pd
import matplotlib.pyplot as plt
import numpy as np
from datetime import datetime, timedelta
import seaborn as sns

class CostMonitor:
    def __init__(self):
        # 定义成本相关参数
        self.gpu_cost_per_hour = {
            "A10G": 3.0,   # 每小时美元成本
            "A100": 12.0,
            "H100": 24.0,
            "L4": 2.5
        }
        
        self.cpu_cost_per_core_hour = 0.05
        self.memory_cost_per_gb_hour = 0.01
        self.storage_cost_per_tb_month = 20.0
        
    def generate_usage_data(self, days=30):
        """生成模拟的资源使用数据"""
        # 生成时间序列
        end_date = datetime.now()
        start_date = end_date - timedelta(days=days)
        dates = pd.date_range(start=start_date, end=end_date, freq='H')
        
        # 创建数据框
        data = []
        
        # 生成每小时的使用数据
        for date in dates:
            # 模拟工作日和周末的负载差异
            is_weekend = date.weekday() >= 5
            
            # 模拟一天中的负载变化（工作日）
            if not is_weekend:
                hour = date.hour
                if 9 <= hour < 18:
                    # 工作时间，负载较高
                    base_load = 10
                elif 18 <= hour < 22:
                    # 晚上，负载中等
                    base_load = 5
                else:
                    # 深夜到早晨，负载较低
                    base_load = 2
            else:
                # 周末，整体负载较低
                base_load = 3
            
            # 添加一些随机波动
            noise = np.random.normal(0, 0.5)
            gpu_count = max(1, int(base_load + noise))
            
            # 随机选择使用的GPU类型
            gpu_types = list(self.gpu_cost_per_hour.keys())
            if gpu_count <= 2:
                # 负载低时倾向于使用更便宜的GPU
                weights = [0.6, 0.2, 0.1, 0.1]  # 偏向A10G和L4
            else:
                # 负载高时使用更多高性能GPU
                weights = [0.3, 0.3, 0.3, 0.1]  # 更多A100和H100
            
            gpu_type = np.random.choice(gpu_types, p=weights)
            
            # 计算其他资源使用
            cpu_cores = gpu_count * np.random.randint(8, 16)
            memory_gb = gpu_count * np.random.randint(64, 128)
            
            # 计算每小时成本
            gpu_cost = gpu_count * self.gpu_cost_per_hour[gpu_type]
            cpu_cost = cpu_cores * self.cpu_cost_per_core_hour
            memory_cost = memory_gb * self.memory_cost_per_gb_hour
            total_hourly_cost = gpu_cost + cpu_cost + memory_cost
            
            # 添加请求数和吞吐量
            requests = int(np.random.normal(gpu_count * 1000, gpu_count * 200))
            tokens_processed = int(np.random.normal(gpu_count * 50000, gpu_count * 10000))
            
            # 记录数据
            data.append({
                'timestamp': date,
                'gpu_count': gpu_count,
                'gpu_type': gpu_type,
                'cpu_cores': cpu_cores,
                'memory_gb': memory_gb,
                'gpu_cost': gpu_cost,
                'cpu_cost': cpu_cost,
                'memory_cost': memory_cost,
                'total_cost': total_hourly_cost,
                'requests': max(0, requests),
                'tokens_processed': max(0, tokens_processed)
            })
        
        return pd.DataFrame(data)
    
    def analyze_costs(self, df):
        """分析成本数据"""
        # 按天汇总成本
        daily_costs = df.resample('D', on='timestamp').agg({
            'gpu_cost': 'sum',
            'cpu_cost': 'sum',
            'memory_cost': 'sum',
            'total_cost': 'sum',
            'requests': 'sum',
            'tokens_processed': 'sum'
        }).reset_index()
        
        # 计算每天的单位成本
        daily_costs['cost_per_request'] = daily_costs['total_cost'] / daily_costs['requests']
        daily_costs['cost_per_token'] = daily_costs['total_cost'] / daily_costs['tokens_processed']
        
        # 按GPU类型汇总
        gpu_type_costs = df.groupby('gpu_type').agg({
            'gpu_cost': 'sum',
            'requests': 'sum',
            'tokens_processed': 'sum'
        }).reset_index()
        
        gpu_type_costs['cost_per_request'] = gpu_type_costs['gpu_cost'] / gpu_type_costs['requests']
        gpu_type_costs['cost_per_token'] = gpu_type_costs['gpu_cost'] / gpu_type_costs['tokens_processed']
        
        # 按小时分析使用模式
        df['hour'] = df['timestamp'].dt.hour
        hourly_stats = df.groupby('hour').agg({
            'gpu_count': 'mean',
            'total_cost': 'mean',
            'requests': 'mean',
            'tokens_processed': 'mean'
        }).reset_index()
        
        return {
            'daily_costs': daily_costs,
            'gpu_type_costs': gpu_type_costs,
            'hourly_stats': hourly_stats
        }
    
    def identify_cost_saving_opportunities(self, analysis_results):
        """识别成本节约机会"""
        opportunities = []
        
        daily_costs = analysis_results['daily_costs']
        gpu_type_costs = analysis_results['gpu_type_costs']
        hourly_stats = analysis_results['hourly_stats']
        
        # 1. 识别低利用率时段
        low_utilization_hours = hourly_stats[hourly_stats['gpu_count'] < hourly_stats['gpu_count'].quantile(0.3)]
        if not low_utilization_hours.empty:
            avg_low_hour_cost = low_utilization_hours['total_cost'].mean()
            potential_savings = avg_low_hour_cost * len(low_utilization_hours) * 0.6  # 假设可以节省60%
            opportunities.append({
                'type': '低利用率时段优化',
                'description': f"在GPU利用率低的时段（{len(low_utilization_hours)}个小时）实施自动缩容",
                'potential_savings_daily': potential_savings,
                'potential_savings_monthly': potential_savings * 30
            })
        
        # 2. GPU类型优化
        if not gpu_type_costs.empty:
            # 找出成本最高的GPU类型
            most_expensive_gpu = gpu_type_costs.loc[gpu_type_costs['cost_per_token'].idxmax()]
            cheapest_gpu = gpu_type_costs.loc[gpu_type_costs['cost_per_token'].idxmin()]
            
            cost_diff = most_expensive_gpu['cost_per_token'] / cheapest_gpu['cost_per_token'] - 1
            
            if cost_diff > 0.3:  # 如果成本差异大于30%
                opportunities.append({
                    'type': 'GPU类型优化',
                    'description': f"将部分{most_expensive_gpu['gpu_type']}任务迁移到{cheapest_gpu['gpu_type']}",
                    'potential_savings_percent': cost_diff * 100,
                    'affected_cost_percent': most_expensive_gpu['gpu_cost'] / gpu_type_costs['gpu_cost'].sum() * 100
                })
        
        # 3. 请求批处理优化
        # 假设通过优化批处理可以提高20%的吞吐量
        opportunities.append({
            'type': '请求批处理优化',
            'description': "实施更高效的请求批处理策略",
            'potential_throughput_improvement': "20%",
            'potential_cost_reduction_percent': "15-20%"
        })
        
        # 4. 模型优化建议
        opportunities.append({
            'type': '模型优化',
            'description': "使用量化、蒸馏等技术优化模型",
            'potential_cost_reduction_percent': "30-50%",
            'implementation_complexity': "中到高"
        })
        
        return opportunities
    
    def visualize_costs(self, df, analysis_results):
        """可视化成本数据"""
        plt.figure(figsize=(20, 15))
        
        # 1. 每日成本趋势
        plt.subplot(3, 2, 1)
        daily_costs = analysis_results['daily_costs']
        plt.plot(daily_costs['timestamp'], daily_costs['total_cost'], marker='o', linestyle='-', color='blue')
        plt.title('每日总成本趋势')
        plt.xlabel('日期')
        plt.ylabel('成本 ($)')
        plt.grid(True)
        
        # 2. 成本构成
        plt.subplot(3, 2, 2)
        cost_composition = df[['gpu_cost', 'cpu_cost', 'memory_cost']].sum()
        plt.pie(cost_composition, labels=cost_composition.index, autopct='%1.1f%%', startangle=90)
        plt.title('成本构成')
        
        # 3. 每小时GPU使用情况
        plt.subplot(3, 2, 3)
        hourly_stats = analysis_results['hourly_stats']
        plt.bar(hourly_stats['hour'], hourly_stats['gpu_count'], color='green')
        plt.title('平均每小时GPU使用数量')
        plt.xlabel('小时')
        plt.ylabel('GPU数量')
        plt.grid(True, axis='y')
        
        # 4. GPU类型成本对比
        plt.subplot(3, 2, 4)
        gpu_type_costs = analysis_results['gpu_type_costs']
        plt.bar(gpu_type_costs['gpu_type'], gpu_type_costs['cost_per_token'], color='orange')
        plt.title('不同GPU类型的每token成本')
        plt.xlabel('GPU类型')
        plt.ylabel('成本 per token')
        plt.grid(True, axis='y')
        
        # 5. 单位成本趋势
        plt.subplot(3, 2, 5)
        plt.plot(daily_costs['timestamp'], daily_costs['cost_per_request'], marker='s', linestyle='-', color='red')
        plt.title('每日每请求成本趋势')
        plt.xlabel('日期')
        plt.ylabel('成本 per request ($)')
        plt.grid(True)
        
        # 6. 每token成本趋势
        plt.subplot(3, 2, 6)
        plt.plot(daily_costs['timestamp'], daily_costs['cost_per_token'] * 1000, marker='^', linestyle='-', color='purple')
        plt.title('每日每千token成本趋势')
        plt.xlabel('日期')
        plt.ylabel('成本 per 1000 tokens ($)')
        plt.grid(True)
        
        plt.tight_layout()
        plt.savefig('cost_analysis.png')
        plt.close()
        
        print("成本分析图表已保存为 cost_analysis.png")
    
    def generate_cost_report(self, df, analysis_results, opportunities):
        """生成成本报告"""
        # 计算总体统计数据
        total_period = (df['timestamp'].max() - df['timestamp'].min()).days
        total_cost = df['total_cost'].sum()
        avg_daily_cost = total_cost / total_period
        
        total_requests = df['requests'].sum()
        total_tokens = df['tokens_processed'].sum()
        
        cost_per_request = total_cost / total_requests
        cost_per_token = total_cost / total_tokens
        
        # 生成报告
        report = {
            'period': {
                'start_date': df['timestamp'].min().strftime('%Y-%m-%d'),
                'end_date': df['timestamp'].max().strftime('%Y-%m-%d'),
                'days': total_period
            },
            'cost_summary': {
                'total_cost': total_cost,
                'avg_daily_cost': avg_daily_cost,
                'avg_monthly_cost': avg_daily_cost * 30,
                'gpu_cost': df['gpu_cost'].sum(),
                'cpu_cost': df['cpu_cost'].sum(),
                'memory_cost': df['memory_cost'].sum()
            },
            'performance_metrics': {
                'total_requests': total_requests,
                'total_tokens_processed': total_tokens,
                'cost_per_request': cost_per_request,
                'cost_per_token': cost_per_token,
                'cost_per_1000_tokens': cost_per_token * 1000
            },
            'saving_opportunities': opportunities
        }
        
        # 打印报告摘要
        print("===== 成本分析报告摘要 =====")
        print(f"分析期间: {report['period']['start_date']} 至 {report['period']['end_date']} ({report['period']['days']}天)")
        print(f"\n总成本: ${report['cost_summary']['total_cost']:.2f}")
        print(f"日均成本: ${report['cost_summary']['avg_daily_cost']:.2f}")
        print(f"预计月均成本: ${report['cost_summary']['avg_monthly_cost']:.2f}")
        print(f"\n成本构成:")
        print(f"  GPU: ${report['cost_summary']['gpu_cost']:.2f} ({report['cost_summary']['gpu_cost']/total_cost*100:.1f}%)")
        print(f"  CPU: ${report['cost_summary']['cpu_cost']:.2f} ({report['cost_summary']['cpu_cost']/total_cost*100:.1f}%)")
        print(f"  内存: ${report['cost_summary']['memory_cost']:.2f} ({report['cost_summary']['memory_cost']/total_cost*100:.1f}%)")
        print(f"\n性能指标:")
        print(f"  总请求数: {report['performance_metrics']['total_requests']}")
        print(f"  总处理token数: {report['performance_metrics']['total_tokens_processed']}")
        print(f"  每请求成本: ${report['performance_metrics']['cost_per_request']:.4f}")
        print(f"  每token成本: ${report['performance_metrics']['cost_per_token']:.8f}")
        print(f"  每千token成本: ${report['performance_metrics']['cost_per_1000_tokens']:.4f}")
        print(f"\n成本节约机会:")
        for i, opportunity in enumerate(opportunities, 1):
            print(f"  {i}. {opportunity['type']}: {opportunity['description']}")
            if 'potential_savings_monthly' in opportunity:
                print(f"     预计月度节约: ${opportunity['potential_savings_monthly']:.2f}")
            elif 'potential_cost_reduction_percent' in opportunity:
                print(f"     预计成本降低: {opportunity['potential_cost_reduction_percent']}%")
        
        return report

# 使用示例
if __name__ == "__main__":
    monitor = CostMonitor()
    
    # 生成模拟数据
    df = monitor.generate_usage_data(days=30)
    
    # 分析成本
    analysis_results = monitor.analyze_costs(df)
    
    # 识别成本节约机会
    opportunities = monitor.identify_cost_saving_opportunities(analysis_results)
    
    # 可视化成本数据
    monitor.visualize_costs(df, analysis_results)
    
    # 生成成本报告
    report = monitor.generate_cost_report(df, analysis_results, opportunities)

# Prometheus与Grafana集成示例 - 自定义导出器
from prometheus_client import start_http_server, Gauge
import time
import psutil
import GPUtil

# 初始化GPU和系统指标
GPU_MEMORY_USAGE = Gauge('gpu_memory_usage_percent', 'GPU memory usage percentage', ['gpu_id'])
GPU_UTILIZATION = Gauge('gpu_utilization_percent', 'GPU utilization percentage', ['gpu_id'])
CPU_USAGE = Gauge('cpu_usage_percent', 'CPU usage percentage')
MEMORY_USAGE = Gauge('memory_usage_percent', 'Memory usage percentage')
LLM_REQUEST_COUNT = Gauge('llm_request_count', 'Number of active LLM requests')
LLM_TOKEN_PROCESSING_RATE = Gauge('llm_token_processing_rate', 'Tokens processed per second')

# 模拟LLM服务状态
class MockLLMService:
    def __init__(self):
        self.active_requests = 0
        self.token_processing_rate = 0
    
    def update_metrics(self):
        # 在实际应用中，这些值应该从实际的LLM服务中获取
        import random
        self.active_requests = random.randint(0, 50)
        self.token_processing_rate = random.uniform(10.0, 200.0)
        
        # 更新Prometheus指标
        LLM_REQUEST_COUNT.set(self.active_requests)
        LLM_TOKEN_PROCESSING_RATE.set(self.token_processing_rate)

# 收集系统指标的函数
def collect_system_metrics():
    # 收集CPU和内存指标
    CPU_USAGE.set(psutil.cpu_percent(interval=1))
    MEMORY_USAGE.set(psutil.virtual_memory().percent)
    
    # 收集GPU指标
    try:
        gpus = GPUtil.getGPUs()
        for i, gpu in enumerate(gpus):
            GPU_MEMORY_USAGE.labels(gpu_id=str(i)).set(gpu.memoryUtil * 100)
            GPU_UTILIZATION.labels(gpu_id=str(i)).set(gpu.load * 100)
    except Exception as e:
        print(f"Error collecting GPU metrics: {e}")

# 主函数
if __name__ == '__main__':
    # 启动HTTP服务器，Prometheus将从中抓取指标
    start_http_server(8000)
    print("Prometheus exporter started on port 8000")
    
    # 初始化模拟LLM服务
    llm_service = MockLLMService()
    
    # 定期收集指标
    while True:
        collect_system_metrics()
        llm_service.update_metrics()
        time.sleep(15)  # 每15秒收集一次指标

# 自定义数据采集代理示例
import time
import json
import requests
import threading
import queue
import psutil
import logging
from datetime import datetime

# 配置日志
logging.basicConfig(level=logging.INFO,
                    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s')
logger = logging.getLogger('custom_monitor')

class MetricsCollector:
    def __init__(self, collection_interval=10):
        self.collection_interval = collection_interval
        self.metrics_queue = queue.Queue()
        self.running = False
        self.collector_thread = None
    
    def start(self):
        self.running = True
        self.collector_thread = threading.Thread(target=self._collect_metrics_loop)
        self.collector_thread.daemon = True
        self.collector_thread.start()
        logger.info(f"Metrics collector started with interval {self.collection_interval}s")
    
    def stop(self):
        self.running = False
        if self.collector_thread:
            self.collector_thread.join(timeout=5)
        logger.info("Metrics collector stopped")
    
    def _collect_metrics_loop(self):
        while self.running:
            try:
                metrics = self._collect_metrics()
                self.metrics_queue.put(metrics)
                logger.debug(f"Collected metrics: {metrics}")
            except Exception as e:
                logger.error(f"Error collecting metrics: {e}")
            time.sleep(self.collection_interval)
    
    def _collect_metrics(self):
        # 基础系统指标
        system_metrics = {
            "timestamp": datetime.utcnow().isoformat(),
            "host": self._get_host_info(),
            "cpu": self._get_cpu_info(),
            "memory": self._get_memory_info(),
            "disk": self._get_disk_info(),
            "network": self._get_network_info()
        }
        
        # 在实际应用中，可以添加更多特定的LLM服务指标
        # system_metrics["llm_service"] = self._get_llm_service_metrics()
        
        return system_metrics
    
    def _get_host_info(self):
        import socket
        return {
            "hostname": socket.gethostname(),
            "ip_address": socket.gethostbyname(socket.gethostname())
        }
    
    def _get_cpu_info(self):
        return {
            "percent": psutil.cpu_percent(interval=0.1, percpu=True),
            "load_avg": psutil.getloadavg() if hasattr(psutil, 'getloadavg') else None,
            "count": psutil.cpu_count(logical=True)
        }
    
    def _get_memory_info(self):
        mem = psutil.virtual_memory()
        return {
            "total": mem.total,
            "available": mem.available,
            "used": mem.used,
            "percent": mem.percent,
            "swap": dict(psutil.swap_memory()._asdict())
        }
    
    def _get_disk_info(self):
        disk_usage = {}
        for partition in psutil.disk_partitions():
            try:
                usage = psutil.disk_usage(partition.mountpoint)
                disk_usage[partition.mountpoint] = {
                    "total": usage.total,
                    "used": usage.used,
                    "free": usage.free,
                    "percent": usage.percent,
                    "device": partition.device
                }
            except (PermissionError, FileNotFoundError):
                # 某些分区可能无法访问
                continue
        return disk_usage
    
    def _get_network_info(self):
        net_io = psutil.net_io_counters()
        return {
            "bytes_sent": net_io.bytes_sent,
            "bytes_recv": net_io.bytes_recv,
            "packets_sent": net_io.packets_sent,
            "packets_recv": net_io.packets_recv,
            "errin": net_io.errin,
            "errout": net_io.errout,
            "dropin": net_io.dropin,
            "dropout": net_io.dropout
        }

class MetricsSender:
    def __init__(self, collector, endpoint_url, send_interval=30):
        self.collector = collector
        self.endpoint_url = endpoint_url
        self.send_interval = send_interval
        self.running = False
        self.sender_thread = None
    
    def start(self):
        self.running = True
        self.sender_thread = threading.Thread(target=self._send_metrics_loop)
        self.sender_thread.daemon = True
        self.sender_thread.start()
        logger.info(f"Metrics sender started with interval {self.send_interval}s")
    
    def stop(self):
        self.running = False
        if self.sender_thread:
            self.sender_thread.join(timeout=5)
        logger.info("Metrics sender stopped")
    
    def _send_metrics_loop(self):
        while self.running:
            try:
                self._send_pending_metrics()
            except Exception as e:
                logger.error(f"Error sending metrics: {e}")
            time.sleep(self.send_interval)
    
    def _send_pending_metrics(self):
        metrics_batch = []
        
        # 收集队列中所有可用的指标
        while not self.collector.metrics_queue.empty():
            try:
                metrics = self.collector.metrics_queue.get(block=False)
                metrics_batch.append(metrics)
                self.collector.metrics_queue.task_done()
            except queue.Empty:
                break
        
        if not metrics_batch:
            return
        
        # 发送批量指标到监控服务器
        try:
            response = requests.post(
                self.endpoint_url,
                headers={"Content-Type": "application/json"},
                json=metrics_batch,
                timeout=10
            )
            
            if response.status_code == 200:
                logger.info(f"Successfully sent {len(metrics_batch)} metrics")
            else:
                logger.error(f"Failed to send metrics: HTTP {response.status_code}")
        except requests.exceptions.RequestException as e:
            # 如果发送失败，将指标放回队列（简化处理）
            logger.error(f"Network error sending metrics: {e}")
            for metrics in metrics_batch:
                try:
                    self.collector.metrics_queue.put(metrics, block=False)
                except queue.Full:
                    logger.error("Queue is full, dropping metrics")

# 使用示例
if __name__ == "__main__":
    # 创建指标收集器
    collector = MetricsCollector(collection_interval=10)
    collector.start()
    
    # 创建指标发送器
    sender = MetricsSender(
        collector,
        endpoint_url="http://your-monitoring-server:8080/metrics",
        send_interval=30
    )
    sender.start()
    
    try:
        # 主线程保持运行
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        print("Stopping monitoring agent...")
        sender.stop()
        collector.stop()
        print("Monitoring agent stopped")

# 请求合并示例 - 简化版
import asyncio
import time
from typing import List, Dict, Any

class RequestBatcher:
    def __init__(self, max_batch_size=32, max_wait_time=0.05):
        self.max_batch_size = max_batch_size
        self.max_wait_time = max_wait_time  # 最大等待时间，单位秒
        self.batch_queue = []
        self.requests = {}  # 存储请求ID和Future的映射
        self.lock = asyncio.Lock()
        self.processing = False
    
    async def add_request(self, request_id: str, prompt: str, params: Dict[str, Any]):
        # 创建Future用于异步等待结果
        future = asyncio.Future()
        request = {
            'id': request_id,
            'prompt': prompt,
            'params': params,
            'future': future
        }
        
        async with self.lock:
            self.batch_queue.append(request)
            self.requests[request_id] = future
            
            # 如果队列达到最大批次大小，立即处理
            if len(self.batch_queue) >= self.max_batch_size and not self.processing:
                asyncio.create_task(self._process_batch())
            elif len(self.batch_queue) == 1 and not self.processing:
                # 队列为空时添加第一个请求，启动定时器
                asyncio.create_task(self._process_batch_with_timeout())
        
        # 等待处理完成
        try:
            result = await future
            return result
        except Exception as e:
            # 清理状态并重新抛出异常
            async with self.lock:
                if request_id in self.requests:
                    del self.requests[request_id]
            raise
    
    async def _process_batch_with_timeout(self):
        # 等待一段时间，收集更多请求
        await asyncio.sleep(self.max_wait_time)
        
        async with self.lock:
            # 如果队列不为空且没有在处理中，则处理批次
            if self.batch_queue and not self.processing:
                asyncio.create_task(self._process_batch())
    
    async def _process_batch(self):
        async with self.lock:
            if not self.batch_queue or self.processing:
                return
            self.processing = True
            
            # 取出当前批次的请求
            current_batch = self.batch_queue[:self.max_batch_size]
            self.batch_queue = self.batch_queue[self.max_batch_size:]
        
        try:
            # 批量处理请求
            results = await self._execute_batch(current_batch)
            
            # 设置结果
            for request, result in results.items():
                if not request['future'].done():
                    request['future'].set_result(result)
                if request['id'] in self.requests:
                    del self.requests[request['id']]
        except Exception as e:
            # 处理错误
            for request in current_batch:
                if not request['future'].done():
                    request['future'].set_exception(e)
                if request['id'] in self.requests:
                    del self.requests[request['id']]
        finally:
            async with self.lock:
                self.processing = False
                
                # 检查是否还有请求需要处理
                if self.batch_queue:
                    asyncio.create_task(self._process_batch())
    
    async def _execute_batch(self, batch: List[Dict]):
        # 在实际应用中，这里会调用大模型API进行批量推理
        # 这里是模拟实现
        print(f"Processing batch of size {len(batch)}")
        
        # 模拟批量处理时间
        await asyncio.sleep(0.2)  # 模拟处理延迟
        
        results = {}
        for request in batch:
            # 模拟生成结果
            results[request] = {
                'id': request['id'],
                'response': f"Response for '{request['prompt'][:20]}...'",
                'timestamp': time.time()
            }
        
        return results

# 使用示例
async def main():
    batcher = RequestBatcher(max_batch_size=5, max_wait_time=0.1)
    
    # 模拟多个并发请求
    tasks = []
    for i in range(10):
        tasks.append(
            batcher.add_request(
                f"req_{i}", 
                f"This is prompt {i}", 
                {"max_tokens": 100, "temperature": 0.7}
            )
        )
    
    # 等待所有请求完成
    results = await asyncio.gather(*tasks)
    for result in results:
        print(f"Got result: {result['id']} - {result['response']}")

if __name__ == "__main__":
    asyncio.run(main())