如何从Deepchecks自定义套件结果中提取失败检查及问题列

我需要从Deepchecks自定义漂移套件的运行结果里，提取出失败的检查项以及对应的具体问题列。比如我有两项失败检查：Feature Drift和Multivariate Drift，涉及的问题列包括col_1、col_5和col_30。

期望输出效果 #

1. 检查项'Feature Drift'失败，原因：'col_1'（漂移分数=0.9）、'col_5'（漂移分数=0.7）
2. 检查项'Multivariate Drift'失败，原因：'col_1'和'col_30'，domain_classifier_drift_score=0.85

我当前的代码 #

columns_metadata = {'cat_features' : categorical_cols, 'label':y_label_col } 
train_dataset = Dataset(df = train_df   , **columns_metadata)
test_dataset  = Dataset(df = test_df    , **columns_metadata )

custom_drift_suite = Suite('My_custom_drift_suite',
                         FeatureDrift().add_condition_drift_score_less_than( max_allowed_categorical_score=0.2, max_allowed_numeric_score=0.2),
                         # 移除原代码中无效注释内容
                         MultivariateDrift().add_condition_overall_drift_value_less_than(0.4), 
                         LabelDrift(),
                         NewCategoryTrainTest()
                        )
custom_suite_ans = custom_drift_suite.run(train_dataset = train_dataset, test_dataset = test_dataset)

解决方案代码 #

通过解析Deepchecks套件运行结果的conditions_results和value属性，可精准提取失败检查项及对应问题信息：

# 遍历套件运行结果中的所有检查项
for result in custom_suite_ans.results:
    check_name = result.check.name()
    # 筛选出条件未通过的检查项
    failed_conditions = [cond for cond in result.conditions_results if not cond.passed]
    if not failed_conditions:
        continue

    print(f"检查项'{check_name}'失败，原因：", end="")
    # 处理Feature Drift的失败详情
    if check_name == 'Feature Drift':
        drift_scores = result.value['drift_scores']
        # 筛选出漂移分数超过阈值的特征
        failed_features = [(col, score) for col, score in drift_scores.items() if score > 0.2]
        reason_parts = [f"'{col}'（漂移分数={score:.1f}）" for col, score in failed_features]
        print('、'.join(reason_parts))
    # 处理Multivariate Drift的失败详情
    elif check_name == 'Multivariate Drift':
        drift_score = result.value['domain_classifier_drift_score']
        # 获取漂移贡献度最高的前2个特征（可按需调整数量）
        top_features = result.value['feature_importances'].sort_values(ascending=False).head(2).index.tolist()
        reason = f"'{'和'.join(top_features)}'，domain_classifier_drift_score={drift_score:.2f}"
        print(reason)
    # 其他检查项的通用处理逻辑
    else:
        reason = '；'.join([cond.details for cond in failed_conditions])
        print(reason)

代码说明 #

• 先遍历所有检查项，仅聚焦条件未通过的项
• 针对Feature Drift：从result.value['drift_scores']中提取超过阈值的特征及其分数，按指定格式输出
• 针对Multivariate Drift：提取域分类器漂移分数，同时根据特征重要性排序获取关键影响特征
• 可根据实际使用的其他检查项（如LabelDrift、NewCategoryTrainTest）扩展对应的解析逻辑

内容的提问来源于stack exchange，提问作者Boris