模型评估Skill model-evaluation

---

name: 模型评估 description: 使用适当的指标和验证技术评估机器学习模型的性能、公平性和可靠性。涵盖训练调试、超参数调优和生产监控。触发关键词: model evaluation, metrics, accuracy, precision, recall, F1, F1-score, ROC, AUC, ROC-AUC, confusion matrix, cross-validation, k-fold, stratified, overfitting, underfitting, bias, variance, bias-variance tradeoff, hyperparameter, hyperparameter tuning, loss, loss function, metric, benchmark, benchmarking, model performance, classification metrics, regression metrics, RMSE, MSE, MAE, MAPE, R2, R-squared, train-test split, validation set, test set, hold-out, learning curve, validation curve, model selection, error analysis, residual analysis, ML testing, training issues, convergence, gradient, vanishing gradient, exploding gradient, training instability, LLM evaluation, language model evaluation, prompt engineering evaluation, A/B testing, champion-challenger, model monitoring, model drift, data drift, concept drift, model decay. allowed-tools: Read, Grep, Glob, Edit, Write, Bash

模型评估

概述

本技能专注于机器学习模型的全面评估，涵盖整个 ML 生命周期。它包括指标选择、验证策略、公平性评估、训练调试、超参数调优、LLM 评估、A/B 测试和生产监控，以确保模型质量和可靠性。

何时使用此技能

在以下情况下使用此技能：

• 选择和计算 ML 模型的适当评估指标
• 设计交叉验证策略和训练/测试分割
• 调试训练问题（过拟合、欠拟合、收敛问题）
• 调优超参数并验证模型改进
• 评估 LLM 和生成模型
• 在生产中进行 A/B 测试以比较模型
• 监控部署模型以检测漂移和退化
• 评估跨人口群体的模型公平性
• 分析错误模式和残差
• 创建评估报告和仪表板

相关技能和代理

何时升级：

• senior-software-engineer (Opus)：用于 ML 系统架构决策、模型选择策略、复杂评估管道设计、分布式评估基础设施、大规模基准测试、生产监控架构、对抗性鲁棒性评估、模型中毒检测、安全感知指标（使用 /security-audit 和 /threat-model 技能处理安全问题）

互补技能：

• 使用 /debugging 进行系统调试评估管道和指标计算问题
• 使用 /testing 进行单元测试评估代码和验证逻辑
• 使用 /data-validation 进行模型评估前的输入数据质量检查

说明

1. 定义评估标准

业务对齐：

• 确定业务目标和成功标准
• 将业务目标转化为 ML 指标
• 定义可接受的性能阈值
• 考虑不同类型错误的成本（假阳性 vs 假阴性）

指标选择：

• 分类：准确率、精确率、召回率、F1、ROC-AUC、PR-AUC
• 回归：MSE、RMSE、MAE、MAPE、R2、解释方差
• 排名：NDCG、MAP、MRR
• LLM：困惑度、BLEU、ROUGE、BERTScore、人工评估
• 针对领域特定需求的自定义指标

公平性要求：

• 识别受保护属性（种族、性别、年龄）
• 选择公平性定义（人口统计平等、均衡机会）
• 设置公平性约束和阈值

2. 设计评估策略

数据分割：

• 训练/验证/测试分割比率（例如，60/20/20）
• 针对类别不平衡的分层分割
• 针对时序数据的时间分割
• 基于组的分割以防止数据泄露

交叉验证：

• K 折交叉验证用于标准问题
• 分层 K 折交叉验证用于不平衡类别
• 时间序列分割用于时序数据
• 组 K 折交叉验证用于聚类数据
• 留一法用于小数据集

处理不平衡：

• 分层抽样
• 指标中的类别权重
• 重采样策略（SMOTE、下采样）
• 适当指标（F1、PR-AUC 代替准确率）

3. 进行评估

性能指标：

• 计算主要和次要指标
• 计算置信区间
• 与基线比较
• 统计显著性测试

错误分析：

• 混淆矩阵分析
• 每类性能细分
• 错误类型分类
• 困难示例挖掘

公平性评估：

• 分组指标比较
• 人口统计平等评估
• 均衡机会分析
• 差异影响测量

边缘案例测试：

• 边界条件验证
• 分布外检测
• 对抗性鲁棒性
• 极端输入的压力测试

4. 调试训练问题

