模型优化:
  - torch>=2.0.0           # PyTorch量化
  - onnx>=1.15.0           # ONNX模型格式
  - onnxruntime-gpu        # ONNX推理
  - tensorrt>=10.0         # NVIDIA加速
  - torch-pruning          # 模型剪枝

服务部署:
  - fastapi>=0.115.0       # Web服务框架
  - uvicorn[standard]      # ASGI服务器
  - docker                 # 容器化
  - tritonserver           # NVIDIA推理服务器
  - redis                  # 缓存

MLOps:
  - wandb                  # 实验跟踪
  - mlflow                 # 模型管理
  - prometheus             # 监控
  - grafana                # 可视化
  - dvc                    # 数据版本控制

边缘部署:
  - onnxruntime-mobile     # 移动端推理
  - coremltools            # iOS部署
  - tensorflow-lite        # 移动端推理
  - openvino               # Intel优化

量化值 = round((实际值 - zero_point) / scale)
反量化值 = 量化值 × scale + zero_point

import torch
from torch import nn

# 原始模型
class SimpleModel(nn.Module):
    def __init__(self):
        super().__init__()
        self.fc1 = nn.Linear(128, 64)
        self.fc2 = nn.Linear(64, 10)

    def forward(self, x):
        x = torch.relu(self.fc1(x))
        return self.fc2(x)

# 动态量化（一行代码）
model = SimpleModel()
quantized_model = torch.quantization.quantize_dynamic(
    model,
    {nn.Linear},
    dtype=torch.qint8
)

from torch.cuda.amp import autocast, GradScaler

scaler = GradScaler()

for images, labels in dataloader:
    images, labels = images.cuda(), labels.cuda()

    with autocast():
        outputs = model(images)
        loss = criterion(outputs, labels)

    scaler.scale(loss).backward()
    scaler.step(optimizer)
    scaler.update()

import torch_pruning as tp

pruner = tp.pruner.MagnitudePruner(
    model,
    example_inputs,
    pruning_ratio=0.5  # 剪枝50%
)

pruner.step()

class DistillationLoss(nn.Module):
    def __init__(self, alpha=0.5, temperature=4.0):
        super().__init__()
        self.alpha = alpha
        self.temperature = temperature

    def forward(self, student_logits, teacher_logits, labels):
        hard_loss = F.cross_entropy(student_logits, labels)
        soft_loss = F.kl_div(
            F.log_softmax(student_logits / self.temperature, dim=1),
            F.softmax(teacher_logits / self.temperature, dim=1),
            reduction='batchmean'
        ) * (self.temperature ** 2)

        return self.alpha * hard_loss + (1 - self.alpha) * soft_loss

# 步骤1：导出ONNX
torch.onnx.export(
    model,
    dummy_input,
    'model.onnx',
    opset_version=17
)

# 步骤2：转换TensorRT（见 tensorrt_inference.py）
# 步骤3：推理

PyTorch模型
    ↓ 导出
ONNX格式（通用中间表示）
    ↓ 部署
┌───────────────────────────────────┐
│ ONNX Runtime  TensorRT  OpenVINO │
│ CoreML       TFLite    DirectML  │
└───────────────────────────────────┘

import torch
import torch.onnx

def export_to_onnx(model, save_path, input_shape=(1, 3, 224, 224)):
    """将PyTorch模型导出为ONNX格式"""
    model.eval()

    # 创建示例输入
    dummy_input = torch.randn(*input_shape)

    # 导出ONNX
    torch.onnx.export(
        model,                          # 模型
        dummy_input,                    # 示例输入
        save_path,                      # 保存路径
        export_params=True,             # 导出权重
        opset_version=17,               # ONNX版本
        do_constant_folding=True,       # 常量折叠优化
        input_names=['input'],          # 输入名称
        output_names=['output'],        # 输出名称
        dynamic_axes={                  # 动态维度
            'input': {0: 'batch_size'},
            'output': {0: 'batch_size'}
        }
    )
    print(f"模型已导出到: {save_path}")

# 使用示例
from torchvision.models import resnet50
model = resnet50(pretrained=True)
export_to_onnx(model, "resnet50.onnx")

from ultralytics import YOLO

