医疗AI智能体：多意图命中下的智能路由：高风险优先的医疗SKILL调度算法详解.141

import time
from typing import Dict, List

# 1. 定义医疗SKILL优先级与配置（核心规则）
SKILL_CONFIG = {
    "critical_emergency": {"name": "危急重症SKILL", "priority": 0, "base_score": 95},  # P0最高
    "bp_warning": {"name": "血压预警SKILL", "priority": 1, "base_score": 80},
    "glucose_monitor": {"name": "血糖监测SKILL", "priority": 2, "base_score": 75},
    "diet_guide": {"name": "饮食指导SKILL", "priority": 3, "base_score": 60},
    "chat": {"name": "闲聊SKILL", "priority": 4, "base_score": 0}
}

# 2. 关键词匹配库
KEYWORD_MAP = {
    "critical_emergency": ["昏迷", "休克", "心梗", "血压≥180", "血糖≥16", "心慌", "喘不上气"],
    "bp_warning": ["血压", "高血压", "头晕"],
    "glucose_monitor": ["血糖", "空腹血糖", "餐后血糖"],
    "diet_guide": ["吃什么", "饮食", "饭菜", "食谱"],
    "chat": ["你好", "谢谢", "在吗", "问候"]
}

class MedicalSkillScheduler:
    def __init__(self):
        self.user_history = {}  # 记录用户历史SKILL使用记录
        self.last_request_time = {}  # 记录上一次请求时间

    # 计算上下文衰减系数
    def calculate_decay(self, skill_name: str) -> float:
        last_time = self.last_request_time.get(skill_name, 0)
        if last_time == 0:
            return 1.0  # 从未调用过，不衰减，使用基础分
        now = time.time()
        interval_min = (now - last_time) / 60  # 时间间隔（分钟）
        decay = 0.8 ** interval_min  # 衰减公式
        return round(decay, 3)

    # 计算历史偏好加权分
    def calculate_preference(self, skill_name: str, user_id: str) -> int:
        history = self.user_history.get(user_id, {}).get(skill_name, 0)
        if history >= 10:
            return 15
        elif 5 <= history < 10:
            return 10
        elif 1 <= history < 5:
            return 5
        return 0

    # 关键词匹配命中SKILL
    def match_skills(self, query: str) -> List[str]:
        matched_skills = []
        for skill, keywords in KEYWORD_MAP.items():
            for kw in keywords:
                if kw in query:
                    matched_skills.append(skill)
                    break
        return matched_skills if matched_skills else ["chat"]

    # 计算SKILL最终得分
    def calculate_final_score(self, skill_name: str, user_id: str) -> float:
        base = SKILL_CONFIG[skill_name]["base_score"]
        decay = self.calculate_decay(skill_name)
        preference = self.calculate_preference(skill_name, user_id)
        final_score = base * decay + preference
        return round(final_score, 2)

    # 核心调度算法
    def schedule(self, query: str, user_id: str = "default_user") -> Dict:
        # Step1：匹配候选SKILL
        matched_skills = self.match_skills(query)
        
        # Step2：计算每个SKILL得分
        skill_scores = {}
        for skill in matched_skills:
            skill_scores[skill] = self.calculate_final_score(skill, user_id)
        
        # Step3：按优先级排序（优先级第一，分数第二）
        sorted_skills = sorted(
            skill_scores.keys(),
            key=lambda x: (SKILL_CONFIG[x]["priority"], -skill_scores[x])
        )
        
        # Step4：选定目标SKILL
        target_skill = sorted_skills[0]
        target_name = SKILL_CONFIG[target_skill]["name"]
        final_score = skill_scores[target_skill]
        
        # Step5：更新历史记录
        self.user_history[user_id] = self.user_history.get(user_id, {})
        self.user_history[user_id][target_skill] = self.user_history[user_id].get(target_skill, 0) + 1
        self.last_request_time[target_skill] = time.time()
        
        # Step6：生成调度结果
        return {
            "user_query": query,
            "matched_skills": {SKILL_CONFIG[s]["name"]: skill_scores[s] for s in matched_skills},
            "target_skill": target_name,
            "final_score": final_score,
            "priority": SKILL_CONFIG[target_skill]["priority"]
        }

