111_微调模型评估与部署：从指标到生产

评估指标
├── 准确性指标 (Precision, Recall, F1, Accuracy, BERTScore)
├── 生成质量指标 (BLEU, ROUGE, METEOR, Perplexity)
└── 效率指标 (延迟, 吞吐量, 内存, 能耗)

安全性与对齐评估
├── 安全性评估
│   ├── 有害输出检测
│   ├── 隐私保护
│   └── 对抗鲁棒性
├── 对齐评估
│   ├── 人类价值观对齐
│   ├── 指令遵循能力
│   └── 多维度一致性
└── 评估工具
    ├── HELM
    ├── TruthfulQA
    ├── MMLU
    └── 人工评估面板

输入 → 预处理 → 模型推理 → 结果分析 → 鲁棒性评分
  ↓                 ↓
扰动生成          对抗检测

评估框架
├── 性能维度
│   ├── 准确性
│   ├── 效率
│   └── 生成质量
├── 安全维度
│   ├── 有害输出
│   ├── 隐私保护
│   └── 对抗防御
├── 对齐维度
│   ├── 价值观一致性
│   ├── 指令遵循
│   └── 伦理原则
└── 实用维度
    ├── 可部署性
    ├── 维护成本
    └── 用户体验

数据收集 → 数据清洗 → 质量检查 → 数据组织 → 版本管理 → 持续更新

def preprocessing_pipeline(text):
    # 1. 文本规范化
    text = normalize_text(text)
    # 2. 特殊字符处理
    text = remove_special_chars(text)
    # 3. 格式标准化
    text = standardize_format(text)
    # 4. 分词处理（如果需要）
    if need_tokenization:
        tokens = tokenize_text(text)
        return tokens
    return text

触发 → 数据加载 → 模型推理 → 结果收集 → 指标计算 → 报告生成 → 通知

from jinja2 import Template
import matplotlib.pyplot as plt
import pandas as pd
import json

def generate_evaluation_report(eval_results, template_path):
    # 加载报告模板
    with open(template_path, 'r') as f:
        template = Template(f.read())
    
    # 生成图表
    metrics = eval_results['metrics']
    plt.figure(figsize=(10, 6))
    plt.bar(metrics.keys(), metrics.values())
    plt.title('Performance Metrics')
    plt.savefig('metrics_chart.png')
    
    # 准备报告数据
    report_data = {
        'summary': eval_results['summary'],
        'metrics': metrics,
        'comparison': eval_results['comparison'],
        'issues': eval_results['issues'],
        'recommendations': eval_results['recommendations'],
        'chart_path': 'metrics_chart.png'
    }
    
    # 生成报告
    report_html = template.render(**report_data)
    
    # 保存报告
    with open('evaluation_report.html', 'w') as f:
        f.write(report_html)
    
    return 'evaluation_report.html'

持续评估框架
├── 评估类型
│   ├── 轻量级评估 (快速检查)
│   ├── 标准评估 (常规评估)
│   ├── 深度评估 (全面评估)
│   └── 专项评估 (针对性评估)
├── 触发机制
│   ├── 定期触发
│   ├── 事件触发
│   ├── 手动触发
│   └── 条件触发
└── 资源管理
    ├── 智能调度
    ├── 资源优化
    └── 优先级管理

import dash
import dash_core_components as dcc
import dash_html_components as html
import plotly.express as px
import pandas as pd

# 创建Dash应用
app = dash.Dash(__name__)

# 加载评估数据
eval_results = pd.read_csv('evaluation_results.csv')

# 创建图表
performance_chart = px.bar(eval_results, x='model_version', y='accuracy',
                          title='模型性能趋势')

radar_chart = px.line_polar(eval_results, r=['accuracy', 'f1', 'precision', 'recall'],
                           theta=['准确率', 'F1分数', '精确率', '召回率'],
                           title='多维度性能指标')

# 定义应用布局
app.layout = html.Div([
    html.H1('模型评估仪表盘'),
    html.Div([
        dcc.Graph(figure=performance_chart),
        dcc.Graph(figure=radar_chart)
    ]),
    # 更多图表和控件
])

# 启动应用
if __name__ == '__main__':
    app.run_server(debug=True)

模型 → 校准数据 → 量化参数计算 → 权重量化 → 激活量化 → 部署量化模型

# PyTorch动态量化示例
import torch
from transformers import AutoModelForCausalLM, AutoTokenizer

# 加载模型
model_name = "gpt2"
model = AutoModelForCausalLM.from_pretrained(model_name)
tokenizer = AutoTokenizer.from_pretrained(model_name)

# 动态量化
quantized_model = torch.quantization.quantize_dynamic(
    model,
    {torch.nn.Linear},
    dtype=torch.qint8
)

# 保存量化后的模型
torch.save(quantized_model.state_dict(), "quantized_model.bin")

# 模型大小对比
original_size = sum(p.numel() * p.element_size() for p in model.parameters()) / 1024**2
quantized_size = sum(p.numel() * p.element_size() for p in quantized_model.parameters()) / 1024**2
print(f"原始模型大小: {original_size:.2f} MB")
print(f"量化模型大小: {quantized_size:.2f} MB")
print(f"压缩率: {original_size/quantized_size:.2f}x")

初始化模型 → 训练 → 评估重要性 → 剪枝 → 微调 → 重复剪枝和微调 → 最终模型

import torch
import torch.nn.utils.prune as prune
from transformers import AutoModelForCausalLM

# 加载模型
model = AutoModelForCausalLM.from_pretrained("gpt2")

# 对模型的线性层应用L1范数剪枝
for name, module in model.named_modules():
    if isinstance(module, torch.nn.Linear):
        # 剪枝40%的权重
        prune.ln_structured(module, name="weight", amount=0.4,
                           n=1, dim=0)

# 永久移除剪枝的权重
for name, module in model.named_modules():
    if isinstance(module, torch.nn.Linear):
        prune.remove(module, "weight")

