模型版本管理MLOps实践

import mlflow

# 记录模型版本
with mlflow.start_run(run_name="qwen3-32b-v2"):
    mlflow.log_params({
        "base_model": "Qwen-3-32B",
        "fine_tune_method": "LoRA",
        "learning_rate": 2e-4,
        "epochs": 3,
        "dataset": "instruction-v3",
    })
    
    mlflow.log_metrics({
        "eval_loss": 0.45,
        "eval_accuracy": 0.89,
        "human_eval_score": 4.2,
    })
    
    # 注册模型
    mlflow.register_model(
        "runs:/abc123/model",
        "qwen3-32b-instruct",
        tags={
            "version": "v2.1",
            "stage": "staging",
            "creator": "team-agi",
        }
    )

# 初始化DVC
dvc init

# 添加模型文件到DVC
dvc add models/qwen3-32b-v2.1/

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

# Git只追踪.dvc文件（指针），不追踪实际大文件
git add models/qwen3-32b-v2.1/.dvc
git commit -m "Add qwen3-32b v2.1"

class CanaryDeployment:
    def __init__(self, stable_version, canary_version, canary_ratio=0.1):
        self.stable = stable_version
        self.canary = canary_version
        self.ratio = canary_ratio
        self.metrics = {"stable": [], "canary": []}
    
    def route(self, request):
        """灰度路由"""
        import random
        if random.random() < self.ratio:
            self.metrics["canary"].append({"time": time.time()})
            return self.canary
        else:
            self.metrics["stable"].append({"time": time.time()})
            return self.stable
    
    def evaluate(self):
        """评估灰度结果"""
        canary_latency = self.get_avg_latency("canary")
        stable_latency = self.get_avg_latency("stable")
        
        canary_error = self.get_error_rate("canary")
        stable_error = self.get_error_rate("stable")
        
        # 灰度通过条件
        if canary_latency > stable_latency * 1.2:
            return "rollback", "Canary latency too high"
        if canary_error > stable_error * 2:
            return "rollback", "Canary error rate too high"
        
        return "promote", "Canary performing well"

class ABTest:
    def __init__(self, models, weights=None):
        self.models = models
        self.weights = weights or [1/len(models)] * len(models)
        self.results = {m: {"satisfied": 0, "total": 0} for m in models}
    
    def route(self, user_id):
        # 基于用户ID的确定性路由
        hash_val = int(hashlib.md5(user_id.encode()).hexdigest(), 16)
        cumulative = 0
        for model, weight in zip(self.models, self.weights):
            cumulative += weight
            if (hash_val % 1000) / 1000 < cumulative:
                return model
    
    def record_feedback(self, model, satisfied):
        self.results[model]["total"] += 1
        if satisfied:
            self.results[model]["satisfied"] += 1
    
    def get_winner(self):
        rates = {m: r["satisfied"]/r["total"] for m, r in self.results.items() if r["total"] > 0}
        return max(rates, key=rates.get) if rates else None

# .github/workflows/model-deploy.yml
name: Model Deploy Pipeline
on:
  push:
    tags: ['v*']

jobs:
  evaluate:
    runs-on: gpu-runner
    steps:
      - uses: actions/checkout@v4
      - name: Run evaluation
        run: |
          python eval.py --model checkpoints/latest --benchmark mmlu,gsm8k,humaneval
      - name: Check quality gates
        run: |
          python check_gates.py --min_accuracy 0.85 --max_regression 0.02
  
  deploy_staging:
    needs: evaluate
    runs-on: deploy-runner
    steps:
      - name: Deploy to staging
        run: |
          ./deploy.sh --env staging --version ${{ github.ref_name }}
      - name: Run smoke tests
        run: |
          python smoke_test.py --env staging
  
  canary:
    needs: deploy_staging
    runs-on: deploy-runner
    steps:
      - name: Canary deployment (10%)
        run: |
          ./deploy.sh --env production --version ${{ github.ref_name }} --canary 0.1
      - name: Monitor for 1 hour
        run: |
          python monitor.py --duration 3600 --check latency,error_rate
  
  full_deploy:
    needs: canary
    runs-on: deploy-runner
    steps:
      - name: Full deployment
        run: |
          ./deploy.sh --env production --version ${{ github.ref_name }} --promote

class ModelRollback:
    def __init__(self, deployment_manager):
        self.deployment = deployment_manager
        self.version_history = []
    
    async def rollback(self, target_version=None):
        """回滚到指定版本或上一个稳定版本"""
        if target_version is None:
            target_version = self.get_previous_stable()
        
        logger.info(f"Rolling back to {target_version}")
        
        # 快速切换流量
        await self.deployment.switch_traffic(
            from_version="current",
            to_version=target_version,
            ratio=1.0  # 100%切换
        )
        
        # 验证回滚
        health = await self.deployment.health_check(target_version)
        if not health:
            logger.error("Rollback target also unhealthy!")
            return False
        
        return True