필사 모드: AI 平台模型注册表与 A/B 部署流水线设计 2026

"""
MLflow Model Registry：模型注册、版本管理、Alias 设置
以 MLflow 2.19+ 为基准
"""
import mlflow
from mlflow.tracking import MlflowClient

# 连接 MLflow Tracking Server
mlflow.set_tracking_uri("http://mlflow.internal.example.com:5000")
client = MlflowClient()

# 1. 训练运行与模型日志记录
with mlflow.start_run(run_name="fraud-detection-v3") as run:
    # 记录训练指标
    mlflow.log_params({
        "model_type": "xgboost",
        "n_estimators": 500,
        "max_depth": 8,
        "learning_rate": 0.05,
        "training_data_version": "2026-03-01",
        "feature_count": 47,
    })
    mlflow.log_metrics({
        "auc_roc": 0.9847,
        "precision_at_95_recall": 0.912,
        "f1_score": 0.943,
        "latency_p99_ms": 12.3,
    })

    # 一次性完成模型 artifact 日志记录 + 注册表注册
    model_info = mlflow.xgboost.log_model(
        xgb_model=trained_model,
        artifact_path="model",
        registered_model_name="fraud-detection",
        input_example=sample_input,
        signature=mlflow.models.infer_signature(X_test, y_pred),
    )
    print(f"Model URI: {model_info.model_uri}")
    # 输出：models:/fraud-detection/3

# 2. 为已注册的模型版本分配 Alias
# 将新模型指定为 challenger
client.set_registered_model_alias(
    name="fraud-detection",
    alias="challenger",
    version=3,
)

# 3. 验证完成后晋升为 champion
def promote_to_champion(model_name: str, version: int):
    """将模型晋升为 champion，并记录此前 champion 的信息。"""
    # 确认当前 champion
    try:
        current_champion = client.get_model_version_by_alias(
            name=model_name, alias="champion"
        )
        # 为此前的 champion 添加 archived 标签
        client.set_model_version_tag(
            name=model_name,
            version=current_champion.version,
            key="previously_champion",
            value="true",
        )
        print(f"Previous champion: v{current_champion.version}")
    except mlflow.exceptions.MlflowException:
        print("No existing champion found")

    # 指定新的 champion
    client.set_registered_model_alias(
        name=model_name, alias="champion", version=version
    )
    # 移除 challenger alias
    client.delete_registered_model_alias(
        name=model_name, alias="challenger"
    )
    print(f"Promoted v{version} to champion")

promote_to_champion("fraud-detection", version=3)

"""
基于 Alias 加载模型并用于服务的模式。
在部署流水线中自动 resolve champion 模型的 URI。
"""
import mlflow

# 加载 champion 模型（基于 Alias）
champion_model = mlflow.pyfunc.load_model("models:/fraud-detection@champion")

# 加载特定版本（用于调试/对比）
v2_model = mlflow.pyfunc.load_model("models:/fraud-detection/2")

# 将模型 URI 转换为 KServe storageUri 的工具函数
def resolve_model_storage_uri(model_name: str, alias: str) -> str:
    """从 MLflow Alias 中提取实际的 artifact storage URI。"""
    client = MlflowClient()
    mv = client.get_model_version_by_alias(name=model_name, alias=alias)
    run = client.get_run(mv.run_id)
    artifact_uri = run.info.artifact_uri
    # 例：s3://ml-artifacts/12/abc123/artifacts/model
    return f"{artifact_uri}/model"

storage_uri = resolve_model_storage_uri("fraud-detection", "champion")
print(f"KServe storageUri: {storage_uri}")
# 输出：s3://ml-artifacts/12/abc123/artifacts/model

"""
在模型注册时校验必备元数据的守门器（gatekeeper）。
插入到 CI/CD 流水线的注册环节，
防止不完整的模型进入注册表。
"""
from dataclasses import dataclass
from typing import Optional

REQUIRED_TAGS = [
    "training_data_version",
    "feature_schema_hash",
    "evaluation_dataset",
    "code_commit_sha",
    "owner_team",
]