# 加载模型
model = YOLO("yolov8n.pt")

# 导出ONNX（一行代码）
model.export(format="onnx", opset=17, simplify=True, dynamic=True)

import onnx
import onnxruntime as ort
import numpy as np

def verify_onnx_model(onnx_path, test_input):
    """验证ONNX模型正确性"""
    # 1. 检查模型结构
    model = onnx.load(onnx_path)
    onnx.checker.check_model(model)
    print("✓ 模型结构验证通过")

    # 2. 推理测试
    session = ort.InferenceSession(onnx_path)
    input_name = session.get_inputs()[0].name
    output = session.run(None, {input_name: test_input})[0]

    print(f"✓ 推理测试通过，输出形状: {output.shape}")
    return output

# 使用
test_input = np.random.randn(1, 3, 224, 224).astype(np.float32)
verify_onnx_model("resnet50.onnx", test_input)

import onnx
from onnxsim import simplify

def simplify_onnx(input_path, output_path):
    """简化ONNX模型"""
    model = onnx.load(input_path)

    # 简化
    model_simplified, check = simplify(model)

    if check:
        onnx.save(model_simplified, output_path)

        # 打印对比
        original_size = os.path.getsize(input_path)
        simplified_size = os.path.getsize(output_path)
        print(f"原始大小: {original_size / 1024 / 1024:.2f} MB")
        print(f"简化后: {simplified_size / 1024 / 1024:.2f} MB")
        print(f"减少: {(1 - simplified_size/original_size)*100:.1f}%")
    else:
        print("简化失败，保持原模型")

simplify_onnx("model.onnx", "model_simplified.onnx")

from fastapi import FastAPI, File, UploadFile, HTTPException
from pydantic import BaseModel
import onnxruntime as ort
import numpy as np
from PIL import Image
import io
import time

app = FastAPI(title="CV推理服务", version="1.0.0")

# 全局模型会话
class ModelManager:
    def __init__(self, model_path: str):
        # 配置GPU加速
        providers = ['CUDAExecutionProvider', 'CPUExecutionProvider']
        self.session = ort.InferenceSession(model_path, providers=providers)
        self.input_name = self.session.get_inputs()[0].name
        self.output_name = self.session.get_outputs()[0].name

    def preprocess(self, image: Image.Image) -> np.ndarray:
        """图像预处理"""
        image = image.resize((224, 224))
        image = np.array(image).astype(np.float32)
        image = image / 255.0
        image = (image - [0.485, 0.456, 0.406]) / [0.229, 0.224, 0.225]
        image = image.transpose(2, 0, 1)
        image = np.expand_dims(image, 0)
        return image

    def predict(self, image: np.ndarray) -> np.ndarray:
        """推理"""
        return self.session.run([self.output_name], {self.input_name: image})[0]

# 初始化模型
model = ModelManager("resnet50.onnx")

# 响应模型
class PredictionResponse(BaseModel):
    class_id: int
    confidence: float
    latency_ms: float

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

@app.post("/predict", response_model=PredictionResponse)
async def predict(file: UploadFile = File(...)):
    """图像分类推理"""
    # 验证文件类型
    if not file.content_type.startswith("image/"):
        raise HTTPException(400, "请上传图片文件")

    try:
        # 读取图片
        contents = await file.read()
        image = Image.open(io.BytesIO(contents)).convert("RGB")

        # 预处理
        input_tensor = model.preprocess(image)

        # 推理计时
        start = time.time()
        output = model.predict(input_tensor)
        latency = (time.time() - start) * 1000

        # 后处理
        probs = np.exp(output) / np.exp(output).sum()  # softmax
        class_id = int(np.argmax(probs))
        confidence = float(probs[0][class_id])

        return PredictionResponse(
            class_id=class_id,
            confidence=confidence,
            latency_ms=round(latency, 2)
        )

    except Exception as e:
        raise HTTPException(500, f"推理失败: {str(e)}")

# 运行: uvicorn main:app --host 0.0.0.0 --port 8000

import asyncio
from asyncio import Queue
from typing import List

