リスク判定

「画像を分類し、所見を生成した。次は、どれだけ緊急に対応が必要かを判定する。」

田中VPoEが強調する。

「リスクの高いケースを見逃すことは許されない。分類確信度、異常の重症度、緊急性。複数の要素を総合してリスクレベルを判定し、適切なアクションにつなげる。」

リスクレベルの定義

レベル	条件	アクション	対応時間
Critical	重度異常 + 高確信度	即時通知 + 専門家レビュー	1時間以内
High	中度異常 or 低確信度の重度	優先レビュー	4時間以内
Medium	軽度異常	通常レビュー	24時間以内
Low	異常なし + 高確信度	定期レビュー	1週間以内

リスク判定の実装

@tool
def assess_risk(classification: dict, findings: str) -> dict:
    """リスクレベルを判定する"""

    confidence = classification['confidence']
    predicted_class = classification['class']

    # 重症度マッピング
    severity_map = {
        'normal': 0,
        'mild_disease': 1,
        'moderate_disease': 2,
        'severe_disease': 3,
    }
    severity = severity_map.get(predicted_class, 1)

    # リスクスコアの計算
    # 高い重症度 + 高い確信度 = 高リスク
    # 高い重症度 + 低い確信度 = 高リスク（不確実だが安全側に）
    # 低い重症度 + 高い確信度 = 低リスク
    risk_score = severity * 0.6 + (1 - confidence) * 0.4 * 3

    # リスクレベルの判定
    if risk_score >= 2.5 or (severity >= 3 and confidence >= 0.7):
        risk_level = 'critical'
        urgency = '即時対応が必要'
        recommendation = '専門家による緊急レビューを実施してください'
    elif risk_score >= 1.5 or (severity >= 2) or (confidence < 0.5):
        risk_level = 'high'
        urgency = '優先的な確認が必要'
        recommendation = '24時間以内に専門家レビューを実施してください'
    elif risk_score >= 0.8:
        risk_level = 'medium'
        urgency = '通常の確認で対応'
        recommendation = '定期レビューで確認してください'
    else:
        risk_level = 'low'
        urgency = '経過観察'
        recommendation = '次回の定期検査まで経過観察'

    return {
        'risk_level': risk_level,
        'risk_score': round(risk_score, 2),
        'urgency': urgency,
        'recommendation': recommendation,
        'factors': {
            'severity': severity,
            'confidence': confidence,
            'uncertainty_risk': confidence < 0.7,
        },
    }

リスクに応じた通知

class RiskNotificationManager:
    """リスクレベルに応じた通知管理"""

    def notify(self, risk_result, case_info):
        """リスクレベルに応じて通知"""
        level = risk_result['risk_level']

        if level == 'critical':
            self._send_emergency_alert(risk_result, case_info)
        elif level == 'high':
            self._send_priority_alert(risk_result, case_info)
        elif level == 'medium':
            self._add_to_review_queue(risk_result, case_info)
        else:
            self._log_for_routine_check(risk_result, case_info)

    def _send_emergency_alert(self, risk_result, case_info):
        """緊急アラート: Slack + メール + SMS"""
        notification = {
            'type': 'emergency',
            'channels': ['slack', 'email', 'sms'],
            'case_id': case_info['id'],
            'risk_level': risk_result['risk_level'],
            'summary': risk_result['recommendation'],
        }
        return notification

    def _send_priority_alert(self, risk_result, case_info):
        """優先アラート: Slack + メール"""
        return {'type': 'priority', 'channels': ['slack', 'email']}

    def _add_to_review_queue(self, risk_result, case_info):
        """レビューキューに追加"""
        return {'type': 'queue', 'channels': ['dashboard']}

    def _log_for_routine_check(self, risk_result, case_info):
        """定期チェック用にログ"""
        return {'type': 'routine', 'channels': ['log']}

まとめ

項目	ポイント
リスクスコア	重症度 × 0.6 + 不確実性 × 0.4 で算出
安全側設計	低確信度は自動的にリスクを引き上げ
4段階	Critical/High/Medium/Lowで対応を分離
通知	リスクに応じてSlack/メール/SMS/ダッシュボード

チェックリスト

リスクスコアの算出方法を説明できる
4段階のリスクレベルと対応を定義できる
安全側に倒す設計の理由を説明できる
リスクに応じた通知設計ができる

次のステップへ

リスク判定を理解した。次は演習で診断支援エージェントを構築しよう。

推定読了時間: 30分