REQUIRED_METRICS = [
    "auc_roc",
    "latency_p99_ms",
]

@dataclass
class RegistrationValidation:
    is_valid: bool
    missing_tags: list[str]
    missing_metrics: list[str]
    message: str

def validate_before_registration(
    run_id: str,
    min_auc: float = 0.95,
    max_latency_ms: float = 50.0,
) -> RegistrationValidation:
    """在模型注册前校验必备条件。"""
    client = MlflowClient()
    run = client.get_run(run_id)

    # 校验必备标签
    existing_tags = set(run.data.tags.keys())
    missing_tags = [t for t in REQUIRED_TAGS if t not in existing_tags]

    # 校验必备指标
    existing_metrics = set(run.data.metrics.keys())
    missing_metrics = [m for m in REQUIRED_METRICS if m not in existing_metrics]

    # 校验性能基线
    threshold_failures = []
    if "auc_roc" in existing_metrics:
        auc = run.data.metrics["auc_roc"]
        if auc < min_auc:
            threshold_failures.append(
                f"auc_roc={auc:.4f} < min={min_auc}"
            )
    if "latency_p99_ms" in existing_metrics:
        latency = run.data.metrics["latency_p99_ms"]
        if latency > max_latency_ms:
            threshold_failures.append(
                f"latency_p99_ms={latency:.1f} > max={max_latency_ms}"
            )

    is_valid = (
        len(missing_tags) == 0
        and len(missing_metrics) == 0
        and len(threshold_failures) == 0
    )

    messages = []
    if missing_tags:
        messages.append(f"Missing tags: {missing_tags}")
    if missing_metrics:
        messages.append(f"Missing metrics: {missing_metrics}")
    if threshold_failures:
        messages.append(f"Threshold failures: {threshold_failures}")

    return RegistrationValidation(
        is_valid=is_valid,
        missing_tags=missing_tags,
        missing_metrics=missing_metrics,
        message=" | ".join(messages) if messages else "All checks passed",
    )

1. 模型注册（MLflow Registry）
   └─ 通过必备元数据校验
   └─ 分配 @challenger alias

2. 部署前校验（Pre-deployment Validation）
   └─ 用离线评估数据集确认准确率
   └─ 推理延迟基准测试（p50、p95、p99）
   └─ 校验输入输出 schema 兼容性

3. 启动金丝雀部署
   └─ 更新 KServe InferenceService
   └─ 设置 canaryTrafficPercent: 5
   └─ 启用监控看板

4. 渐进式提升流量
   └─ 5% -> 10% -> 25% -> 50% -> 100%
   └─ 每个阶段自动对比指标
   └─ 检测到异常时自动回滚

5. 晋升或回滚
   └─ 成功：切换 @champion alias，旧版本设为 @archived
   └─ 失败：canaryTrafficPercent: 0，记录原因分析标签

# kserve-canary-deploy.yaml
# KServe InferenceService：金丝雀部署配置
# 给新模型分配 10% 的流量，并逐步提升。
apiVersion: serving.kserve.io/v1beta1
kind: InferenceService
metadata:
  name: fraud-detection
  namespace: ml-serving
  annotations:
    # 最小副本数设置（防止冷启动）
    autoscaling.knative.dev/min-scale: '2'
    # 部署元数据
    mlflow.model.version: '3'
    mlflow.model.alias: 'challenger'
    deploy.pipeline.run-id: 'deploy-20260304-001'
spec:
  predictor:
    # 金丝雀流量比例（0-100）
    canaryTrafficPercent: 10
    # 新模型版本（金丝雀）
    model:
      modelFormat:
        name: xgboost
      runtime: kserve-xgbserver
      storageUri: 's3://ml-artifacts/fraud-detection/v3/model'
      resources:
        requests:
          cpu: '2'
          memory: '4Gi'
        limits:
          cpu: '4'
          memory: '8Gi'
    # Autoscaler 配置
    scaleTarget: 10
    scaleMetric: concurrency
  # 流量路由配置（可选）
  transformer:
    containers:
      - name: feature-enricher
        image: registry.example.com/ml/feature-enricher:v2.1
        resources:
          requests:
            cpu: '1'
            memory: '2Gi'