class BatchInferenceService:
    """批处理推理服务"""

    def __init__(self, model_path: str, batch_size: int = 8, max_wait: float = 0.05):
        self.model = ModelManager(model_path)
        self.batch_size = batch_size
        self.max_wait = max_wait  # 最大等待时间(秒)
        self.queue: Queue = Queue()
        self.running = True

    async def start(self):
        """启动批处理循环"""
        asyncio.create_task(self._batch_loop())

    async def _batch_loop(self):
        """批处理主循环"""
        while self.running:
            batch = []
            futures = []

            # 收集请求
            try:
                # 等待第一个请求
                item = await asyncio.wait_for(self.queue.get(), timeout=1.0)
                batch.append(item[0])
                futures.append(item[1])

                # 收集更多请求（带超时）
                deadline = time.time() + self.max_wait
                while len(batch) < self.batch_size and time.time() < deadline:
                    try:
                        item = await asyncio.wait_for(
                            self.queue.get(),
                            timeout=deadline - time.time()
                        )
                        batch.append(item[0])
                        futures.append(item[1])
                    except asyncio.TimeoutError:
                        break

            except asyncio.TimeoutError:
                continue

            if batch:
                # 批量推理
                inputs = np.concatenate(batch, axis=0)
                outputs = self.model.predict(inputs)

                # 分发结果
                for i, future in enumerate(futures):
                    future.set_result(outputs[i:i+1])

    async def predict(self, input_tensor: np.ndarray) -> np.ndarray:
        """添加到批处理队列"""
        future = asyncio.get_event_loop().create_future()
        await self.queue.put((input_tensor, future))
        return await future

# 使用批处理服务
batch_service = BatchInferenceService("model.onnx", batch_size=8)

@app.on_event("startup")
async def startup():
    await batch_service.start()

@app.post("/predict_batch")
async def predict_batch(file: UploadFile = File(...)):
    # ... 预处理 ...
    output = await batch_service.predict(input_tensor)
    # ... 后处理 ...

model_repository/
├── yolov8/
│   ├── config.pbtxt
│   └── 1/
│       └── model.onnx
├── resnet50/
│   ├── config.pbtxt
│   └── 1/
│       └── model.plan  # TensorRT格式

name: "yolov8"
platform: "onnxruntime_onnx"
max_batch_size: 8
input [
  {
    name: "images"
    data_type: TYPE_FP32
    dims: [ 3, 640, 640 ]
  }
]
output [
  {
    name: "output0"
    data_type: TYPE_FP32
    dims: [ 84, 8400 ]
  }
]
instance_group [
  {
    count: 2
    kind: KIND_GPU
    gpus: [ 0 ]
  }
]
dynamic_batching {
  preferred_batch_size: [ 4, 8 ]
  max_queue_delay_microseconds: 50000
}

# 使用Docker启动
docker run --gpus all --rm -p 8000:8000 -p 8001:8001 -p 8002:8002 \
  -v $(pwd)/model_repository:/models \
  nvcr.io/nvidia/tritonserver:24.05-py3 \
  tritonserver --model-repository=/models

import tritonclient.grpc as grpcclient
import numpy as np

def triton_inference(image: np.ndarray) -> np.ndarray:
    """调用Triton推理服务"""
    client = grpcclient.InferenceServerClient(url="localhost:8001")

    # 准备输入
    inputs = [
        grpcclient.InferInput("images", image.shape, "FP32")
    ]
    inputs[0].set_data_from_numpy(image)

    # 准备输出
    outputs = [
        grpcclient.InferRequestedOutput("output0")
    ]

    # 推理
    response = client.infer(
        model_name="yolov8",
        inputs=inputs,
        outputs=outputs
    )

    return response.as_numpy("output0")

FROM python:3.11-slim

WORKDIR /app

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

# 复制代码和模型
COPY app.py .
COPY models/ ./models/

# 暴露端口
EXPOSE 8000

# 健康检查
HEALTHCHECK --interval=30s --timeout=10s --start-period=5s --retries=3 \
  CMD curl -f http://localhost:8000/health || exit 1

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

version: '3.8'
services:
  inference:
    build: .
    ports:
      - "8000:8000"
    deploy:
      resources:
        reservations:
          devices:
            - driver: nvidia
              count: 1
              capabilities: [gpu]
    environment:
      - CUDA_VISIBLE_DEVICES=0
    volumes:
      - ./models:/app/models
    restart: unless-stopped

  nginx:
    image: nginx:alpine
    ports:
      - "80:80"
    volumes:
      - ./nginx.conf:/etc/nginx/nginx.conf:ro
    depends_on:
      - inference