过拟合检测：

• 训练与验证性能差距
• 学习曲线分析
• 验证指标平稳而训练改善
• 缓解：正则化、丢弃、早停、数据增强

欠拟合检测：

• 训练和验证性能均差
• 学习曲线未收敛
• 模型过于简单以应对问题复杂性
• 缓解：增加模型容量、特征工程、更长时间训练

收敛问题：

• 损失未减少
• 损失振荡或不稳定
• 梯度爆炸（损失变为 NaN）
• 梯度消失（损失保持恒定）
• 缓解：学习率调优、梯度裁剪、批归一化

学习率问题：

• 过高：训练不稳定，损失振荡
• 过低：训练过慢，陷入局部最小值
• 解决方案：学习率调度、预热、余弦退火

批大小效应：

• 小批：梯度噪声大，泛化差
• 大批：内存问题，尖锐最小值
• 通过实验找到最佳点

5. 超参数调优

搜索策略：

• 网格搜索：详尽但昂贵
• 随机搜索：高维空间覆盖更好
• 贝叶斯优化：采样高效
• 超带：自适应资源分配

关键超参数：

• 学习率（最关键）
• 批大小
• 正则化强度（L1、L2、丢弃）
• 网络架构（层数、单元数）
• 优化器选择（Adam、SGD、AdamW）

验证：

• 使用验证集进行超参数选择
• 绝不在测试集上调优
• 对小数据集考虑嵌套交叉验证

6. LLM 和生成模型评估

自动指标：

• 语言模型的困惑度
• 文本生成的 BLEU、ROUGE
• 语义相似度的 BERTScore
• QA 任务的精确匹配、F1

人工评估：

• 流畅性、连贯性、相关性
• 事实准确性
• 安全性和毒性
• 指令遵循

提示工程评估：

• 少样本与零样本比较
• 提示模板 A/B 测试
• 思维链效果
• 系统消息影响

LLM 作为评委：

• 使用更强的模型评估较弱的模型
• 配对比较用于排名
• 基于评分标准
• 校准人工判断

7. A/B 测试用于模型比较

实验设计：

• 随机流量分割（50/50 或 90/10）
• 最小样本量计算
• 统计功效分析
• 季节性持续时间规划

指标：

• 主要业务指标（转化、收入）
• 次要指标（延迟、用户满意度）
• 防护指标（错误率、偏见）
• 样本比率不匹配检查

分析：

• 统计显著性测试（t 检验、Mann-Whitney）
• 效应大小估计
• 置信区间
• 多重测试校正（Bonferroni）

决策标准：

• 主要指标改进阈值
• 防护指标无退化
• 足够的统计功效
• 业务案例验证

8. 生产模型监控

性能监控：

• 随时间跟踪关键指标
• 与基线/冠军模型比较
• 检测性能退化
• 阈值违规警报

数据漂移检测：

• 输入分布变化
• 特征统计跟踪
• KL 散度、KS 检验、PSI
• 协变量漂移检测

概念漂移检测：

• 模型预测分布变化
• 标签分布变化（当可用时）
• 性能指标趋势
• 对抗性验证

监控基础设施：

• 实时指标计算
• 可视化仪表板
• 警报和值班轮换
• 自动重训练触发

9. 报告和文档

评估报告结构：

• 执行摘要带关键发现
• 方法和实验设置
• 全面的指标表
• 错误分析和案例研究
• 公平性评估结果
• 建议和后续步骤

可视化：

• ROC 和 PR 曲线
• 混淆矩阵
• 学习曲线
• 残差图
• 公平性比较图表

版本控制：

• 模型版本和检查点
• 数据集版本和分割
• 超参数配置
• 评估代码和环境

最佳实践

一般原则

1. 指标与目标匹配：选择与业务目标对齐的指标，不仅仅是学术标准
2. 使用多个指标：单个指标无法讲述整个故事；使用互补指标
3. 适当验证：使用适当的交叉验证方案以避免过拟合验证集
4. 测试分布漂移：评估分布外数据以评估泛化
5. 检查偏见：部署前评估跨人口群体的公平性
6. 版本控制一切：跟踪模型、数据、指标和代码以实现可复现性
7. 监控生产：部署后持续跟踪模型性能

训练和调试