"""
分阶段提升 KServe 金丝雀流量的自动化脚本。
每个阶段都会检查核心指标，一旦未达标准就立即回滚。
"""
import time
import subprocess
import json
from dataclasses import dataclass

@dataclass
class CanaryStage:
    traffic_percent: int
    observation_minutes: int
    max_error_rate: float
    max_latency_p99_ms: float

CANARY_STAGES = [
    CanaryStage(traffic_percent=5,   observation_minutes=10, max_error_rate=0.01, max_latency_p99_ms=50),
    CanaryStage(traffic_percent=10,  observation_minutes=15, max_error_rate=0.01, max_latency_p99_ms=50),
    CanaryStage(traffic_percent=25,  observation_minutes=20, max_error_rate=0.005, max_latency_p99_ms=45),
    CanaryStage(traffic_percent=50,  observation_minutes=30, max_error_rate=0.005, max_latency_p99_ms=45),
    CanaryStage(traffic_percent=100, observation_minutes=0,  max_error_rate=0.005, max_latency_p99_ms=45),
]

def update_canary_traffic(
    isvc_name: str,
    namespace: str,
    traffic_percent: int,
) -> bool:
    """更新 KServe InferenceService 的 canaryTrafficPercent。"""
    patch = {
        "spec": {
            "predictor": {
                "canaryTrafficPercent": traffic_percent
            }
        }
    }
    cmd = [
        "kubectl", "patch", "inferenceservice", isvc_name,
        "-n", namespace,
        "--type=merge",
        f"-p={json.dumps(patch)}",
    ]
    result = subprocess.run(cmd, capture_output=True, text=True)
    if result.returncode != 0:
        print(f"Failed to patch: {result.stderr}")
        return False
    print(f"Updated canaryTrafficPercent to {traffic_percent}%")
    return True

def check_canary_health(
    isvc_name: str,
    namespace: str,
    stage: CanaryStage,
    prometheus_url: str = "http://prometheus.monitoring:9090",
) -> bool:
    """从 Prometheus 中检查金丝雀指标。"""
    import requests

    # 检查错误率
    error_query = (
        f'sum(rate(revision_request_count'
        f'{{service_name="{isvc_name}",response_code!="200",'
        f'namespace="{namespace}"}}[5m])) / '
        f'sum(rate(revision_request_count'
        f'{{service_name="{isvc_name}",'
        f'namespace="{namespace}"}}[5m]))'
    )
    resp = requests.get(
        f"{prometheus_url}/api/v1/query",
        params={"query": error_query},
    )
    data = resp.json()
    if data["data"]["result"]:
        error_rate = float(data["data"]["result"][0]["value"][1])
        if error_rate > stage.max_error_rate:
            print(f"Error rate {error_rate:.4f} exceeds {stage.max_error_rate}")
            return False

    # 检查 p99 延迟
    latency_query = (
        f'histogram_quantile(0.99, '
        f'sum(rate(revision_request_latencies_bucket'
        f'{{service_name="{isvc_name}",'
        f'namespace="{namespace}"}}[5m])) by (le))'
    )
    resp = requests.get(
        f"{prometheus_url}/api/v1/query",
        params={"query": latency_query},
    )
    data = resp.json()
    if data["data"]["result"]:
        latency_ms = float(data["data"]["result"][0]["value"][1])
        if latency_ms > stage.max_latency_p99_ms:
            print(f"p99 latency {latency_ms:.1f}ms exceeds {stage.max_latency_p99_ms}ms")
            return False

    return True