# 保存剪枝后的模型
torch.save(model.state_dict(), "pruned_model.bin")

教师模型 → 生成软标签 → 学生模型训练 → 结合软/硬标签 → 优化学生模型 → 部署

import torch
import torch.nn as nn
from transformers import AutoModelForCausalLM, AutoTokenizer

# 加载教师和学生模型
teacher_model = AutoModelForCausalLM.from_pretrained("gpt2-large")
student_model = AutoModelForCausalLM.from_pretrained("gpt2-medium")
tokenizer = AutoTokenizer.from_pretrained("gpt2-large")

# 蒸馏配置
temperature = 2.0
alpha = 0.5  # 软标签损失权重

# 蒸馏损失函数
def distillation_loss(student_logits, teacher_logits, labels, temperature, alpha):
    # 软标签损失
    soft_loss = nn.KLDivLoss(reduction='batchmean')(
        nn.functional.log_softmax(student_logits/temperature, dim=-1),
        nn.functional.softmax(teacher_logits/temperature, dim=-1)
    ) * (temperature * temperature * 2.0 * alpha)
    
    # 硬标签损失
    hard_loss = nn.functional.cross_entropy(student_logits, labels) * (1. - alpha)
    
    return soft_loss + hard_loss

# 蒸馏训练循环示例
def distillation_train(teacher_model, student_model, dataloader, optimizer, temperature, alpha, epochs):
    teacher_model.eval()
    student_model.train()
    
    for epoch in range(epochs):
        for batch in dataloader:
            inputs = batch['input_ids'].to(device)
            attention_mask = batch['attention_mask'].to(device)
            labels = batch['labels'].to(device)
            
            # 教师模型推理（无梯度）
            with torch.no_grad():
                teacher_outputs = teacher_model(input_ids=inputs, attention_mask=attention_mask)
                teacher_logits = teacher_outputs.logits
            
            # 学生模型推理
            student_outputs = student_model(input_ids=inputs, attention_mask=attention_mask)
            student_logits = student_outputs.logits
            
            # 计算蒸馏损失
            loss = distillation_loss(student_logits, teacher_logits, labels, temperature, alpha)
            
            # 反向传播和优化
            optimizer.zero_grad()
            loss.backward()
            optimizer.step()

from peft import PeftModel
from transformers import AutoModelForCausalLM, AutoTokenizer

# 加载基础模型和LoRA适配器
base_model = AutoModelForCausalLM.from_pretrained("meta-llama/Llama-2-7b-hf")
tokenizer = AutoTokenizer.from_pretrained("meta-llama/Llama-2-7b-hf")
lora_model = PeftModel.from_pretrained(base_model, "path/to/lora/adapter")

# 合并LoRA权重到基础模型
merged_model = lora_model.merge_and_unload()

# 保存合并后的模型
merged_model.save_pretrained("path/to/merged/model")
tokenizer.save_pretrained("path/to/merged/model")

# 推理使用示例
inputs = tokenizer("你好，请介绍一下自己。", return_tensors="pt")
outputs = merged_model.generate(**inputs, max_length=100)
print(tokenizer.decode(outputs[0], skip_special_tokens=True))

准备测试环境 → 基准测试 → 负载测试 → 端到端测试 → 用户体验测试 → 综合评估报告

在线部署: 实时响应 → 低延迟要求 → 高资源消耗
离线部署: 批处理   → 高吞吐量   → 资源利用率高
混合部署: 平衡策略 → 灵活应对   → 复杂但高效

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from transformers import AutoModelForCausalLM, AutoTokenizer
import torch

# 创建FastAPI应用
app = FastAPI(title="LLM微调模型服务")

# 定义请求和响应模型
class PredictRequest(BaseModel):
    text: str
    max_length: int = 100
    temperature: float = 0.7
    top_p: float = 0.95

class PredictResponse(BaseModel):
    generated_text: str
    model_id: str
    processing_time: float

# 加载模型
model_cache = {}

def load_model(model_id):
    if model_id not in model_cache:
        tokenizer = AutoTokenizer.from_pretrained(model_id)
        model = AutoModelForCausalLM.from_pretrained(
            model_id,
            torch_dtype=torch.float16,
            device_map="auto"
        )
        model_cache[model_id] = (tokenizer, model)
    return model_cache[model_id]

# 健康检查端点
@app.get("/health")
async def health_check():
    return {"status": "healthy", "models": list(model_cache.keys())}

# 模型列表端点
@app.get("/models")
async def list_models():
    return {"models": ["model_1", "model_2", "model_3"]}

# 预测端点
@app.post("/models/{model_id}/predict", response_model=PredictResponse)
async def predict(model_id: str, request: PredictRequest):
    import time
    start_time = time.time()
    
    try:
        # 加载模型
        tokenizer, model = load_model(model_id)
        
        # 执行推理
        inputs = tokenizer(request.text, return_tensors="pt").to(model.device)
        with torch.no_grad():
            outputs = model.generate(
                **inputs,
                max_length=request.max_length,
                temperature=request.temperature,
                top_p=request.top_p,
                do_sample=True
            )
        
        # 处理结果
        generated_text = tokenizer.decode(outputs[0], skip_special_tokens=True)
        
        # 计算处理时间
        processing_time = time.time() - start_time
        
        return PredictResponse(
            generated_text=generated_text,
            model_id=model_id,
            processing_time=processing_time
        )
    
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

# 启动服务
if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

# 首先创建model_service.proto文件定义服务
"""
syntax = "proto3";

package model_service;

service LLMModelService {
  rpc HealthCheck(HealthCheckRequest) returns (HealthCheckResponse);
  rpc ListModels(ListModelsRequest) returns (ListModelsResponse);
  rpc Predict(PredictRequest) returns (PredictResponse);
  rpc StreamingPredict(stream PredictRequest) returns (stream PredictResponse);
}

message HealthCheckRequest {}

message HealthCheckResponse {
  string status = 1;
  repeated string models = 2;
}

message ListModelsRequest {}

message ModelInfo {
  string id = 1;
  string name = 2;
  string version = 3;
}

message ListModelsResponse {
  repeated ModelInfo models = 1;
}

message PredictRequest {
  string model_id = 1;
  string text = 2;
  int32 max_length = 3;
  float temperature = 4;
  float top_p = 5;
}

message PredictResponse {
  string generated_text = 1;
  string model_id = 2;
  float processing_time = 3;
}
"""