1. 从简单开始：在复杂模型之前从简单基线开始
2. 可视化学习：早期并经常绘制学习曲线
3. 增量调试：调试训练问题时一次更改一件事
4. 完整性检查：首先在小批量上过拟合以验证模型可以学习
5. 早停：使用基于验证的早停以防止过拟合
6. 梯度监控：跟踪梯度范数以检测梯度消失/爆炸

评估严谨性

1. 保留测试集：直到最终评估才接触测试集
2. 分层分割：用于不平衡数据集
3. 统计测试：用于模型比较的显著性测试
4. 错误分析：深入错误以理解失败模式
5. 时间验证：对于时间序列，仅在未来数据上验证

生产和监控

1. 影子模式：在切换流量前以影子模式部署新模型
2. 逐步推出：使用金丝雀部署或逐步百分比推出
3. 回滚计划：为性能退化设置自动回滚触发
4. 警报疲劳：设置有意义的警报阈值以避免噪音

示例

示例 1：分类模型评估

import numpy as np
import pandas as pd
from sklearn.metrics import (
    accuracy_score, precision_score, recall_score, f1_score,
    roc_auc_score, average_precision_score, confusion_matrix,
    classification_report, roc_curve, precision_recall_curve
)
import matplotlib.pyplot as plt

class ClassificationEvaluator:
    """全面的分类模型评估器。"""

    def __init__(self, y_true, y_pred, y_prob=None, class_names=None):
        self.y_true = y_true
        self.y_pred = y_pred
        self.y_prob = y_prob
        self.class_names = class_names or ['Negative', 'Positive']

    def compute_metrics(self) -> dict:
        """计算所有分类指标。"""
        metrics = {
            'accuracy': accuracy_score(self.y_true, self.y_pred),
            'precision': precision_score(self.y_true, self.y_pred, average='weighted'),
            'recall': recall_score(self.y_true, self.y_pred, average='weighted'),
            'f1': f1_score(self.y_true, self.y_pred, average='weighted'),
        }

        if self.y_prob is not None:
            metrics['roc_auc'] = roc_auc_score(self.y_true, self.y_prob)
            metrics['average_precision'] = average_precision_score(self.y_true, self.y_prob)

        return metrics

    def confusion_matrix_analysis(self) -> dict:
        """详细分析混淆矩阵。"""
        cm = confusion_matrix(self.y_true, self.y_pred)
        tn, fp, fn, tp = cm.ravel()

        return {
            'confusion_matrix': cm,
            'true_negatives': tn,
            'false_positives': fp,
            'false_negatives': fn,
            'true_positives': tp,
            'specificity': tn / (tn + fp),
            'sensitivity': tp / (tp + fn),
            'false_positive_rate': fp / (fp + tn),
            'false_negative_rate': fn / (fn + tp),
        }

    def plot_roc_curve(self, save_path=None):
        """绘制带 AUC 的 ROC 曲线。"""
        if self.y_prob is None:
            raise ValueError("ROC 曲线需要概率值")

        fpr, tpr, thresholds = roc_curve(self.y_true, self.y_prob)
        auc = roc_auc_score(self.y_true, self.y_prob)

        plt.figure(figsize=(8, 6))
        plt.plot(fpr, tpr, label=f'ROC (AUC = {auc:.3f})')
        plt.plot([0, 1], [0, 1], 'k--', label='随机')
        plt.xlabel('假阳性率')
        plt.ylabel('真阳性率')
        plt.title('接收者操作特征（ROC）曲线')
        plt.legend()
        plt.grid(True, alpha=0.3)

        if save_path:
            plt.savefig(save_path, dpi=150, bbox_inches='tight')
        plt.show()

    def generate_report(self) -> str:
        """生成全面的评估报告。"""
        metrics = self.compute_metrics()
        cm_analysis = self.confusion_matrix_analysis()

        report = f"""
# 分类模型评估报告

## 总体指标
| 指标 | 值 |
|--------|-------|
| 准确率 | {metrics['accuracy']:.4f} |
| 精确率 | {metrics['precision']:.4f} |
| 召回率 | {metrics['recall']:.4f} |
| F1 分数 | {metrics['f1']:.4f} |
| ROC AUC | {metrics.get('roc_auc', 'N/A'):.4f if isinstance(metrics.get('roc_auc'), float) else 'N/A'} |

## 混淆矩阵分析
| 指标 | 值 |
|--------|-------|
| 真阳性 | {cm_analysis['true_positives']} |
| 真阴性 | {cm_analysis['true_negatives']} |
| 假阳性 | {cm_analysis['false_positives']} |
| 假阴性 | {cm_analysis['false_negatives']} |
| 敏感性 | {cm_analysis['sensitivity']:.4f} |
| 特异性 | {cm_analysis['specificity']:.4f} |

## 详细分类报告
{classification_report(self.y_true, self.y_pred, target_names=self.class_names)}
"""
        return report

