针对不同类型，高风险类拒打概率增加，低风险类增加提问可能¶

18 Class-Aware Safety Agent¶

这一份 notebook 的目标是做第一版 按类别分流的安全 agent。

前面的类别分析已经说明：

  • 有些类别可能从提问中受益
  • 有些类别会被提问明显带偏
  • 如果所有类别都用同一套提问/拒答策略，整体效果会被平均掉

所以这一版我们不再“一刀切”，而是先按当前实验现象把 7 类分成 3 组：

  • potential_benefit: nv, mel
  • high_risk: bkl, df, akiec
  • neutral: bcc, vasc

然后为不同组设置不同的策略强度。

In [1]:
from pathlib import Path
from datetime import datetime
import ast
import json

import numpy as np
import pandas as pd
from sklearn.metrics import accuracy_score, balanced_accuracy_score, f1_score

PROJECT_ROOT = Path('/Users/applesues01/Documents/Medical_Agent')
DATA_DIR = PROJECT_ROOT / 'data' / 'HAM10000'
SPLIT_DIR = DATA_DIR / 'splits'
SUPPORT_DIR = PROJECT_ROOT / 'supports'

BASELINE_RESULTS_PATH = SUPPORT_DIR / 'baseline_results.csv'
MODEL_REGISTRY_PATH = SUPPORT_DIR / '2026-08-03_194704_all_saved_metadata_models.csv'
IMAGE_ONLY_CASE_PATH = SUPPORT_DIR / 'image_only_case_level_predictions.csv'
VALIDATED_AGENT_PATH = SUPPORT_DIR / '2026-08-03_185733_validated_agent_comparison.csv'

print(BASELINE_RESULTS_PATH)
print(MODEL_REGISTRY_PATH)
print(IMAGE_ONLY_CASE_PATH)
print(VALIDATED_AGENT_PATH)

/Users/applesues01/Documents/Medical_Agent/supports/baseline_results.csv
/Users/applesues01/Documents/Medical_Agent/supports/2026-08-03_194704_all_saved_metadata_models.csv
/Users/applesues01/Documents/Medical_Agent/supports/image_only_case_level_predictions.csv
/Users/applesues01/Documents/Medical_Agent/supports/2026-08-03_185733_validated_agent_comparison.csv

1. 读取数据和当前模型结果¶

In [2]:
test_df = pd.read_csv(SPLIT_DIR / 'test.csv')
baseline_df = pd.read_csv(BASELINE_RESULTS_PATH)
model_registry_df = pd.read_csv(MODEL_REGISTRY_PATH)
image_only_case_df = pd.read_csv(IMAGE_ONLY_CASE_PATH)
validated_agent_df = pd.read_csv(VALIDATED_AGENT_PATH)

image_only_row = baseline_df[baseline_df['Method'] == 'Image Only'].iloc[0]

policy_model_scores = {
    'image_only': {'macro_f1': float(image_only_row['Macro-F1'])},
}

report_model_scores = {
    'image_only': {
        'accuracy': float(image_only_row['Accuracy']),
        'balanced_accuracy': float(image_only_row['Balanced Accuracy']),
        'macro_f1': float(image_only_row['Macro-F1']),
    }
}

for _, row in model_registry_df.iterrows():
    key = str(row['Method'])
    policy_model_scores[key] = {'macro_f1': float(row['Best Val Macro-F1'])}
    report_model_scores[key] = {
        'accuracy': float(row['Test Accuracy']),
        'balanced_accuracy': float(row['Test Balanced Accuracy']),
        'macro_f1': float(row['Test Macro-F1']),
    }

case_df = test_df.merge(image_only_case_df, on=['image_id'], how='left')
len(case_df)
Out[2]:
1481

2. 定义类别分组¶

In [3]:
CLASS_GROUPS = {
    'potential_benefit': ['nv', 'mel'],
    'high_risk': ['bkl', 'df', 'akiec'],
    'neutral': ['bcc', 'vasc'],
}