def run_canary_rollout(isvc_name: str, namespace: str):
    """按阶段执行金丝雀发布。"""
    for i, stage in enumerate(CANARY_STAGES):
        print(f"\n--- Stage {i+1}/{len(CANARY_STAGES)}: "
              f"{stage.traffic_percent}% traffic ---")

        if not update_canary_traffic(isvc_name, namespace, stage.traffic_percent):
            print("ABORT: Failed to update traffic")
            rollback(isvc_name, namespace)
            return False

        if stage.observation_minutes > 0:
            print(f"Observing for {stage.observation_minutes} minutes...")
            # 在观察期内定期检查指标
            checks = stage.observation_minutes // 2  # 每 2 分钟检查一次
            for check in range(checks):
                time.sleep(120)  # 等待 2 分钟
                if not check_canary_health(isvc_name, namespace, stage):
                    print(f"ROLLBACK: Health check failed at stage {i+1}")
                    rollback(isvc_name, namespace)
                    return False
                print(f"  Check {check+1}/{checks}: OK")

    print("\nCanary rollout completed successfully!")
    return True

def rollback(isvc_name: str, namespace: str):
    """将金丝雀流量设置为 0%，立即回滚。"""
    update_canary_traffic(isvc_name, namespace, 0)
    print("Rolled back to previous stable version")

# istio-ab-routing.yaml
# 基于请求头的 A/B 流量路由
# 根据 x-model-variant 请求头路由到不同的模型版本。
apiVersion: networking.istio.io/v1beta1
kind: VirtualService
metadata:
  name: fraud-detection-ab
  namespace: ml-serving
spec:
  hosts:
    - fraud-detection.ml-serving.svc.cluster.local
  http:
    # 规则 1：明确请求 variant B -> 新模型
    - match:
        - headers:
            x-model-variant:
              exact: 'B'
      route:
        - destination:
            host: fraud-detection-predictor.ml-serving.svc.cluster.local
            port:
              number: 80
          headers:
            request:
              set:
                x-model-version: 'v3'
          weight: 100
    # 规则 2：特定用户分组 -> 新模型（基于 Cookie）
    - match:
        - headers:
            cookie:
              regex: '.*ab_group=treatment.*'
      route:
        - destination:
            host: fraud-detection-predictor.ml-serving.svc.cluster.local
            port:
              number: 80
          weight: 100
    # 规则 3：默认流量 -> 按权重拆分
    - route:
        - destination:
            host: fraud-detection-predictor.ml-serving.svc.cluster.local
            subset: stable
          weight: 90
        - destination:
            host: fraud-detection-predictor.ml-serving.svc.cluster.local
            subset: canary
          weight: 10
---
# 用 DestinationRule 定义 subset
apiVersion: networking.istio.io/v1beta1
kind: DestinationRule
metadata:
  name: fraud-detection-subsets
  namespace: ml-serving
spec:
  host: fraud-detection-predictor.ml-serving.svc.cluster.local
  subsets:
    - name: stable
      labels:
        model-version: 'v2'
    - name: canary
      labels:
        model-version: 'v3'

"""
A/B 部署监控与自动回滚控制器。
从 Prometheus 采集指标，一旦违反标准就触发自动回滚。
"""
import time
import requests
import logging
from dataclasses import dataclass, field
from enum import Enum

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("ab-monitor")

class RollbackReason(Enum):
    ERROR_RATE = "error_rate_exceeded"
    LATENCY = "latency_exceeded"
    PREDICTION_DRIFT = "prediction_distribution_drift"
    BUSINESS_KPI = "business_kpi_degradation"

@dataclass
class MonitorConfig:
    prometheus_url: str = "http://prometheus.monitoring:9090"
    isvc_name: str = "fraud-detection"
    namespace: str = "ml-serving"
    check_interval_sec: int = 60
    # 阈值设置
    max_error_rate: float = 0.01
    max_latency_p99_ms: float = 50.0
    max_psi: float = 0.2  # PSI > 0.2 视为有意义的分布变化
    # 连续失败次数标准
    consecutive_failures_to_rollback: int = 3
    # 告警设置
    slack_webhook: str = ""
    pagerduty_key: str = ""