# 用法
evaluator = ClassificationEvaluator(y_true, y_pred, y_prob)
print(evaluator.generate_report())
evaluator.plot_roc_curve('roc_curve.png')

示例 2：回归模型评估

from sklearn.metrics import (
    mean_squared_error, mean_absolute_error, r2_score,
    mean_absolute_percentage_error, explained_variance_score
)
import numpy as np

class RegressionEvaluator:
    """全面的回归模型评估器。"""

    def __init__(self, y_true, y_pred):
        self.y_true = np.array(y_true)
        self.y_pred = np.array(y_pred)
        self.residuals = self.y_true - self.y_pred

    def compute_metrics(self) -> dict:
        """计算所有回归指标。"""
        mse = mean_squared_error(self.y_true, self.y_pred)

        return {
            'mse': mse,
            'rmse': np.sqrt(mse),
            'mae': mean_absolute_error(self.y_true, self.y_pred),
            'mape': mean_absolute_percentage_error(self.y_true, self.y_pred) * 100,
            'r2': r2_score(self.y_true, self.y_pred),
            'explained_variance': explained_variance_score(self.y_true, self.y_pred),
        }

    def residual_analysis(self) -> dict:
        """分析残差模式。"""
        return {
            'mean_residual': np.mean(self.residuals),
            'std_residual': np.std(self.residuals),
            'max_overestimate': np.min(self.residuals),
            'max_underestimate': np.max(self.residuals),
            'residual_skewness': self._skewness(self.residuals),
        }

    def _skewness(self, data):
        """计算偏度。"""
        n = len(data)
        mean = np.mean(data)
        std = np.std(data)
        return (n / ((n-1) * (n-2))) * np.sum(((data - mean) / std) ** 3)

    def plot_diagnostics(self, save_path=None):
        """绘制残差分析的诊断图。"""
        fig, axes = plt.subplots(2, 2, figsize=(12, 10))

        # 实际 vs 预测
        ax1 = axes[0, 0]
        ax1.scatter(self.y_true, self.y_pred, alpha=0.5)
        ax1.plot([self.y_true.min(), self.y_true.max()],
                 [self.y_true.min(), self.y_true.max()], 'r--')
        ax1.set_xlabel('实际')
        ax1.set_ylabel('预测')
        ax1.set_title('实际 vs 预测')

        # 残差 vs 预测
        ax2 = axes[0, 1]
        ax2.scatter(self.y_pred, self.residuals, alpha=0.5)
        ax2.axhline(y=0, color='r', linestyle='--')
        ax2.set_xlabel('预测')
        ax2.set_ylabel('残差')
        ax2.set_title('残差 vs 预测')

        # 残差直方图
        ax3 = axes[1, 0]
        ax3.hist(self.residuals, bins=30, edgecolor='black')
        ax3.set_xlabel('残差')
        ax3.set_ylabel('频率')
        ax3.set_title('残差分布')

        # Q-Q 图
        ax4 = axes[1, 1]
        from scipy import stats
        stats.probplot(self.residuals, dist="norm", plot=ax4)
        ax4.set_title('Q-Q 图')

        plt.tight_layout()
        if save_path:
            plt.savefig(save_path, dpi=150, bbox_inches='tight')
        plt.show()

示例 3：交叉验证策略

from sklearn.model_selection import (
    cross_val_score, StratifiedKFold, TimeSeriesSplit,
    GroupKFold, cross_validate
)