def get_class_group(pred_cls):
    for group_name, classes in CLASS_GROUPS.items():
        if pred_cls in classes:
            return group_name
    return 'neutral'

CLASS_GROUPS
Out[3]:
{'potential_benefit': ['nv', 'mel'],
 'high_risk': ['bkl', 'df', 'akiec'],
 'neutral': ['bcc', 'vasc']}

3. 提取 image-only 的病例级结构¶

这里继续使用 image-only 的预测分布来判断当前病例更像哪一组。

In [4]:
prob_cols = [c for c in case_df.columns if c.startswith('prob_')]
class_names = [c.replace('prob_', '') for c in prob_cols]

def extract_case_structure(row):
    probs = np.array([row[c] for c in prob_cols], dtype=float)
    order = np.argsort(-probs)
    top1_idx = int(order[0])
    top2_idx = int(order[1])
    top1_cls = class_names[top1_idx]
    top2_cls = class_names[top2_idx]
    top1_prob = float(probs[top1_idx])
    top2_prob = float(probs[top2_idx])
    margin = top1_prob - top2_prob
    entropy = float(-(probs * np.log(probs + 1e-12)).sum())
    return pd.Series({
        'top1_cls': top1_cls,
        'top2_cls': top2_cls,
        'top1_prob': top1_prob,
        'top2_prob': top2_prob,
        'margin': margin,
        'entropy': entropy,
        'predicted_group': get_class_group(top1_cls),
    })

case_df = pd.concat([case_df, case_df.apply(extract_case_structure, axis=1)], axis=1)
case_df[['image_id', 'dx', 'top1_cls', 'top2_cls', 'predicted_group', 'margin', 'entropy']].head()
Out[4]:
image_id dx top1_cls top2_cls predicted_group margin entropy
0 ISIC_0025837 bkl bkl mel high_risk 0.953966 0.150215
1 ISIC_0025209 bkl bkl akiec high_risk 0.186130 1.487569
2 ISIC_0029161 bkl bkl nv high_risk 0.665526 0.669199
3 ISIC_0026273 bkl bkl mel high_risk 0.707970 0.723783
4 ISIC_0025819 bkl bkl nv high_risk 0.969386 0.123900

4. 定义按组策略¶

这一版先用最简单、最可解释的组策略：

potential_benefit¶

  • 允许积极提问
  • 默认顺序：age -> location -> sex

high_risk¶

  • 更保守
  • 只允许问相对更稳的 age
  • 如果不够，就直接诊断（后面可以再升级成拒答）

neutral¶

  • 中间策略
  • 默认顺序：age -> sex -> location
In [5]:
GROUP_POLICY = {
    'potential_benefit': ['age', 'location', 'sex'],
    'high_risk': ['age'],
    'neutral': ['age', 'sex', 'location'],
}

GROUP_POLICY
Out[5]:
{'potential_benefit': ['age', 'location', 'sex'],
 'high_risk': ['age'],
 'neutral': ['age', 'sex', 'location']}

5. 状态函数¶

In [6]:
def build_initial_state(row):
    return {
        'image_id': row['image_id'],
        'true_label': row['dx'],
        'known_metadata': {},
        'asked_questions': [],
        'done': False,
        'top1_cls': row['top1_cls'],
        'top2_cls': row['top2_cls'],
        'margin': float(row['margin']),
        'entropy': float(row['entropy']),
        'predicted_group': row['predicted_group'],
    }


def ask_question(state, row, question):
    new_state = dict(state)
    new_state['known_metadata'] = dict(state['known_metadata'])
    new_state['asked_questions'] = list(state['asked_questions'])

    if question == 'age':
        new_state['known_metadata']['age'] = row['age']
    elif question == 'sex':
        new_state['known_metadata']['sex'] = row['sex']
    elif question == 'location':
        new_state['known_metadata']['location'] = row['localization']

    new_state['asked_questions'].append(question)
    return new_state