import coremltools as ct

# PyTorch转CoreML
def convert_to_coreml(pytorch_model, save_path):
    """转换为CoreML格式"""
    # 转为TorchScript
    traced_model = torch.jit.trace(pytorch_model, torch.randn(1, 3, 224, 224))

    # 转换
    mlmodel = ct.convert(
        traced_model,
        inputs=[ct.ImageType(name="input", shape=(1, 3, 224, 224))],
        minimum_deployment_target=ct.target.iOS15
    )

    mlmodel.save(save_path)
    print(f"CoreML模型已保存: {save_path}")

import tensorflow as tf

def convert_to_tflite(saved_model_dir, output_path, quantize=True):
    """转换为TFLite格式"""
    converter = tf.lite.TFLiteConverter.from_saved_model(saved_model_dir)

    if quantize:
        converter.optimizations = [tf.lite.Optimize.DEFAULT]
        converter.target_spec.supported_types = [tf.float16]

    tflite_model = converter.convert()

    with open(output_path, 'wb') as f:
        f.write(tflite_model)

    print(f"TFLite模型大小: {len(tflite_model) / 1024 / 1024:.2f} MB")

from ultralytics import YOLO

# 1. 导出TensorRT引擎（在Jetson上执行）
model = YOLO("yolov8n.pt")
model.export(format="engine", device=0, half=True)  # FP16

# 2. 推理
engine_model = YOLO("yolov8n.engine")
results = engine_model("image.jpg")

# 性能对比（Jetson Orin Nano）
# yolov8n.pt:     ~25 FPS
# yolov8n.engine: ~80 FPS (FP16)

import onnxruntime as ort

def create_mobile_session(model_path):
    """创建移动端优化的推理会话"""
    options = ort.SessionOptions()

    # 优化设置
    options.graph_optimization_level = ort.GraphOptimizationLevel.ORT_ENABLE_ALL
    options.intra_op_num_threads = 4
    options.inter_op_num_threads = 1

    # 使用NNAPI（Android）或CoreML（iOS）
    providers = ['NnapiExecutionProvider', 'CoreMLExecutionProvider', 'CPUExecutionProvider']

    return ort.InferenceSession(model_path, options, providers=providers)

┌─────────────────────────────────────────────────────────┐
│                       负载均衡器                          │
│                    (Nginx/Traefik)                      │
└─────────────────────────┬───────────────────────────────┘
                          │
        ┌─────────────────┼─────────────────┐
        ↓                 ↓                 ↓
┌───────────────┐ ┌───────────────┐ ┌───────────────┐
│   推理服务1    │ │   推理服务2    │ │   推理服务3    │
│  (GPU 0)      │ │  (GPU 1)      │ │  (GPU 2)      │
└───────────────┘ └───────────────┘ └───────────────┘
        │                 │                 │
        └────────┬────────┴────────┬────────┘
                 ↓                 ↓
        ┌───────────────┐  ┌───────────────┐
        │    Redis      │  │   监控系统     │
        │  (结果缓存)    │  │(Prometheus)   │
        └───────────────┘  └───────────────┘

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, Field
from typing import List, Optional
from datetime import datetime

app = FastAPI(
    title="视觉AI服务",
    version="2.0.0",
    description="生产级计算机视觉推理API"
)

# === 数据模型 ===

class BoundingBox(BaseModel):
    x1: float = Field(..., description="左上角x坐标")
    y1: float = Field(..., description="左上角y坐标")
    x2: float = Field(..., description="右下角x坐标")
    y2: float = Field(..., description="右下角y坐标")

class Detection(BaseModel):
    class_name: str
    confidence: float = Field(..., ge=0, le=1)
    bbox: BoundingBox

class DetectionResponse(BaseModel):
    request_id: str
    timestamp: datetime
    detections: List[Detection]
    inference_time_ms: float
    image_size: tuple

# === 端点 ===