def evaluate_with_cv(model, X, y, cv_strategy='stratified', n_splits=5, groups=None):
    """
    使用适当的交叉验证策略评估模型。

    参数:
        model: Sklearn 兼容模型
        X: 特征
        y: 目标
        cv_strategy: 'stratified', 'timeseries', 'group', 或 'kfold'
        n_splits: 交叉验证折数
        groups: GroupKFold 的组标签

    返回:
        包含交叉验证结果的字典
    """
    # 选择交叉验证策略
    if cv_strategy == 'stratified':
        cv = StratifiedKFold(n_splits=n_splits, shuffle=True, random_state=42)
    elif cv_strategy == 'timeseries':
        cv = TimeSeriesSplit(n_splits=n_splits)
    elif cv_strategy == 'group':
        cv = GroupKFold(n_splits=n_splits)
    else:
        cv = n_splits

    # 定义评分指标
    scoring = {
        'accuracy': 'accuracy',
        'precision': 'precision_weighted',
        'recall': 'recall_weighted',
        'f1': 'f1_weighted',
        'roc_auc': 'roc_auc'
    }

    # 执行交叉验证
    cv_results = cross_validate(
        model, X, y,
        cv=cv,
        scoring=scoring,
        groups=groups,
        return_train_score=True,
        n_jobs=-1
    )

    # 总结结果
    summary = {}
    for metric in scoring.keys():
        test_scores = cv_results[f'test_{metric}']
        train_scores = cv_results[f'train_{metric}']
        summary[metric] = {
            'test_mean': np.mean(test_scores),
            'test_std': np.std(test_scores),
            'train_mean': np.mean(train_scores),
            'train_std': np.std(train_scores),
            'overfit_gap': np.mean(train_scores) - np.mean(test_scores)
        }

    return summary

# 用法示例
results = evaluate_with_cv(model, X, y, cv_strategy='stratified', n_splits=5)
for metric, values in results.items():
    print(f"{metric}: {values['test_mean']:.4f} (+/- {values['test_std']:.4f})")
    print(f"  过拟合差距: {values['overfit_gap']:.4f}")

示例 4：公平性评估

def evaluate_fairness(y_true, y_pred, sensitive_attr, favorable_label=1):
    """
    评估模型跨人口群体的公平性。

    参数:
        y_true: 真实标签
        y_pred: 预测标签
        sensitive_attr: 受保护属性值
        favorable_label: 有利结果标签

    返回:
        包含公平性指标的字典
    """
    groups = np.unique(sensitive_attr)
    results = {'group_metrics': {}}

    for group in groups:
        mask = sensitive_attr == group
        group_true = y_true[mask]
        group_pred = y_pred[mask]

        # 计算组特定指标
        tp = np.sum((group_true == favorable_label) & (group_pred == favorable_label))
        fp = np.sum((group_true != favorable_label) & (group_pred == favorable_label))
        fn = np.sum((group_true == favorable_label) & (group_pred != favorable_label))
        tn = np.sum((group_true != favorable_label) & (group_pred != favorable_label))

        results['group_metrics'][group] = {
            'selection_rate': np.mean(group_pred == favorable_label),
            'tpr': tp / (tp + fn) if (tp + fn) > 0 else 0,
            'fpr': fp / (fp + tn) if (fp + tn) > 0 else 0,
            'accuracy': np.mean(group_true == group_pred),
            'size': len(group_true)
        }

    # 计算公平性指标
    selection_rates = [m['selection_rate'] for m in results['group_metrics'].values()]
    tprs = [m['tpr'] for m in results['group_metrics'].values()]
    fprs = [m['fpr'] for m in results['group_metrics'].values()]

    results['fairness_metrics'] = {
        'demographic_parity_diff': max(selection_rates) - min(selection_rates),
        'equalized_odds_tpr_diff': max(tprs) - min(tprs),
        'equalized_odds_fpr_diff': max(fprs) - min(fprs),
    }

    return results

示例 5：训练调试与学习曲线

import matplotlib.pyplot as plt
from sklearn.model_selection import learning_curve