@dataclass
class MonitorState:
    consecutive_failures: int = 0
    total_checks: int = 0
    failure_reasons: list = field(default_factory=list)

def query_prometheus(config: MonitorConfig, query: str) -> float | None:
    """执行 Prometheus 查询并返回单个值。"""
    try:
        resp = requests.get(
            f"{config.prometheus_url}/api/v1/query",
            params={"query": query},
            timeout=10,
        )
        data = resp.json()
        if data["status"] == "success" and data["data"]["result"]:
            return float(data["data"]["result"][0]["value"][1])
    except Exception as e:
        logger.error(f"Prometheus query failed: {e}")
    return None

def check_model_health(config: MonitorConfig) -> tuple[bool, list[RollbackReason]]:
    """综合检查模型的健康状态。"""
    reasons = []

    # 1. 检查错误率
    error_rate = query_prometheus(config, (
        f'sum(rate(revision_request_count'
        f'{{service_name="{config.isvc_name}",'
        f'response_code=~"5..",namespace="{config.namespace}"}}[5m]))'
        f' / sum(rate(revision_request_count'
        f'{{service_name="{config.isvc_name}",'
        f'namespace="{config.namespace}"}}[5m]))'
    ))
    if error_rate is not None and error_rate > config.max_error_rate:
        logger.warning(f"Error rate: {error_rate:.4f} > {config.max_error_rate}")
        reasons.append(RollbackReason.ERROR_RATE)

    # 2. 检查 p99 延迟
    latency = query_prometheus(config, (
        f'histogram_quantile(0.99, sum(rate('
        f'revision_request_latencies_bucket'
        f'{{service_name="{config.isvc_name}",'
        f'namespace="{config.namespace}"}}[5m])) by (le))'
    ))
    if latency is not None and latency > config.max_latency_p99_ms:
        logger.warning(f"p99 latency: {latency:.1f}ms > {config.max_latency_p99_ms}ms")
        reasons.append(RollbackReason.LATENCY)

    # 3. 检查预测分布变化（PSI）
    psi = query_prometheus(config, (
        f'ml_prediction_psi'
        f'{{model="{config.isvc_name}",namespace="{config.namespace}"}}'
    ))
    if psi is not None and psi > config.max_psi:
        logger.warning(f"PSI: {psi:.4f} > {config.max_psi}")
        reasons.append(RollbackReason.PREDICTION_DRIFT)

    is_healthy = len(reasons) == 0
    return is_healthy, reasons

def send_alert(config: MonitorConfig, message: str):
    """向 Slack 和 PagerDuty 发送告警。"""
    if config.slack_webhook:
        requests.post(config.slack_webhook, json={
            "text": f":rotating_light: [AB Deploy Monitor] {message}",
            "channel": "#ml-alerts",
        })
    logger.info(f"Alert: {message}")

def trigger_rollback(config: MonitorConfig, reasons: list[RollbackReason]):
    """执行自动回滚。"""
    import subprocess, json
    reason_str = ", ".join([r.value for r in reasons])
    logger.critical(f"Triggering rollback. Reasons: {reason_str}")

    # 将 KServe canaryTrafficPercent 修补为 0
    patch = json.dumps({
        "spec": {"predictor": {"canaryTrafficPercent": 0}}
    })
    subprocess.run([
        "kubectl", "patch", "inferenceservice", config.isvc_name,
        "-n", config.namespace,
        "--type=merge", f"-p={patch}",
    ], check=True)

    send_alert(config, f"Auto-rollback executed for {config.isvc_name}. Reasons: {reason_str}")

def run_monitor(config: MonitorConfig):
    """执行监控循环。"""
    state = MonitorState()
    logger.info(f"Starting AB deploy monitor for {config.isvc_name}")

    while True:
        is_healthy, reasons = check_model_health(config)
        state.total_checks += 1

        if is_healthy:
            state.consecutive_failures = 0
            state.failure_reasons.clear()
            if state.total_checks % 10 == 0:
                logger.info(f"Check #{state.total_checks}: Healthy")
        else:
            state.consecutive_failures += 1
            state.failure_reasons.extend(reasons)
            logger.warning(
                f"Check #{state.total_checks}: Unhealthy "
                f"({state.consecutive_failures}/{config.consecutive_failures_to_rollback})"
            )

            if state.consecutive_failures >= config.consecutive_failures_to_rollback:
                trigger_rollback(config, reasons)
                break

        time.sleep(config.check_interval_sec)

