用 AI Agent 动态调度放射科工作流，终结"挑拣病例"顽疾

"""
radiology_agent_scheduler.py
最小放射科智能调度 Agent —— 可直接运行

依赖：pip install numpy
运行：python radiology_agent_scheduler.py
"""

import numpy as np
from dataclasses import dataclass, field
from typing import List
from datetime import datetime, timedelta

# ── 数据模型 ──────────────────────────────────────────

@dataclass
class Radiologist:
    name: str
    specialties: List[str]          # e.g. ["neuro", "chest"]
    shift_start: datetime
    cases_read_today: int = 0
    total_read_minutes_today: float = 0.0

    @property
    def fatigue_score(self) -> float:
        """0~1，越高越疲劳；超过 6 小时线性增长，上限 1.0"""
        hours = self.total_read_minutes_today / 60.0
        return min(1.0, max(0.0, (hours - 6.0) / 4.0)) if hours > 6 else 0.0

    @property
    def current_load(self) -> int:
        return self.cases_read_today


@dataclass
class Study:
    study_id: str
    modality: str                   # e.g. "CT", "MRI"
    body_part: str                  # e.g. "head", "chest", "spine"
    complexity: float               # 0~1，模型预测或人工标注
    priority_label: str             # "stat", "urgent", "routine"
    rvu: float                      # 相对价值单位
    wait_minutes: float = 0.0       # 已等待时长


# ── Agent 决策核心 ────────────────────────────────────

def specialty_match(radiologist: Radiologist, study: Study) -> float:
    """专长匹配度：body_part 在 specialties 里得 1.0，否则 0.3"""
    return 1.0 if study.body_part in radiologist.specialties else 0.3


def urgency_weight(study: Study) -> float:
    """紧急度加权：stat=3.0, urgent=2.0, routine=1.0，再加等待惩罚"""
    base = {"stat": 3.0, "urgent": 2.0, "routine": 1.0}.get(study.priority_label, 1.0)
    # 等待超过 60 分钟，每 30 分钟加 0.5 紧急度
    wait_penalty = max(0.0, (study.wait_minutes - 60.0) / 30.0) * 0.5
    return base + wait_penalty


def assignment_score(radiologist: Radiologist, study: Study) -> float:
    """
    综合评分 = 专长匹配 × 紧急度 × (1 - 疲劳惩罚) × 复杂度调节
    高分 = 更应该分配给该医生
    """
    spec = specialty_match(radiologist, study)
    urg = urgency_weight(study)
    fatigue_factor = 1.0 - 0.4 * radiologist.fatigue_score   # 疲劳最多扣 40%
    # 复杂病例更应分配给专长匹配的医生
    complexity_boost = 1.0 + 0.3 * study.complexity * spec
    return spec * urg * fatigue_factor * complexity_boost


def agent_assign(
    studies: List[Study],
    radiologists: List[Radiologist],
    top_n: int = 5,
) -> dict:
    """
    Agent 调度：对每个 study，计算所有医生的 assignment_score，
    返回 top_n 候选及分数，供最终分发决策使用。
    """
    assignments = {}
    for study in studies:
        scores = []
        for rad in radiologists:
            s = assignment_score(rad, study)
            scores.append((rad.name, s, {
                "specialty_match": specialty_match(rad, study),
                "urgency_weight": urgency_weight(study),
                "fatigue_factor": 1.0 - 0.4 * rad.fatigue_score,
                "rad_load": rad.current_load,
            }))
        # 按分数降序排列
        scores.sort(key=lambda x: x[1], reverse=True)
        assignments[study.study_id] = scores[:top_n]
    return assignments


# ── 模拟场景 ──────────────────────────────────────────

def run_simulation():
    now = datetime.now()

    radiologists = [
        Radiologist("张医生", specialties=["neuro", "head"],
                    shift_start=now - timedelta(hours=7),
                    cases_read_today=18, total_read_minutes_today=420),
        Radiologist("李医生", specialties=["chest", "abdomen"],
                    shift_start=now - timedelta(hours=3),
                    cases_read_today=6, total_read_minutes_today=150),
        Radiologist("王医生", specialties=["spine", "msk"],
                    shift_start=now - timedelta(hours=5),
                    cases_read_today=12, total_read_minutes_today=280),
    ]

    studies = [
        Study("S001", "MRI", "head", complexity=0.85, priority_label="stat", rvu=12.0, wait_minutes=45),
        Study("S002", "CT",  "chest", complexity=0.4, priority_label="urgent", rvu=8.0, wait_minutes=90),
        Study("S003", "CT",  "abdomen", complexity=0.6, priority_label="routine", rvu=6.5, wait_minutes=120),
        Study("S004", "MRI", "spine", complexity=0.7, priority_label="urgent", rvu=9.0, wait_minutes=30),
        Study("S005", "CT",  "head", complexity=0.9, priority_label="stat", rvu=14.0, wait_minutes=75),
    ]

    result = agent_assign(studies, radiologists)

    print("=" * 60)
    print("放射科 AI Agent 调度结果")
    print("=" * 60)

    for sid, candidates in result.items():
        study = next(s for s in studies if s.study_id == sid)
        print(f"\n病例 {sid} | {study.modality}-{study.body_part} "
              f"| 复杂度={study.complexity} | 优先级={study.priority_label} "
              f"| 等待={study.wait_minutes}min")
        print("-" * 50)
        for name, score, detail in candidates:
            print(f"  → {name}: 综合分={score:.2f} "
                  f"(专长={detail['specialty_match']:.1f} "
                  f"紧急={detail['urgency_weight']:.1f} "
                  f"疲劳因子={detail['fatigue_factor']:.2f} "
                  f"已读={detail['rad_load']}例)")

    # 对比：传统规则分配（按优先级标签 + 时间顺序，无视专长和疲劳）
    print("\n" + "=" * 60)
    print("对比：传统规则分配（按优先级 + 到达顺序，轮询分发）")
    print("=" * 60)
    priority_order = {"stat": 0, "urgent": 1, "routine": 2}
    sorted_studies = sorted(studies, key=lambda s: (priority_order[s.priority_label], s.wait_minutes))
    for i, study in enumerate(sorted_studies):
        rad = radiologists[i % len(radiologists)]
        print(f"  {study.study_id} → {rad.name} (轮询，无专长/疲劳考量)")

if __name__ == "__main__":
    run_simulation()