@app.post("/v2/detect", response_model=DetectionResponse)
async def detect_objects(
    file: UploadFile = File(...),
    confidence_threshold: float = 0.5,
    max_detections: int = 100
):
    """
    目标检测API

    - **file**: 待检测图片
    - **confidence_threshold**: 置信度阈值 (0-1)
    - **max_detections**: 最大检测数量
    """
    # ... 实现 ...

from prometheus_client import Counter, Histogram, Gauge, generate_latest
from fastapi import Response

# 定义指标
REQUEST_COUNT = Counter(
    'inference_requests_total',
    'Total inference requests',
    ['model', 'status']
)

INFERENCE_LATENCY = Histogram(
    'inference_latency_seconds',
    'Inference latency',
    ['model'],
    buckets=[0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1.0]
)

GPU_UTILIZATION = Gauge(
    'gpu_utilization_percent',
    'GPU utilization',
    ['gpu_id']
)

# 在推理中记录
@app.post("/predict")
async def predict(file: UploadFile):
    start_time = time.time()

    try:
        result = model.predict(...)
        REQUEST_COUNT.labels(model='yolov8', status='success').inc()
    except Exception as e:
        REQUEST_COUNT.labels(model='yolov8', status='error').inc()
        raise
    finally:
        INFERENCE_LATENCY.labels(model='yolov8').observe(time.time() - start_time)

    return result

# 暴露指标端点
@app.get("/metrics")
def metrics():
    return Response(content=generate_latest(), media_type="text/plain")

需要部署模型?
    │
    ├─ 云端服务 ─→ 高并发? ─→ 是 ─→ Triton Server
    │                      └─ 否 ─→ FastAPI + ONNX Runtime
    │
    ├─ 边缘设备 ─→ NVIDIA硬件? ─→ 是 ─→ TensorRT
    │                          └─ 否 ─→ ONNX Runtime
    │
    └─ 移动端 ─→ iOS? ─→ CoreML
               └─ Android? ─→ TFLite/NNAPI

研究环境:                生产环境:
├── notebooks/           ├── 模型性能下降
│   ├── exp1.ipynb      ├── 无法复现结果
│   ├── exp2.ipynb      ├── 数据漂移
│   └── final_v3.ipynb  └── 版本混乱
└── "这个版本是哪个?"

┌──────────────────────────────────────────────────────┐
│                    MLOps流水线                        │
├──────────────────────────────────────────────────────┤
│  数据 → 训练 → 评估 → 部署 → 监控 → 反馈 → 数据...   │
│   ↓      ↓      ↓      ↓      ↓      ↓              │
│  DVC   W&B   测试    CI/CD  监控   A/B              │
└──────────────────────────────────────────────────────┘

# 安装
pip install dvc dvc-s3  # 或 dvc-gs, dvc-azure

# 初始化
dvc init

# 跟踪数据目录
dvc add data/images
git add data/images.dvc .gitignore
git commit -m "Add training images"

# 推送到远程存储
dvc remote add -d myremote s3://my-bucket/dvc
dvc push

# 其他人拉取数据
git clone https://github.com/...
dvc pull

# dvc.yaml
stages:
  preprocess:
    cmd: python src/preprocess.py
    deps:
      - data/raw
      - src/preprocess.py
    outs:
      - data/processed

  train:
    cmd: python src/train.py --epochs 100
    deps:
      - data/processed
      - src/train.py
    params:
      - train.epochs
      - train.lr
    outs:
      - models/model.pt
    metrics:
      - metrics.json:
          cache: false

dvc repro  # 自动执行有变更的阶段
dvc metrics show  # 查看指标

# 安装
pip install label-studio

# 启动
label-studio start

# 访问 http://localhost:8080

import pandas as pd
from pathlib import Path