def get_model_key_from_known_set(known_set):
    if known_set == set():
        return 'image_only'
    if known_set == {'age'}:
        return 'image_age'
    if known_set == {'sex'}:
        return 'image_sex'
    if known_set == {'location'}:
        return 'image_location'
    if known_set == {'age', 'sex'}:
        return 'image_age_sex'
    if known_set == {'age', 'location'}:
        return 'image_age_location'
    if known_set == {'sex', 'location'}:
        return 'image_sex_location'
    if known_set == {'age', 'sex', 'location'}:
        return 'image_all_metadata'
    raise ValueError(f'Unknown known_set: {known_set}')


def get_model_key_from_state(state):
    return get_model_key_from_known_set(set(state['known_metadata'].keys()))


def score_state(state):
    model_key = get_model_key_from_state(state)
    return policy_model_scores[model_key]['macro_f1']

6. 定义 class-aware policy¶

In [7]:
def class_aware_policy(state, max_questions=2):
    if len(state['asked_questions']) >= max_questions:
        return 'diagnose'

    group_name = state['predicted_group']
    preferred_order = GROUP_POLICY[group_name]

    for question in preferred_order:
        if question not in state['asked_questions']:
            return question

    return 'diagnose'

7. 先看几个病例¶

In [8]:
def run_class_aware_episode(row, max_questions=2):
    state = build_initial_state(row)
    trajectory = []

    while not state['done']:
        action = class_aware_policy(state, max_questions=max_questions)

        trajectory.append({
            'step': len(trajectory),
            'action': action,
            'predicted_group': state['predicted_group'],
            'model_key_before_action': get_model_key_from_state(state),
            'score_before_action': score_state(state),
            'known_metadata_before_action': dict(state['known_metadata']),
        })

        if action == 'diagnose':
            state['done'] = True
            state['final_action'] = 'diagnose'
            state['final_model_key'] = get_model_key_from_state(state)
            state['final_score'] = score_state(state)
            break

        state = ask_question(state, row, action)

    return trajectory, state
In [9]:
trajectory, final_state = run_class_aware_episode(case_df.iloc[0], max_questions=2)
trajectory, final_state
Out[9]:
([{'step': 0,
   'action': 'age',
   'predicted_group': 'high_risk',
   'model_key_before_action': 'image_only',
   'score_before_action': 0.6083234281767498,
   'known_metadata_before_action': {}},
  {'step': 1,
   'action': 'diagnose',
   'predicted_group': 'high_risk',
   'model_key_before_action': 'image_age',
   'score_before_action': 0.5926280556025649,
   'known_metadata_before_action': {'age': np.float64(70.0)}}],
 {'image_id': 'ISIC_0025837',
  'true_label': 'bkl',
  'known_metadata': {'age': np.float64(70.0)},
  'asked_questions': ['age'],
  'done': True,
  'top1_cls': 'bkl',
  'top2_cls': 'mel',
  'margin': 0.9539660867303611,
  'entropy': 0.15021531890606907,
  'predicted_group': 'high_risk',
  'final_action': 'diagnose',
  'final_model_key': 'image_age',
  'final_score': 0.5926280556025649})

8. 全测试集评估¶

In [10]:
def evaluate_class_aware_agent(max_questions=2):
    records = []
    for idx in range(len(case_df)):
        row = case_df.iloc[idx]
        trajectory, final_state = run_class_aware_episode(row, max_questions=max_questions)
        records.append({
            'image_id': row['image_id'],
            'true_label': row['dx'],
            'top1_cls': row['top1_cls'],
            'predicted_group': row['predicted_group'],
            'asked_questions': list(final_state['asked_questions']),
            'num_questions': len(final_state['asked_questions']),
            'final_model_key': final_state['final_model_key'],
            'final_policy_score': final_state['final_score'],
        })

    cases_df = pd.DataFrame(records)
    summary_df = cases_df.groupby('final_model_key').size().reset_index(name='count')
    summary_df['ratio'] = summary_df['count'] / len(cases_df)
    summary_df['test_accuracy'] = summary_df['final_model_key'].map(lambda k: report_model_scores[k]['accuracy'])
    summary_df['test_balanced_accuracy'] = summary_df['final_model_key'].map(lambda k: report_model_scores[k]['balanced_accuracy'])
    summary_df['test_macro_f1'] = summary_df['final_model_key'].map(lambda k: report_model_scores[k]['macro_f1'])

    result = {
        'agent_name': 'class_aware_safety_agent',
        'max_questions': max_questions,
        'avg_questions': float(cases_df['num_questions'].mean()),
        'expected_accuracy': float((summary_df['ratio'] * summary_df['test_accuracy']).sum()),
        'expected_balanced_accuracy': float((summary_df['ratio'] * summary_df['test_balanced_accuracy']).sum()),
        'expected_macro_f1': float((summary_df['ratio'] * summary_df['test_macro_f1']).sum()),
    }
    return cases_df, summary_df, result