# 测试示例
if __name__ == "__main__":
    scheduler = MedicalSkillScheduler()
    
    # 测试用例：危急重症场景
    query1 = "我血压190，心慌头晕，眼前发黑，该吃什么"
    result1 = scheduler.schedule(query1)
    print("=== 测试用例1：危急重症场景 ===")
    print(f"用户输入：{result1['user_query']}")
    print(f"命中SKILL及得分：{result1['matched_skills']}")
    print(f"最终调度：{result1['target_skill']}，优先级：{result1['priority']}，最终得分：{result1['final_score']}")
    print("-" * 80)
    
    # 测试用例：常规慢病场景
    query2 = "我空腹血糖7.0，今天饮食注意什么"
    result2 = scheduler.schedule(query2)
    print("=== 测试用例2：常规慢病场景 ===")
    print(f"用户输入：{result2['user_query']}")
    print(f"命中SKILL及得分：{result2['matched_skills']}")
    print(f"最终调度：{result2['target_skill']}，优先级：{result2['priority']}，最终得分：{result2['final_score']}")

import matplotlib.pyplot as plt
import numpy as np

# 设置中文字体
plt.rcParams["font.family"] = ["SimHei", "WenQuanYi Micro Hei", "Heiti TC"]
plt.rcParams["axes.unicode_minus"] = False

# 1. 医疗SKILL优先级可视化
def draw_skill_priority():
    priorities = ["P0\n危急重症", "P1\n血压预警", "P2\n血糖监测", "P3\n饮食指导", "P4\n闲聊"]
    levels = [100, 80, 60, 40, 20]
    colors = ["#DC143C", "#FF4500", "#FFA500", "#32CD32", "#1E90FF"]
    
    plt.figure(figsize=(10, 6))
    bars = plt.bar(priorities, levels, color=colors, edgecolor="black", linewidth=1.5)
    plt.title("医疗SKILL优先级与风险等级可视化", fontsize=16, pad=20)
    plt.xlabel("SKILL优先级", fontsize=12)
    plt.ylabel("风险等级（分值越高风险越高）", fontsize=12)
    plt.ylim(0, 110)
    
    # 添加数值标签
    for bar, level in zip(bars, levels):
        plt.text(bar.get_x() + bar.get_width()/2, bar.get_height() + 2,
                f"{level}分", ha="center", va="bottom", fontsize=11, fontweight="bold")
    
    plt.grid(axis="y", alpha=0.3)
    plt.tight_layout()
    plt.savefig("medical_skill_priority.png", dpi=300, bbox_inches="tight")
    plt.close()
    print("优先级图表已保存：medical_skill_priority.png")

# 2. 上下文衰减曲线可视化
def draw_decay_curve():
    minutes = np.arange(0, 21, 1)
    decay = 0.8 ** minutes
    
    plt.figure(figsize=(10, 6))
    plt.plot(minutes, decay, color="#FF4500", linewidth=3, marker="o", markersize=4)
    plt.title("医疗场景上下文衰减曲线（衰减系数=0.8）", fontsize=16, pad=20)
    plt.xlabel("时间间隔（分钟）", fontsize=12)
    plt.ylabel("上下文权重系数", fontsize=12)
    plt.grid(alpha=0.3)
    plt.ylim(0, 1.05)
    
    # 标注关键节点
    key_points = [5, 10, 15, 20]
    for p in key_points:
        plt.annotate(f"{p}分钟: {0.8**p:.2f}",
                    xy=(p, 0.8**p),
                    xytext=(p+0.5, 0.8**p+0.05),
                    arrowprops=dict(arrowstyle="->", color="red"),
                    fontsize=10)
    
    plt.tight_layout()
    plt.savefig("context_decay_curve.png", dpi=300, bbox_inches="tight")
    plt.close()
    print("衰减曲线图表已保存：context_decay_curve.png")

# 执行生成
if __name__ == "__main__":
    draw_skill_priority()
    draw_decay_curve()