class DataQualityChecker:
    """数据质量检查器"""

    def __init__(self, data_dir: str):
        self.data_dir = Path(data_dir)

    def check_images(self):
        """检查图像完整性"""
        issues = []

        for img_path in self.data_dir.glob("**/*.jpg"):
            try:
                img = Image.open(img_path)
                img.verify()

                # 检查尺寸
                if min(img.size) < 32:
                    issues.append(f"图片太小: {img_path}")

            except Exception as e:
                issues.append(f"损坏文件: {img_path} - {e}")

        return issues

    def check_labels(self, labels_file: str):
        """检查标签质量"""
        df = pd.read_csv(labels_file)
        issues = []

        # 检查缺失值
        missing = df.isnull().sum()
        if missing.any():
            issues.append(f"缺失值: {missing[missing > 0].to_dict()}")

        # 检查类别分布
        class_counts = df['label'].value_counts()
        imbalance = class_counts.max() / class_counts.min()
        if imbalance > 10:
            issues.append(f"类别不平衡严重: {imbalance:.1f}x")

        return issues

    def generate_report(self):
        """生成数据质量报告"""
        image_issues = self.check_images()
        label_issues = self.check_labels("labels.csv")

        report = {
            "total_images": len(list(self.data_dir.glob("**/*.jpg"))),
            "image_issues": len(image_issues),
            "label_issues": len(label_issues),
            "details": {
                "images": image_issues[:10],  # 前10个问题
                "labels": label_issues
            }
        }

        return report

import wandb

# 登录（首次需要API key）
wandb.login()

# 初始化实验
wandb.init(
    project="yolov8-detection",
    name="exp-001-baseline",
    config={
        "model": "yolov8n",
        "epochs": 100,
        "batch_size": 16,
        "lr": 0.01,
        "optimizer": "SGD",
        "augmentation": "mosaic"
    }
)

from ultralytics import YOLO
import wandb

def train_with_wandb():
    # 初始化
    wandb.init(project="yolov8-detection")

    # 训练
    model = YOLO("yolov8n.pt")

    # 训练循环中记录指标
    for epoch in range(wandb.config.epochs):
        # 训练...
        train_loss = ...
        val_map = ...

        # 记录指标
        wandb.log({
            "epoch": epoch,
            "train/loss": train_loss,
            "val/mAP50": val_map,
            "lr": optimizer.param_groups[0]['lr']
        })

        # 记录样本预测（每10个epoch）
        if epoch % 10 == 0:
            predictions = model.predict(val_images[:5])
            wandb.log({
                "predictions": [
                    wandb.Image(img, caption=f"Epoch {epoch}")
                    for img in predictions
                ]
            })

    # 保存最佳模型
    wandb.save("best.pt")
    wandb.finish()

from ultralytics import YOLO

model = YOLO("yolov8n.pt")

# 自动集成W&B
model.train(
    data="coco128.yaml",
    epochs=100,
    project="yolov8-wandb"  # 自动上传到W&B
)

import wandb

# 定义搜索空间
sweep_config = {
    "method": "bayes",  # random, grid, bayes
    "metric": {
        "name": "val/mAP50",
        "goal": "maximize"
    },
    "parameters": {
        "lr": {
            "min": 0.0001,
            "max": 0.1,
            "distribution": "log_uniform_values"
        },
        "batch_size": {
            "values": [8, 16, 32, 64]
        },
        "optimizer": {
            "values": ["SGD", "Adam", "AdamW"]
        },
        "augmentation": {
            "values": ["none", "basic", "mosaic"]
        }
    }
}

# 创建sweep
sweep_id = wandb.sweep(sweep_config, project="yolov8-sweep")

# 训练函数
def train():
    wandb.init()
    config = wandb.config

    model = YOLO("yolov8n.pt")
    model.train(
        data="coco128.yaml",
        epochs=50,
        lr0=config.lr,
        batch=config.batch_size,
        optimizer=config.optimizer
    )

    wandb.finish()

# 运行sweep
wandb.agent(sweep_id, train, count=20)  # 跑20组实验

import wandb

# 注册模型
def register_model(model_path: str, model_name: str, metrics: dict):
    """将训练好的模型注册到W&B Model Registry"""

    run = wandb.init(project="model-registry", job_type="register")

    # 创建模型artifact
    artifact = wandb.Artifact(
        name=model_name,
        type="model",
        metadata=metrics
    )
    artifact.add_file(model_path)

    # 记录模型
    run.log_artifact(artifact)

    # 链接到Model Registry
    run.link_artifact(artifact, f"model-registry/{model_name}")

    run.finish()

# 使用
register_model(
    "best.pt",
    "yolov8-detector",
    {"mAP50": 0.85, "mAP50-95": 0.65, "FPS": 120}
)