def plot_learning_curves(model, X, y, cv=5, train_sizes=np.linspace(0.1, 1.0, 10)):
    """
    绘制学习曲线以诊断过拟合/欠拟合。

    参数:
        model: Sklearn 兼容模型
        X: 特征
        y: 目标
        cv: 交叉验证折数
        train_sizes: 训练集大小分数数组
    """
    train_sizes, train_scores, val_scores = learning_curve(
        model, X, y,
        cv=cv,
        train_sizes=train_sizes,
        scoring='accuracy',
        n_jobs=-1
    )

    train_mean = np.mean(train_scores, axis=1)
    train_std = np.std(train_scores, axis=1)
    val_mean = np.mean(val_scores, axis=1)
    val_std = np.std(val_scores, axis=1)

    plt.figure(figsize=(10, 6))
    plt.plot(train_sizes, train_mean, label='训练分数', color='blue', marker='o')
    plt.fill_between(train_sizes, train_mean - train_std, train_mean + train_std, alpha=0.15, color='blue')
    plt.plot(train_sizes, val_mean, label='验证分数', color='red', marker='o')
    plt.fill_between(train_sizes, val_mean - val_std, val_mean + val_std, alpha=0.15, color='red')

    plt.xlabel('训练集大小')
    plt.ylabel('准确率')
    plt.title('学习曲线')
    plt.legend(loc='lower right')
    plt.grid(True, alpha=0.3)

    # 添加诊断注释
    final_gap = train_mean[-1] - val_mean[-1]
    if final_gap > 0.1:
        plt.text(0.5, 0.05, '高度过拟合：训练和验证之间差距大',
                 transform=plt.gca().transAxes, color='red', fontweight='bold')
    elif val_mean[-1] < 0.7:
        plt.text(0.5, 0.05, '欠拟合：训练和验证分数均低',
                 transform=plt.gca().transAxes, color='orange', fontweight='bold')

    plt.tight_layout()
    plt.show()

    return {
        'final_train_score': train_mean[-1],
        'final_val_score': val_mean[-1],
        'overfit_gap': final_gap
    }

示例 6：LLM 评估

from typing import List, Dict
import openai

def evaluate_llm_generation(
    prompts: List[str],
    references: List[str],
    model: str,
    judge_model: str = "gpt-4"
) -> Dict:
    """
    使用 LLM 作为评委评估 LLM 生成质量。

    参数:
        prompts: 输入提示
        references: 参考输出（如可用）
        model: 要评估的模型
        judge_model: 用作评委的模型

    返回:
        包含评估分数的字典
    """
    results = []

    for prompt, reference in zip(prompts, references):
        # 生成响应
        response = openai.ChatCompletion.create(
            model=model,
            messages=[{"role": "user", "content": prompt}]
        )
        generation = response.choices[0].message.content

        # LLM 作为评委评估
        judge_prompt = f"""评估以下 AI 生成响应，按 1-5 分打分：
1. 准确性：信息是否正确？
2. 相关性：是否回应了提示？
3. 流畅性：是否写得好且连贯？
4. 有帮助性：对用户是否有用？

提示： {prompt}
参考（如可用）： {reference}
响应： {generation}

以 JSON 格式提供分数： {{"accuracy": X, "relevance": X, "fluency": X, "helpfulness": X, "overall": X}}
"""

        judge_response = openai.ChatCompletion.create(
            model=judge_model,
            messages=[{"role": "user", "content": judge_prompt}]
        )

        scores = json.loads(judge_response.choices[0].message.content)
        results.append({
            'prompt': prompt,
            'generation': generation,
            'scores': scores
        })

    # 聚合分数
    avg_scores = {}
    for key in ['accuracy', 'relevance', 'fluency', 'helpfulness', 'overall']:
        avg_scores[key] = np.mean([r['scores'][key] for r in results])

    return {
        'individual_results': results,
        'average_scores': avg_scores
    }

示例 7：A/B 测试分析

from scipy import stats

def analyze_ab_test(control_metric: np.ndarray, treatment_metric: np.ndarray, alpha: float = 0.05):
    """
    使用统计显著性测试分析 A/B 测试结果。

    参数:
        control_metric: 对照组指标值
        treatment_metric: 处理组指标值
        alpha: 显著性水平

    返回:
        包含测试结果的字典
    """
    # 描述性统计
    control_mean = np.mean(control_metric)
    treatment_mean = np.mean(treatment_metric)
    relative_lift = (treatment_mean - control_mean) / control_mean * 100

    # 统计测试
    t_stat, p_value = stats.ttest_ind(treatment_metric, control_metric)
    is_significant = p_value < alpha

    # 效应大小（Cohen's d）
    pooled_std = np.sqrt((np.std(control_metric)**2 + np.std(treatment_metric)**2) / 2)
    cohens_d = (treatment_mean - control_mean) / pooled_std

    # 置信区间
    ci = stats.t.interval(
        confidence=1-alpha,
        df=len(control_metric) + len(treatment_metric) - 2,
        loc=treatment_mean - control_mean,
        scale=stats.sem(np.concatenate([control_metric, treatment_metric]))
    )

    return {
        'control_mean': control_mean,
        'treatment_mean': treatment_mean,
        'relative_lift_pct': relative_lift,
        'p_value': p_value,
        'is_significant': is_significant,
        'cohens_d': cohens_d,
        'confidence_interval': ci,
        'recommendation': 'LAUNCH' if is_significant and relative_lift > 0 else 'DO NOT LAUNCH'
    }