# 使用protoc编译生成Python代码
# protoc -I=. --python_out=. --grpc_python_out=. model_service.proto

# 实现gRPC服务器
import grpc
from concurrent import futures
import time
import torch
from transformers import AutoModelForCausalLM, AutoTokenizer

# 导入生成的模块
import model_service_pb2
import model_service_pb2_grpc

# 模型缓存
model_cache = {}

def load_model(model_id):
    if model_id not in model_cache:
        tokenizer = AutoTokenizer.from_pretrained(model_id)
        model = AutoModelForCausalLM.from_pretrained(
            model_id,
            torch_dtype=torch.float16,
            device_map="auto"
        )
        model_cache[model_id] = (tokenizer, model)
    return model_cache[model_id]

# 实现服务类
class LLMModelServiceServicer(model_service_pb2_grpc.LLMModelServiceServicer):
    def HealthCheck(self, request, context):
        return model_service_pb2.HealthCheckResponse(
            status="healthy",
            models=list(model_cache.keys())
        )
    
    def ListModels(self, request, context):
        models = [
            model_service_pb2.ModelInfo(id="model_1", name="微调模型1", version="1.0.0"),
            model_service_pb2.ModelInfo(id="model_2", name="微调模型2", version="1.1.0")
        ]
        return model_service_pb2.ListModelsResponse(models=models)
    
    def Predict(self, request, context):
        start_time = time.time()
        
        try:
            # 加载模型
            tokenizer, model = load_model(request.model_id)
            
            # 执行推理
            inputs = tokenizer(request.text, return_tensors="pt").to(model.device)
            with torch.no_grad():
                outputs = model.generate(
                    **inputs,
                    max_length=request.max_length,
                    temperature=request.temperature,
                    top_p=request.top_p,
                    do_sample=True
                )
            
            # 处理结果
            generated_text = tokenizer.decode(outputs[0], skip_special_tokens=True)
            
            # 计算处理时间
            processing_time = time.time() - start_time
            
            return model_service_pb2.PredictResponse(
                generated_text=generated_text,
                model_id=request.model_id,
                processing_time=processing_time
            )
        
        except Exception as e:
            context.set_code(grpc.StatusCode.INTERNAL)
            context.set_details(str(e))
            return model_service_pb2.PredictResponse()
    
    def StreamingPredict(self, request_iterator, context):
        for request in request_iterator:
            # 处理每个流式请求
            response = self.Predict(request, context)
            if context.code() != grpc.StatusCode.OK:
                break
            yield response

# 启动服务器
def serve():
    server = grpc.server(futures.ThreadPoolExecutor(max_workers=10))
    model_service_pb2_grpc.add_LLMModelServiceServicer_to_server(
        LLMModelServiceServicer(), server
    )
    server.add_insecure_port('[::]:50051')
    server.start()
    print("服务器运行在端口50051")
    server.wait_for_termination()

if __name__ == '__main__':
    serve()

# 使用多阶段构建优化镜像大小

# 第一阶段：构建环境
FROM python:3.10-slim AS builder

# 设置工作目录
WORKDIR /app