# 执行示例
if __name__ == "__main__":
    config = MonitorConfig(
        isvc_name="fraud-detection",
        namespace="ml-serving",
        slack_webhook="https://hooks.slack.com/services/xxx/yyy/zzz",
    )
    run_monitor(config)

"""
模型晋升流水线：金丝雀部署成功后自动切换为 champion。
处理 MLflow Alias 更新 + KServe 金丝雀清理 + 告警。
"""
import mlflow
from mlflow.tracking import MlflowClient
import subprocess
import json
from datetime import datetime

def promote_model_pipeline(
    model_name: str,
    new_version: int,
    isvc_name: str,
    namespace: str,
    ab_test_results: dict | None = None,
):
    """
    模型晋升的整体流水线。

    Args:
        model_name: MLflow 中注册的模型名称
        new_version: 要晋升的模型版本号
        isvc_name: KServe InferenceService 名称
        namespace: Kubernetes 命名空间
        ab_test_results: A/B 测试结果指标字典
    """
    client = MlflowClient()
    timestamp = datetime.utcnow().isoformat()

    # Step 1：备份当前 champion 信息
    try:
        old_champion = client.get_model_version_by_alias(
            name=model_name, alias="champion"
        )
        old_version = int(old_champion.version)
        print(f"Current champion: v{old_version}")

        # 为此前的 champion 记录元数据
        client.set_model_version_tag(
            name=model_name, version=old_version,
            key="demoted_at", value=timestamp,
        )
        client.set_model_version_tag(
            name=model_name, version=old_version,
            key="demoted_by", value=f"v{new_version}",
        )
        # 将此前的 champion 移动为 archived
        client.set_registered_model_alias(
            name=model_name, alias="archived", version=old_version,
        )
    except mlflow.exceptions.MlflowException:
        old_version = None
        print("No existing champion found")

    # Step 2：将新模型晋升为 champion
    client.set_registered_model_alias(
        name=model_name, alias="champion", version=new_version,
    )
    # 移除 challenger alias
    try:
        client.delete_registered_model_alias(
            name=model_name, alias="challenger",
        )
    except mlflow.exceptions.MlflowException:
        pass

    # 记录晋升元数据
    client.set_model_version_tag(
        name=model_name, version=new_version,
        key="promoted_to_champion_at", value=timestamp,
    )
    if ab_test_results:
        for k, v in ab_test_results.items():
            client.set_model_version_tag(
                name=model_name, version=new_version,
                key=f"ab_result_{k}", value=str(v),
            )

    # Step 3：在 KServe 中清理金丝雀配置
    # 移除 canaryTrafficPercent，让新模型接收 100% 流量
    patch = json.dumps({
        "spec": {
            "predictor": {
                "canaryTrafficPercent": 100
            }
        }
    })
    subprocess.run([
        "kubectl", "patch", "inferenceservice", isvc_name,
        "-n", namespace,
        "--type=merge", f"-p={patch}",
    ], check=True)

    print(f"Promotion complete: {model_name} v{new_version} is now champion")

    return {
        "model_name": model_name,
        "new_champion_version": new_version,
        "old_champion_version": old_version,
        "promoted_at": timestamp,
    }

# 执行示例
result = promote_model_pipeline(
    model_name="fraud-detection",
    new_version=3,
    isvc_name="fraud-detection",
    namespace="ml-serving",
    ab_test_results={
        "auc_improvement": 0.012,
        "latency_p99_delta_ms": -2.1,
        "error_rate_delta": -0.003,
        "test_duration_hours": 24,
        "traffic_percentage": 50,
    },
)