In [11]:
class_aware_cases_df, class_aware_summary_df, class_aware_result = evaluate_class_aware_agent(max_questions=2)
class_aware_result
Out[11]:
{'agent_name': 'class_aware_safety_agent',
 'max_questions': 2,
 'avg_questions': 1.811613774476705,
 'expected_accuracy': 0.7782225543355609,
 'expected_balanced_accuracy': 0.579323177037876,
 'expected_macro_f1': 0.5825781552298765}

9. 和已有 baseline 对比¶

In [12]:
comparison_df = pd.concat([
    validated_agent_df,
    pd.DataFrame([class_aware_result])
], ignore_index=True)
comparison_df
Out[12]:
agent_name max_questions avg_questions expected_accuracy expected_balanced_accuracy expected_macro_f1 threshold
0 fixed_order 2 2.000000 0.793383 0.582281 0.581666 NaN
1 uncertainty 2 2.000000 0.793383 0.582281 0.581666 0.605
2 lookahead 2 2.000000 0.792708 0.598458 0.596497 NaN
3 class_aware_safety_agent 2 1.811614 0.778223 0.579323 0.582578 NaN

10. 保存结果¶

In [13]:
timestamp = datetime.now().strftime('%Y-%m-%d_%H%M%S')

cases_path = SUPPORT_DIR / f'{timestamp}_class_aware_agent_cases.csv'
summary_path = SUPPORT_DIR / f'{timestamp}_class_aware_agent_summary.csv'
result_path = SUPPORT_DIR / f'{timestamp}_class_aware_agent_result.json'
comparison_path = SUPPORT_DIR / f'{timestamp}_class_aware_agent_comparison.csv'

class_aware_cases_df.to_csv(cases_path, index=False)
class_aware_summary_df.to_csv(summary_path, index=False)
comparison_df.to_csv(comparison_path, index=False)

with open(result_path, 'w', encoding='utf-8') as f:
    json.dump(class_aware_result, f, ensure_ascii=False, indent=2)

print(cases_path)
print(summary_path)
print(result_path)
print(comparison_path)

/Users/applesues01/Documents/Medical_Agent/supports/2026-08-04_194420_class_aware_agent_cases.csv
/Users/applesues01/Documents/Medical_Agent/supports/2026-08-04_194420_class_aware_agent_summary.csv
/Users/applesues01/Documents/Medical_Agent/supports/2026-08-04_194420_class_aware_agent_result.json
/Users/applesues01/Documents/Medical_Agent/supports/2026-08-04_194420_class_aware_agent_comparison.csv

论文里怎么讲会比较好

这版结果最适合被写成：

A preliminary class-aware policy reduced the average number of questions, but did not yield a substantial improvement in overall Macro-F1, suggesting that coarse class grouping alone is insufficient for effective adaptive questioning.

中文就是： 初步的类别感知策略降低了平均提问次数，但未能带来显著的整体 Macro-F1 提升，这说明仅依赖粗粒度类别分组还不足以支持有效的自适应提问。

这个表述是正面的，因为它不是“失败”，而是明确告诉我们： 单靠粗分组不够，下一步要做更细的类别条件化决策，或者把类别感知和安全门结合起来。