from dataclasses import dataclass
from datetime import datetime, timedelta
import numpy as np

@dataclass
class ModelMetrics:
    """模型性能指标"""
    timestamp: datetime
    latency_p50: float
    latency_p95: float
    latency_p99: float
    throughput: float
    error_rate: float
    confidence_mean: float
    confidence_std: float

class ModelMonitor:
    """模型监控系统"""

    def __init__(self, window_size: int = 1000):
        self.window_size = window_size
        self.predictions = []
        self.latencies = []
        self.errors = 0

    def record_prediction(self, confidence: float, latency: float, success: bool):
        """记录一次预测"""
        self.predictions.append(confidence)
        self.latencies.append(latency)
        if not success:
            self.errors += 1

        # 保持窗口大小
        if len(self.predictions) > self.window_size:
            self.predictions.pop(0)
            self.latencies.pop(0)

    def get_metrics(self) -> ModelMetrics:
        """获取当前指标"""
        return ModelMetrics(
            timestamp=datetime.now(),
            latency_p50=np.percentile(self.latencies, 50),
            latency_p95=np.percentile(self.latencies, 95),
            latency_p99=np.percentile(self.latencies, 99),
            throughput=len(self.predictions) / 60,  # per minute
            error_rate=self.errors / len(self.predictions) if self.predictions else 0,
            confidence_mean=np.mean(self.predictions),
            confidence_std=np.std(self.predictions)
        )

    def check_alerts(self) -> list:
        """检查告警条件"""
        alerts = []
        metrics = self.get_metrics()

        # 延迟告警
        if metrics.latency_p95 > 100:  # 100ms
            alerts.append(f"高延迟告警: P95={metrics.latency_p95:.1f}ms")

        # 错误率告警
        if metrics.error_rate > 0.01:  # 1%
            alerts.append(f"错误率告警: {metrics.error_rate*100:.2f}%")

        # 置信度异常（可能是数据漂移）
        if metrics.confidence_mean < 0.5:
            alerts.append(f"置信度下降告警: mean={metrics.confidence_mean:.3f}")

        return alerts

from scipy import stats
import numpy as np

class DriftDetector:
    """数据漂移检测器"""

    def __init__(self, reference_data: np.ndarray):
        self.reference = reference_data
        self.reference_mean = np.mean(reference_data, axis=0)
        self.reference_std = np.std(reference_data, axis=0)

    def detect_drift(self, current_data: np.ndarray, alpha: float = 0.05) -> dict:
        """
        检测数据漂移

        使用多种统计检验：
        - KS检验：分布变化
        - Z检验：均值变化
        - F检验：方差变化
        """
        results = {
            "has_drift": False,
            "tests": {}
        }

        # KS检验（分布）
        ks_stat, ks_pvalue = stats.ks_2samp(
            self.reference.flatten(),
            current_data.flatten()
        )
        results["tests"]["ks"] = {
            "statistic": ks_stat,
            "pvalue": ks_pvalue,
            "drift": ks_pvalue < alpha
        }

        # 均值变化
        current_mean = np.mean(current_data, axis=0)
        z_score = (current_mean - self.reference_mean) / (self.reference_std / np.sqrt(len(current_data)))
        mean_drift = np.abs(z_score) > 3
        results["tests"]["mean"] = {
            "z_score": float(np.mean(np.abs(z_score))),
            "drift": bool(np.any(mean_drift))
        }

        # 综合判断
        results["has_drift"] = any(t["drift"] for t in results["tests"].values())

        return results

# 使用示例
detector = DriftDetector(training_features)
drift_result = detector.detect_drift(production_features)

if drift_result["has_drift"]:
    print("⚠️ 检测到数据漂移，建议重新训练模型")

import hashlib
from typing import Literal