# 用法
results = analyze_ab_test(control_conversions, treatment_conversions)
print(f"相对提升: {results['relative_lift_pct']:.2f}%")
print(f"P 值: {results['p_value']:.4f}")
print(f"推荐: {results['recommendation']}")

示例 8：生产模型监控

from scipy.stats import ks_2samp
import pandas as pd

class ModelMonitor:
    """监控部署模型以检测漂移和退化。"""

    def __init__(self, baseline_data: pd.DataFrame, baseline_predictions: np.ndarray):
        self.baseline_data = baseline_data
        self.baseline_predictions = baseline_predictions

    def detect_data_drift(self, current_data: pd.DataFrame, threshold: float = 0.05) -> Dict:
        """使用 KS 检验检测特征分布漂移。"""
        drift_results = {}

        for col in self.baseline_data.columns:
            if pd.api.types.is_numeric_dtype(self.baseline_data[col]):
                statistic, p_value = ks_2samp(
                    self.baseline_data[col].dropna(),
                    current_data[col].dropna()
                )
                drift_results[col] = {
                    'ks_statistic': statistic,
                    'p_value': p_value,
                    'drift_detected': p_value < threshold
                }

        return drift_results

    def detect_prediction_drift(self, current_predictions: np.ndarray, threshold: float = 0.05) -> Dict:
        """检测预测分布漂移。"""
        statistic, p_value = ks_2samp(self.baseline_predictions, current_predictions)

        return {
            'ks_statistic': statistic,
            'p_value': p_value,
            'drift_detected': p_value < threshold,
            'baseline_mean': np.mean(self.baseline_predictions),
            'current_mean': np.mean(current_predictions),
            'mean_shift': np.mean(current_predictions) - np.mean(self.baseline_predictions)
        }

    def performance_degradation_check(
        self,
        current_metric: float,
        baseline_metric: float,
        threshold_pct: float = 5.0
    ) -> Dict:
        """检查性能退化。"""
        degradation_pct = (baseline_metric - current_metric) / baseline_metric * 100

        return {
            'baseline_metric': baseline_metric,
            'current_metric': current_metric,
            'degradation_pct': degradation_pct,
            'alert': degradation_pct > threshold_pct,
            'recommendation': 'RETRAIN MODEL' if degradation_pct > threshold_pct else 'OK'
        }

# 用法
monitor = ModelMonitor(baseline_df, baseline_preds)
drift_check = monitor.detect_data_drift(current_df)
pred_drift = monitor.detect_prediction_drift(current_preds)
perf_check = monitor.performance_degradation_check(current_accuracy, baseline_accuracy)

常见陷阱

1. 测试集污染：绝不要使用测试集进行超参数调优或模型选择
2. 数据泄露：确保验证/测试数据不会泄露到训练中（时序排序、组分隔）
3. 错误指标选择：在不平衡数据集中使用准确率，不考虑业务成本
4. 忽略置信区间：没有不确定性的点估计可能具有误导性
5. 多重比较：测试多个假设时不校正 p 值
6. 幸存者偏差：仅在成功案例上评估，忽略失败
7. 过拟合验证集：反复在验证集上调优实际上使其成为第二个训练集
8. 忽略公平性：部署未评估公平性的模型可能造成伤害
9. 无基线：不与简单基线比较（随机、多数类、线性模型）
10. 生产训练偏差：评估设置与生产环境不匹配

附加资源

• 指标：Scikit-learn 指标文档，Hugging Face 评估库
• 公平性：AI Fairness 360，Fairlearn
• LLM 评估：HELM，lm-evaluation-harness，BIG-bench
• A/B 测试：Evan Miller 的 A/B 测试工具，实验平台文档
• 监控：Evidently AI，WhyLabs，Fiddler