# 安装系统依赖
RUN apt-get update && apt-get install -y --no-install-recommends \
    gcc \
    g++ \
    && rm -rf /var/lib/apt/lists/*

# 创建Python虚拟环境
RUN python -m venv /opt/venv
ENV PATH="/opt/venv/bin:$PATH"

# 安装Python依赖
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

# 第二阶段：运行环境
FROM python:3.10-slim

# 设置工作目录
WORKDIR /app

# 从构建阶段复制虚拟环境
COPY --from=builder /opt/venv /opt/venv
ENV PATH="/opt/venv/bin:$PATH"

# 安装必要的系统依赖
RUN apt-get update && apt-get install -y --no-install-recommends \
    libgomp1 \
    && rm -rf /var/lib/apt/lists/*

# 复制模型服务代码
COPY app/ .

# 复制模型文件（可选，也可以通过卷挂载或下载）
# COPY models/ /app/models/

# 设置环境变量
ENV MODEL_CACHE_DIR=/app/models
ENV PORT=8000

# 暴露端口
EXPOSE 8000

# 启动命令
CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]

version: '3.8'

services:
  llm-service:
    build: .
    ports:
      - "8000:8000"
    volumes:
      - ./models:/app/models
      - ./logs:/app/logs
    environment:
      - MODEL_CACHE_DIR=/app/models
      - LOG_LEVEL=INFO
      - MAX_WORKERS=4
    deploy:
      resources:
        limits:
          cpus: '4'
          memory: 16G

  api-gateway:
    image: nginx:latest
    ports:
      - "80:80"
    volumes:
      - ./nginx/nginx.conf:/etc/nginx/nginx.conf:ro
    depends_on:
      - llm-service

apiVersion: apps/v1
kind: Deployment
metadata:
  name: llm-service
  namespace: models
spec:
  replicas: 3
  selector:
    matchLabels:
      app: llm-service
  template:
    metadata:
      labels:
        app: llm-service
    spec:
      containers:
      - name: llm-service
        image: llm-service:latest
        ports:
        - containerPort: 8000
        resources:
          requests:
            cpu: "1"
            memory: "4Gi"
          limits:
            cpu: "2"
            memory: "8Gi"
        env:
        - name: MODEL_CACHE_DIR
          value: /app/models
        - name: LOG_LEVEL
          valueFrom:
            configMapKeyRef:
              name: llm-config
              key: log_level
        volumeMounts:
        - name: model-storage
          mountPath: /app/models
        - name: log-storage
          mountPath: /app/logs
      volumes:
      - name: model-storage
        persistentVolumeClaim:
          claimName: model-pvc
      - name: log-storage
        emptyDir: {}

apiVersion: autoscaling/v2
kind: HorizontalPodAutoscaler
metadata:
  name: llm-service-hpa
  namespace: models
spec:
  scaleTargetRef:
    apiVersion: apps/v1
    kind: Deployment
    name: llm-service
  minReplicas: 2
  maxReplicas: 10
  metrics:
  - type: Resource
    resource:
      name: cpu
      target:
        type: Utilization
        averageUtilization: 70
  - type: Resource
    resource:
      name: memory
      target:
        type: Utilization
        averageUtilization: 80

livenessProbe:
  httpGet:
    path: /health
    port: 8000
  initialDelaySeconds: 30
  periodSeconds: 10
  timeoutSeconds: 5
  failureThreshold: 3
readinessProbe:
  httpGet:
    path: /health
    port: 8000
  initialDelaySeconds: 10
  periodSeconds: 5

用户请求 → 边缘节点(简单推理) → 复杂请求 → 云服务器(复杂推理) → 统一响应 → 用户
                ↓                            ↑
              本地缓存                  模型更新
                ↓                            ↑
             数据收集 ←----------------- 模型训练

用户输入 → 预处理模型 → 主模型 → 后处理模型 → 结果输出
             ↓            ↓           ↓
          缓存层 ←--------→ 配置管理 ←----→ 监控系统

监控指标结构
├── 推理性能
│   ├── 延迟 (P50/P95/P99)
│   ├── 吞吐量
│   └── 并发数
├── 模型质量
│   ├── 准确性
│   ├── 漂移指标
│   └── 置信度
└── 系统健康
    ├── 可用性
    ├── 资源使用
    └── 依赖服务

# 使用Prometheus监控模型服务
from fastapi import FastAPI, BackgroundTasks
from prometheus_client import Counter, Histogram, Gauge, start_http_server
import time
import random

# 创建监控指标
# 请求计数器
REQUEST_COUNT = Counter('llm_requests_total', 'Total number of requests', ['model_id', 'endpoint', 'status'])
# 处理时间直方图
REQUEST_LATENCY = Histogram('llm_request_latency_seconds', 'Request latency in seconds', ['model_id', 'endpoint'])
# 活跃请求数
ACTIVE_REQUESTS = Gauge('llm_active_requests', 'Number of active requests', ['model_id'])
# 模型准确率（模拟）
MODEL_ACCURACY = Gauge('llm_model_accuracy', 'Model accuracy', ['model_id'])
# 资源使用情况
MEMORY_USAGE = Gauge('llm_memory_usage_bytes', 'Memory usage in bytes')

# 创建FastAPI应用
app = FastAPI(title="监控示例")

# 模拟模型预测函数
def predict(model_id, text):
    # 模拟模型推理时间
    processing_time = 0.1 + random.random() * 0.5
    time.sleep(processing_time)
    return f"Generated text for model {model_id}: {text}"

# 预测端点
@app.post("/models/{model_id}/predict")
async def predict_endpoint(model_id: str, text: str, background_tasks: BackgroundTasks):
    # 增加活跃请求数
    ACTIVE_REQUESTS.labels(model_id=model_id).inc()
    
    # 记录请求开始时间
    start_time = time.time()
    status = 'success'
    
    try:
        # 执行预测
        result = predict(model_id, text)
        
        # 随机模拟一些错误
        if random.random() < 0.05:
            raise Exception("Simulated error")
            
    except Exception as e:
        status = 'error'
        result = str(e)
    finally:
        # 计算处理时间
        processing_time = time.time() - start_time
        
        # 记录指标
        REQUEST_COUNT.labels(model_id=model_id, endpoint="predict", status=status).inc()
        REQUEST_LATENCY.labels(model_id=model_id, endpoint="predict").observe(processing_time)
        ACTIVE_REQUESTS.labels(model_id=model_id).dec()
        
        # 更新模型准确率（模拟）
        MODEL_ACCURACY.labels(model_id=model_id).set(0.9 + random.random() * 0.05)
        
        # 异步更新资源使用情况
        background_tasks.add_task(update_memory_usage)
    
    return {"result": result, "processing_time": processing_time}

# 更新内存使用情况
def update_memory_usage():
    # 模拟内存使用
    memory_usage = 100 * 1024 * 1024 + random.randint(0, 50 * 1024 * 1024)
    MEMORY_USAGE.set(memory_usage)

# 启动监控服务器
def start_monitoring_server(port=8001):
    start_http_server(port)
    print(f"Monitoring server running on port {port}")

# 启动时初始化监控服务器
@app.on_event("startup")
async def startup_event():
    start_monitoring_server()

模型漂移
├── 数据漂移
│   ├── 协变量漂移 (输入分布变)
│   ├── 概念漂移 (关系变化)
│   └── 先验概率漂移 (输出边缘分布变)
└── 性能漂移
    ├── 准确性下降
    ├── 错误类型变化
    └── 业务指标下降

import numpy as np
from scipy import stats
from sklearn.neighbors import KernelDensity
from sklearn.model_selection import train_test_split

class DriftDetector:
    def __init__(self, reference_data, significance_level=0.05):
        self.reference_data = reference_data
        self.significance_level = significance_level
        self.detectors = {
            'ks_test': self._ks_test,
            'psi': self._psi,
            'kl_divergence': self._kl_divergence
        }
    
    def _ks_test(self, current_data):
        """使用KS检验检测漂移"""
        if len(current_data) < 2:
            return 1.0  # 数据太少，不检测
        
        # 对每个特征进行KS检验
        p_values = []
        for i in range(self.reference_data.shape[1]):
            try:
                _, p_value = stats.ks_2samp(
                    self.reference_data[:, i],
                    current_data[:, i]
                )
                p_values.append(p_value)
            except:
                # 如果检验失败，假设没有漂移
                p_values.append(1.0)
        
        # 使用Bonferroni校正
        min_p_value = np.min(p_values)
        return min_p_value * len(p_values)  # 校正后的p值
    
    def _psi(self, current_data, bins=10):
        """计算PSI (Population Stability Index)"""
        psis = []
        
        for i in range(self.reference_data.shape[1]):
            # 获取参考数据的分位数
            ref_quantiles = np.linspace(0, 1, bins+1)[1:-1]
            bins_edges = np.quantile(self.reference_data[:, i], ref_quantiles)
            
            # 计算参考分布和当前分布在各bin中的频率
            ref_counts, _ = np.histogram(self.reference_data[:, i], bins=bins_edges)
            curr_counts, _ = np.histogram(current_data[:, i], bins=bins_edges)
            
            # 避免除零和log(0)
            ref_dist = ref_counts / max(len(self.reference_data), 1)
            curr_dist = curr_counts / max(len(current_data), 1)
            
            # 计算PSI
            epsilon = 1e-10
            psi = np.sum((curr_dist - ref_dist) * np.log((curr_dist + epsilon) / (ref_dist + epsilon)))
            psis.append(psi)
        
        return np.mean(psis)  # 返回平均PSI
    
    def _kl_divergence(self, current_data, bandwidth=0.5):
        """使用核密度估计计算KL散度"""
        klds = []
        
        for i in range(self.reference_data.shape[1]):
            try:
                # 训练核密度估计器
                kde_ref = KernelDensity(kernel='gaussian', bandwidth=bandwidth).fit(
                    self.reference_data[:, i].reshape(-1, 1)
                )
                kde_curr = KernelDensity(kernel='gaussian', bandwidth=bandwidth).fit(
                    current_data[:, i].reshape(-1, 1)
                )
                
                # 生成评估点
                x_min = min(self.reference_data[:, i].min(), current_data[:, i].min())
                x_max = max(self.reference_data[:, i].max(), current_data[:, i].max())
                x = np.linspace(x_min, x_max, 100).reshape(-1, 1)
                
                # 计算密度
                log_dens_ref = kde_ref.score_samples(x)
                log_dens_curr = kde_curr.score_samples(x)
                
                # 计算KL散度
                # KL(P||Q) = ∫ P(x) log(P(x)/Q(x)) dx
                p = np.exp(log_dens_ref)
                q = np.exp(log_dens_curr)
                
                # 避免除零和log(0)
                p = np.maximum(p, 1e-10)
                q = np.maximum(q, 1e-10)
                
                # 使用数值积分计算KL散度
                kl_div = np.sum(p * (np.log(p) - np.log(q))) * (x[1] - x[0])
                klds.append(kl_div)
            except:
                # 如果计算失败，使用默认值
                klds.append(0.0)
        
        return np.mean(klds)  # 返回平均KL散度
    
    def detect_drift(self, current_data, method='ks_test', return_score=False):
        """检测漂移
        
        Args:
            current_data: 当前数据
            method: 使用的检测方法
            return_score: 是否返回得分
            
        Returns:
            如果return_score=True，返回(是否漂移, 得分)
            否则，返回是否漂移
        """
        if method not in self.detectors:
            raise ValueError(f"Unknown method: {method}")
        
        # 计算得分
        score = self.detectors[method](current_data)
        
        # 判断是否漂移
        if method == 'ks_test':
            # KS检验：p值小于显著性水平表示漂移
            is_drift = score < self.significance_level
        else:
            # PSI和KL散度：需要设定阈值
            thresholds = {
                'psi': 0.1,  # PSI > 0.1 表示有明显漂移
                'kl_divergence': 0.5  # KL散度 > 0.5 表示有明显漂移
            }
            is_drift = score > thresholds.get(method, 0.1)
        
        if return_score:
            return is_drift, score
        return is_drift

# 使用示例
def example_usage():
    # 生成参考数据
    np.random.seed(42)
    reference_data = np.random.normal(0, 1, (1000, 5))
    
    # 创建漂移检测器
    detector = DriftDetector(reference_data)
    
    # 1. 无漂移的当前数据
    current_data_no_drift = np.random.normal(0, 1, (200, 5))
    is_drift, score = detector.detect_drift(current_data_no_drift, method='ks_test', return_score=True)
    print(f"无漂移检测 (KS test): 漂移={is_drift}, p值={score:.4f}")
    
    # 2. 有漂移的当前数据
    current_data_with_drift = np.random.normal(1, 1, (200, 5))  # 均值漂移
    is_drift, score = detector.detect_drift(current_data_with_drift, method='ks_test', return_score=True)
    print(f"有漂移检测 (KS test): 漂移={is_drift}, p值={score:.4f}")
    
    # 3. 使用PSI检测
    is_drift, score = detector.detect_drift(current_data_with_drift, method='psi', return_score=True)
    print(f"PSI检测: 漂移={is_drift}, PSI={score:.4f}")
    
    # 4. 使用KL散度检测
    is_drift, score = detector.detect_drift(current_data_with_drift, method='kl_divergence', return_score=True)
    print(f"KL散度检测: 漂移={is_drift}, KL散度={score:.4f}")

定期健康检查 → 日志管理 → 资源管理 → 安全更新 → 备份与恢复 → 报告生成
    ↑                                                              ↓
    └───────────────────────── 异常处理 ───────────────────────────┘

import os
import time
import json
import logging
import datetime
import threading
import numpy as np
from sklearn.metrics import accuracy_score

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

class ModelUpdater:
    def __init__(self, model_path, data_path, config):
        self.model_path = model_path
        self.data_path = data_path
        self.config = config
        self.current_model = None
        self.current_version = None
        self.update_thread = None
        self.stop_event = threading.Event()
        self.last_update_time = None
        
        # 加载配置
        self.update_interval = config.get('update_interval', 86400)  # 默认24小时
        self.performance_threshold = config.get('performance_threshold', 0.85)
        self.min_data_size = config.get('min_data_size', 1000)
        
    def load_model(self, version=None):
        """加载指定版本的模型"""
        if version is None:
            version = self._get_latest_version()
            
        if version is None:
            logger.warning("No model found")
            return None
            
        model_file = os.path.join(self.model_path, f"model_v{version}.pkl")
        try:
            # 这里使用pickle作为示例，实际应用中可能使用其他格式
            import pickle
            with open(model_file, 'rb') as f:
                model = pickle.load(f)
            
            self.current_model = model
            self.current_version = version
            logger.info(f"Loaded model version {version}")
            return model
        except Exception as e:
            logger.error(f"Error loading model version {version}: {e}")
            return None
    
    def _get_latest_version(self):
        """获取最新的模型版本"""
        try:
            versions = []
            for filename in os.listdir(self.model_path):
                if filename.startswith('model_v') and filename.endswith('.pkl'):
                    try:
                        version = int(filename[7:-4])  # 提取版本号
                        versions.append(version)
                    except:
                        continue
            
            if not versions:
                return None
                
            return max(versions)
        except Exception as e:
            logger.error(f"Error getting latest version: {e}")
            return None
    
    def evaluate_model(self, model, test_data):
        """评估模型性能"""
        try:
            X_test, y_test = test_data
            y_pred = model.predict(X_test)
            accuracy = accuracy_score(y_test, y_pred)
            logger.info(f"Model evaluation accuracy: {accuracy:.4f}")
            return accuracy
        except Exception as e:
            logger.error(f"Error evaluating model: {e}")
            return 0.0
    
    def collect_data(self):
        """收集训练数据"""
        try:
            # 在实际应用中，这里会从数据库或文件系统收集数据
            # 这里使用随机数据作为示例
            logger.info("Collecting training data")
            
            # 检查数据量
            data_files = os.listdir(self.data_path)
            if len(data_files) < self.min_data_size:
                logger.warning(f"Not enough data: {len(data_files)} < {self.min_data_size}")
                return None
                
            # 加载数据
            # 这里应该是实际的数据加载代码
            X = np.random.rand(len(data_files), 10)  # 示例特征
            y = np.random.randint(0, 2, len(data_files))  # 示例标签
            
            # 分割训练集和测试集
            from sklearn.model_selection import train_test_split
            X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2)
            
            return {
                'train': (X_train, y_train),
                'test': (X_test, y_test)
            }
        except Exception as e:
            logger.error(f"Error collecting data: {e}")
            return None
    
    def train_model(self, training_data):
        """训练新模型"""
        try:
            logger.info("Starting model training")
            
            X_train, y_train = training_data
            
            # 这里使用简单的分类器作为示例
            # 在实际应用中，这里应该是实际的模型训练代码
            from sklearn.ensemble import RandomForestClassifier
            model = RandomForestClassifier(n_estimators=100, random_state=42)
            model.fit(X_train, y_train)
            
            logger.info("Model training completed")
            return model
        except Exception as e:
            logger.error(f"Error training model: {e}")
            return None
    
    def save_model(self, model):
        """保存新模型"""
        try:
            # 生成新版本号
            new_version = (self.current_version or 0) + 1
            
            # 保存模型文件
            model_file = os.path.join(self.model_path, f"model_v{new_version}.pkl")
            import pickle
            with open(model_file, 'wb') as f:
                pickle.dump(model, f)
            
            # 保存元数据
            metadata = {
                'version': new_version,
                'timestamp': datetime.datetime.now().isoformat(),
                'config': self.config
            }
            metadata_file = os.path.join(self.model_path, f"metadata_v{new_version}.json")
            with open(metadata_file, 'w') as f:
                json.dump(metadata, f, indent=2)
            
            logger.info(f"Saved model version {new_version}")
            return new_version
        except Exception as e:
            logger.error(f"Error saving model: {e}")
            return None
    
    def should_update(self):
        """检查是否应该更新模型"""
        # 检查时间间隔
        if self.last_update_time:
            time_since_update = time.time() - self.last_update_time
            if time_since_update < self.update_interval:
                logger.info(f"Time since last update: {time_since_update:.0f}s < {self.update_interval}s")
                return False
        
        # 检查性能
        if self.current_model:
            # 加载测试数据
            # 这里应该是实际的测试数据加载代码
            test_data = (np.random.rand(100, 10), np.random.randint(0, 2, 100))
            current_performance = self.evaluate_model(self.current_model, test_data)
            
            if current_performance >= self.performance_threshold:
                logger.info(f"Current performance ({current_performance:.4f}) >= threshold ({self.performance_threshold})")
                return False
        
        return True
    
    def update_model(self):
        """执行模型更新"""
        logger.info("Starting model update process")
        
        try:
            # 检查是否应该更新
            if not self.should_update():
                logger.info("No update needed")
                return False
            
            # 收集数据
            data = self.collect_data()
            if data is None:
                return False
            
            # 训练模型
            new_model = self.train_model(data['train'])
            if new_model is None:
                return False
            
            # 评估模型
            new_performance = self.evaluate_model(new_model, data['test'])
            
            # 与当前模型比较
            if self.current_model:
                current_performance = self.evaluate_model(self.current_model, data['test'])
                if new_performance <= current_performance:
                    logger.warning(f"New model performance ({new_performance:.4f}) <= current model ({current_performance:.4f}), skipping update")
                    return False
            
            # 保存新模型
            new_version = self.save_model(new_model)
            if new_version is None:
                return False
            
            # 加载新模型
            self.load_model(new_version)
            
            # 更新时间
            self.last_update_time = time.time()
            
            logger.info(f"Model updated successfully to version {new_version}")
            return True
            
        except Exception as e:
            logger.error(f"Error in update process: {e}")
            return False
    
    def start_auto_update(self):
        """启动自动更新线程"""
        if self.update_thread and self.update_thread.is_alive():
            logger.warning("Auto update thread already running")
            return
        
        self.stop_event.clear()
        self.update_thread = threading.Thread(target=self._auto_update_loop)
        self.update_thread.daemon = True
        self.update_thread.start()
        logger.info("Auto update thread started")
    
    def stop_auto_update(self):
        """停止自动更新线程"""
        if self.update_thread and self.update_thread.is_alive():
            self.stop_event.set()
            self.update_thread.join(timeout=5)
            logger.info("Auto update thread stopped")
    
    def _auto_update_loop(self):
        """自动更新循环"""
        while not self.stop_event.is_set():
            try:
                self.update_model()
            except Exception as e:
                logger.error(f"Error in auto update loop: {e}")
            
            # 等待下一次更新
            self.stop_event.wait(self.update_interval)

# 使用示例
def example_usage():
    # 配置
    config = {
        'update_interval': 3600,  # 1小时
        'performance_threshold': 0.85,
        'min_data_size': 1000
    }
    
    # 创建模型更新器
    updater = ModelUpdater(
        model_path="./models",
        data_path="./data",
        config=config
    )
    
    # 确保目录存在
    os.makedirs("./models", exist_ok=True)
    os.makedirs("./data", exist_ok=True)
    
    # 启动自动更新
    updater.start_auto_update()
    
    # 主线程继续运行
    try:
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        updater.stop_auto_update()
        print("Stopped")

发现问题 → 收集信息 → 分析问题 → 定位原因 → 修复问题 → 验证解决 → 记录总结
    ↑                                                                  ↓
    └─────────────────────────── 监控告警 ─────────────────────────────┘

# 使用OpenTelemetry实现分布式追踪
from opentelemetry import trace
from opentelemetry.sdk.resources import SERVICE_NAME, Resource
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.exporter.jaeger.thrift import JaegerExporter
from opentelemetry.instrumentation.fastapi import FastAPIInstrumentor
from fastapi import FastAPI, Depends, HTTPException
import requests
import os

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

provider = TracerProvider(resource=resource)
trace.set_tracer_provider(provider)

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

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

# 创建FastAPI应用
app = FastAPI(title="带追踪的LLM服务")

# 自动检测FastAPI应用
FastAPIInstrumentor.instrument_app(app)

# 获取追踪器
tracer = trace.get_tracer(__name__)

# 模拟模型推理函数
def model_inference(text, model_id):
    with tracer.start_as_current_span("model_inference") as span:
        # 添加属性
        span.set_attribute("model.id", model_id)
        span.set_attribute("input.length", len(text))
        
        # 模拟推理过程
        import time
        time.sleep(0.2)  # 模拟计算时间
        
        # 调用预处理服务
        with tracer.start_as_current_span("preprocessing_call"):
            try:
                # 这里应该是实际的服务调用
                # 为了演示，我们只是模拟一个HTTP请求
                # response = requests.post("http://preprocess-service/process", json={"text": text})
                # preprocessed_text = response.json()["result"]
                preprocessed_text = text.lower()
                time.sleep(0.1)  # 模拟网络延迟
                span.set_attribute("preprocessing.success", True)
            except Exception as e:
                span.set_attribute("preprocessing.success", False)
                span.record_exception(e)
                raise HTTPException(status_code=500, detail="预处理服务错误")
        
        # 模拟模型计算
        with tracer.start_as_current_span("model_computation"):
            try:
                # 模拟模型计算
                time.sleep(0.5)  # 模拟计算时间
                result = f"Processed by {model_id}: {preprocessed_text}"
                span.set_attribute("output.length", len(result))
                return result
            except Exception as e:
                span.record_exception(e)
                raise HTTPException(status_code=500, detail="模型计算错误")

# 健康检查端点
@app.get("/health")
async def health_check():
    return {"status": "healthy"}

# 模型预测端点
@app.post("/models/{model_id}/predict")
async def predict(model_id: str, text: str):
    # 使用当前活动的span（由FastAPIInstrumentor创建）
    with tracer.start_as_current_span("predict_endpoint") as span:
        span.set_attribute("endpoint", "predict")
        span.set_attribute("model.id", model_id)
        
        try:
            # 验证输入
            if not text or not isinstance(text, str):
                raise HTTPException(status_code=400, detail="无效的输入文本")
            
            # 执行模型推理
            result = model_inference(text, model_id)
            
            # 添加结果属性
            span.set_attribute("prediction.success", True)
            
            return {
                "model_id": model_id,
                "input": text,
                "output": result
            }
            
        except HTTPException:
            span.set_attribute("prediction.success", False)
            raise
        except Exception as e:
            span.set_attribute("prediction.success", False)
            span.record_exception(e)
            raise HTTPException(status_code=500, detail="预测过程中发生错误")

# 启动事件
@app.on_event("startup")
async def startup_event():
    print("服务启动，追踪已配置")

# 关闭事件
@app.on_event("shutdown")
async def shutdown_event():
    # 关闭追踪提供商
    trace.get_tracer_provider().shutdown()
    print("服务关闭，追踪已停止")

import numpy as np

class DifferentialPrivacy:
    def __init__(self, epsilon=1.0, delta=1e-5):
        """
        初始化差分隐私模块
        
        Args:
            epsilon: 隐私预算，较小的值提供更强的隐私保护，但可能降低准确性
            delta: 松弛参数，用于近似差分隐私
        """
        self.epsilon = epsilon
        self.delta = delta
    
    def add_laplace_noise(self, value, sensitivity, epsilon=None):
        """
        添加拉普拉斯噪声实现差分隐私
        
        Args:
            value: 原始值
            sensitivity: 函数的敏感度（最大变化量）
            epsilon: 特定查询的隐私预算，默认为类初始化时的值
            
        Returns:
            添加噪声后的值
        """
        if epsilon is None:
            epsilon = self.epsilon
        
        # 计算噪声参数
        scale = sensitivity / epsilon
        
        # 从拉普拉斯分布生成噪声
        noise = np.random.laplace(0, scale)
        
        # 返回添加噪声后的值
        return value + noise
    
    def add_gaussian_noise(self, value, sensitivity, epsilon=None, delta=None):
        """
        添加高斯噪声实现差分隐私
        
        Args:
            value: 原始值
            sensitivity: 函数的敏感度（最大变化量）
            epsilon: 特定查询的隐私预算，默认为类初始化时的值
            delta: 特定查询的松弛参数，默认为类初始化时的值
            
        Returns:
            添加噪声后的值
        """
        if epsilon is None:
            epsilon = self.epsilon
        if delta is None:
            delta = self.delta
        
        # 计算噪声参数
        # 高斯机制的标准差
        sigma = sensitivity * np.sqrt(2 * np.log(1.25 / delta)) / epsilon
        
        # 从高斯分布生成噪声
        noise = np.random.normal(0, sigma)
        
        # 返回添加噪声后的值
        return value + noise
    
    def private_mean(self, data, epsilon=None):
        """
        计算带差分隐私保护的均值
        
        Args:
            data: 数据数组
            epsilon: 隐私预算
            
        Returns:
            带噪声的均值
        """
        # 计算真实均值
        true_mean = np.mean(data)
        
        # 计算敏感度：添加或删除一个数据点时均值的最大变化
        # 假设数据范围在[min_val, max_val]之间
        min_val = np.min(data)
        max_val = np.max(data)
        sensitivity = (max_val - min_val) / len(data)
        
        # 添加拉普拉斯噪声
        return self.add_laplace_noise(true_mean, sensitivity, epsilon)
    
    def private_count(self, data, condition, epsilon=None):
        """
        计算带差分隐私保护的计数
        
        Args:
            data: 数据数组
            condition: 计数条件（函数）
            epsilon: 隐私预算
            
        Returns:
            带噪声的计数
        """
        # 计算真实计数
        true_count = sum(condition(x) for x in data)
        
        # 计数查询的敏感度为1
        sensitivity = 1
        
        # 添加拉普拉斯噪声
        return self.add_laplace_noise(true_count, sensitivity, epsilon)
    
    def private_histogram(self, data, bins, epsilon=None):
        """
        计算带差分隐私保护的直方图
        
        Args:
            data: 数据数组
            bins: 直方图的分箱数
            epsilon: 隐私预算
            
        Returns:
            带噪声的直方图计数
        """
        if epsilon is None:
            epsilon = self.epsilon
        
        # 为每个分箱分配相等的隐私预算
        epsilon_per_bin = epsilon / bins
        
        # 计算真实直方图
        hist, bin_edges = np.histogram(data, bins=bins)
        
        # 对每个分箱计数添加拉普拉斯噪声
        noisy_hist = np.zeros_like(hist, dtype=float)
        for i in range(len(hist)):
            noisy_hist[i] = self.add_laplace_noise(hist[i], 1, epsilon_per_bin)
        
        # 确保计数为非负
        noisy_hist = np.maximum(0, noisy_hist)
        
        return noisy_hist, bin_edges
    
    def compose_budget(self, epsilon_values):
        """
        组合多个查询的隐私预算
        使用高级组合定理
        
        Args:
            epsilon_values: 各查询的隐私预算列表
            
        Returns:
            组合后的总隐私预算
        """
        # 简单求和（顺序组合）
        return sum(epsilon_values)
    
    def advanced_compose_budget(self, k, epsilon_per_query, delta=None):
        """
        使用高级组合定理计算k个相同查询的总隐私预算
        
        Args:
            k: 查询次数
            epsilon_per_query: 每次查询的隐私预算
            delta: 松弛参数
            
        Returns:
            组合后的总隐私预算
        """
        if delta is None:
            delta = self.delta
        
        # 高级组合定理
        epsilon_total = epsilon_per_query * np.sqrt(2 * k * np.log(1 / delta))
        return epsilon_total

# 使用示例
def example_usage():
    # 创建差分隐私实例
    dp = DifferentialPrivacy(epsilon=1.0)
    
    # 生成示例数据
    np.random.seed(42)
    data = np.random.normal(0, 1, 1000)
    
    # 1. 计算带差分隐私保护的均值
    true_mean = np.mean(data)
    private_mean = dp.private_mean(data)
    print(f"真实均值: {true_mean:.4f}")
    print(f"带噪声均值: {private_mean:.4f}")
    print(f"误差: {abs(private_mean - true_mean):.4f}")
    print()
    
    # 2. 计算带差分隐私保护的计数
    condition = lambda x: x > 0  # 计算大于0的元素数量
    true_count = sum(condition(x) for x in data)
    private_count = dp.private_count(data, condition)
    print(f"真实计数: {true_count}")
    print(f"带噪声计数: {private_count:.1f}")
    print(f"误差: {abs(private_count - true_count):.1f}")
    print()
    
    # 3. 计算带差分隐私保护的直方图
    bins = 10
    true_hist, bin_edges = np.histogram(data, bins=bins)
    private_hist, _ = dp.private_histogram(data, bins)
    
    print("直方图比较:")
    print("分箱范围\t真实计数\t带噪声计数")
    for i in range(bins):
        bin_range = f"[{bin_edges[i]:.2f}, {bin_edges[i+1]:.2f})"
        print(f"{bin_range}\t{true_hist[i]}\t{private_hist[i]:.1f}")
    
    # 4. 演示预算组合
    epsilon_values = [0.1, 0.2, 0.3]
    total_epsilon = dp.compose_budget(epsilon_values)
    print(f"\n预算组合示例:")
    print(f"各查询预算: {epsilon_values}")
    print(f"总预算(顺序组合): {total_epsilon:.3f}")
    
    # 5. 演示高级组合
    k = 10  # 查询次数
    epsilon_per_query = 0.1
    advanced_epsilon = dp.advanced_compose_budget(k, epsilon_per_query)
    print(f"\n高级组合示例:")
    print(f"查询次数: {k}")
    print(f"每次查询预算: {epsilon_per_query}")
    print(f"总预算(高级组合): {advanced_epsilon:.3f}")
    print(f"总预算(顺序组合): {k * epsilon_per_query:.3f}")