class ABTestManager:
    """A/B测试管理器"""

    def __init__(self):
        self.experiments = {}
        self.results = {}

    def create_experiment(
        self,
        name: str,
        model_a: str,
        model_b: str,
        traffic_split: float = 0.5
    ):
        """创建A/B测试实验"""
        self.experiments[name] = {
            "model_a": model_a,
            "model_b": model_b,
            "traffic_split": traffic_split,
            "start_time": datetime.now()
        }
        self.results[name] = {
            "a": {"requests": 0, "success": 0, "latency": []},
            "b": {"requests": 0, "success": 0, "latency": []}
        }

    def get_variant(self, experiment_name: str, user_id: str) -> Literal["a", "b"]:
        """
        基于用户ID分配变体（确保同一用户始终看到相同变体）
        """
        exp = self.experiments[experiment_name]

        # 使用哈希确保一致性
        hash_val = int(hashlib.md5(f"{experiment_name}:{user_id}".encode()).hexdigest(), 16)

        if (hash_val % 100) / 100 < exp["traffic_split"]:
            return "a"
        return "b"

    def record_result(
        self,
        experiment_name: str,
        variant: str,
        success: bool,
        latency: float
    ):
        """记录实验结果"""
        result = self.results[experiment_name][variant]
        result["requests"] += 1
        if success:
            result["success"] += 1
        result["latency"].append(latency)

    def analyze(self, experiment_name: str) -> dict:
        """分析实验结果"""
        results = self.results[experiment_name]

        analysis = {}
        for variant in ["a", "b"]:
            r = results[variant]
            analysis[variant] = {
                "success_rate": r["success"] / r["requests"] if r["requests"] > 0 else 0,
                "avg_latency": np.mean(r["latency"]) if r["latency"] else 0,
                "p95_latency": np.percentile(r["latency"], 95) if r["latency"] else 0,
                "sample_size": r["requests"]
            }

        # 统计显著性检验
        if results["a"]["requests"] > 100 and results["b"]["requests"] > 100:
            # 成功率卡方检验
            contingency = [
                [results["a"]["success"], results["a"]["requests"] - results["a"]["success"]],
                [results["b"]["success"], results["b"]["requests"] - results["b"]["success"]]
            ]
            chi2, p_value = stats.chi2_contingency(contingency)[:2]
            analysis["significance"] = {
                "chi2": chi2,
                "p_value": p_value,
                "is_significant": p_value < 0.05
            }

        return analysis

# 使用示例
ab_manager = ABTestManager()

# 创建实验
ab_manager.create_experiment(
    name="yolov8_vs_yolo11",
    model_a="yolov8n.pt",
    model_b="yolo11n.pt",
    traffic_split=0.5
)

# 在推理服务中使用
@app.post("/predict")
async def predict(file: UploadFile, user_id: str):
    variant = ab_manager.get_variant("yolov8_vs_yolo11", user_id)
    model = models[variant]

    start = time.time()
    try:
        result = model.predict(...)
        success = True
    except:
        success = False
    latency = time.time() - start

    ab_manager.record_result("yolov8_vs_yolo11", variant, success, latency)
    return result

## 项目结构

project/
├── data/
│   ├── raw/           # 原始数据（DVC跟踪）
│   ├── processed/     # 处理后数据
│   └── data.dvc       # DVC元数据
├── src/
│   ├── data/          # 数据处理
│   ├── models/        # 模型定义
│   ├── training/      # 训练脚本
│   └── inference/     # 推理服务
├── configs/
│   └── train.yaml     # 训练配置
├── tests/
│   ├── test_data.py
│   └── test_model.py
├── notebooks/         # 实验notebook
├── docker/
│   └── Dockerfile
├── dvc.yaml           # DVC流水线
├── requirements.txt
└── README.md

# 使用类型注解
def preprocess_image(
    image_path: str,
    target_size: tuple[int, int] = (640, 640),
    normalize: bool = True
) -> np.ndarray:
    """
    预处理图像

    Args:
        image_path: 图像文件路径
        target_size: 目标尺寸 (height, width)
        normalize: 是否归一化到[0,1]

    Returns:
        处理后的图像数组，形状为 (C, H, W)

    Raises:
        FileNotFoundError: 图像文件不存在
        ValueError: 图像格式不支持
    """
    ...

# 使用配置文件而非硬编码
# config.yaml
model:
  name: yolov8n
  input_size: 640

training:
  epochs: 100
  batch_size: 16
  lr: 0.01

# 加载配置
import yaml
with open("config.yaml") as f:
    config = yaml.safe_load(f)