大模型应用：医疗AI智能体幻觉防控：幻觉抑制的四层防护体系应用实践.128

def medical_ai_prompt_constraint(user_question: str) -> str:
    """
    医疗AI提示词约束函数：第一层防护，构建强制约束的输入提示词
    :param user_question: 用户医疗问题
    :return: 约束后的标准提示词
    """
    # 核心约束模板：医疗场景零幻觉强制提示词
    prompt_template = """
    你是【医疗AI辅助智能体】，严格遵守以下规则，零容忍任何虚假信息：
    1. 身份约束：仅提供医疗科普、用药咨询辅助服务，禁止出具执业医师级诊断结论；
    2. 知识约束：**仅使用内置权威医疗知识库中的信息回答问题**，不得使用任何外部知识；
    3. 行为约束：禁止编造药品名称、用药剂量、诊疗方案、疾病诊断结论；
    4. 兜底规则：若知识库中无对应问题的信息，**唯一回复：暂无相关建议**；
    5. 格式约束：回答简洁准确，无夸张、虚假、超范围表述，有据可查。
    
    用户问题：{question}
    请严格按照规则生成回答：
    """.strip()
    
    # 填充用户问题，生成约束后的提示词
    final_prompt = prompt_template.format(question=user_question)
    return final_prompt

# 测试示例
if __name__ == '__main__':
    # 用户问题
    test_question = "高血压应该吃什么药？"
    # 生成约束提示词
    constrained_prompt = medical_ai_prompt_constraint(test_question)
    print("约束后的医疗AI提示词：")
    print(constrained_prompt)

import re

class MedicalOutputValidator:
    """医疗AI输出校验引擎：第二层防护，实时拦截幻觉内容"""
    def __init__(self):
        # 1. 权威药品白名单（国家药监局认证药品）
        self.legal_drugs = [
            "氨氯地平", "硝苯地平", "缬沙坦", "氯沙坦", "美托洛尔",  # 高血压
            "阿托伐他汀", "辛伐他汀", "瑞舒伐他汀",  # 降脂药
            "二甲双胍", "格列美脲", "阿卡波糖",  # 降糖药
            "阿司匹林", "布洛芬", "对乙酰氨基酚"  # 常见药
        ]
        # 2. 药品后缀（用于识别药品类词汇）
        self.drug_suffixes = ["地平", "沙坦", "洛尔", "普利", "他汀", "西平", "东平", "南平", "北平"]
        # 3. 违规诊断关键词（禁止AI出具的诊断表述）
        self.illegal_diagnosis = ["确诊为", "确诊患有", "你患有", "就是", "肯定是", "绝对是", "确诊"]
        # 4. 虚假疗法关键词（无科学依据的医疗建议）
        self.fake_therapies = ["断食治愈", "偏方根治", "不用吃药", "百分百治愈"]
        # 5. 标准兜底回复
        self.standard_unknown = "暂无相关建议"

    def validate(self, ai_output: str) -> dict:
        """
        校验AI输出内容，返回校验结果
        :param ai_output: 大模型生成的原始回复
        :return: 校验结果（是否通过、拦截原因）
        """
        result = {
            "is_pass": True,
            "intercept_reason": "",
            "final_output": ai_output
        }

        # 规则1：拦截虚假药品名称（优先级最高，防止幻觉药品）
        # 使用智能分割方法：按标点和空格分割，然后检查每个片段
        # 分割符：逗号、顿号、句号、空格
        segments = re.split(r'[,，。、\s]+', ai_output)

        for segment in segments:
            # 对于每个片段，查找4-5字的药品名称
            if len(segment) >= 4:
                # 查找所有可能的药品名称（4字或5字）
                for i in range(len(segment) - 3):
                    # 检查4字药名
                    if i + 4 <= len(segment):
                        candidate = segment[i:i+4]
                        if any(candidate.endswith(suffix) for suffix in self.drug_suffixes):
                            # 检查是否包含白名单中的药品
                            contains_legal_drug = False
                            for legal_drug in self.legal_drugs:
                                if legal_drug in candidate:
                                    contains_legal_drug = True
                                    break
                            if not contains_legal_drug and candidate not in self.legal_drugs:
                                result["is_pass"] = False
                                result["intercept_reason"] = f"拦截原因：包含虚假药品名称【{candidate}】"
                                return result

                for i in range(len(segment) - 4):
                    # 检查5字药名
                    if i + 5 <= len(segment):
                        candidate = segment[i:i+5]
                        if any(candidate.endswith(suffix) for suffix in self.drug_suffixes):
                            # 检查是否包含白名单中的药品
                            contains_legal_drug = False
                            for legal_drug in self.legal_drugs:
                                if legal_drug in candidate:
                                    contains_legal_drug = True
                                    break
                            if not contains_legal_drug and candidate not in self.legal_drugs:
                                result["is_pass"] = False
                                result["intercept_reason"] = f"拦截原因：包含虚假药品名称【{candidate}】"
                                return result

        # 规则2：拦截违规诊断结论
        for keyword in self.illegal_diagnosis:
            if keyword in ai_output:
                result["is_pass"] = False
                result["intercept_reason"] = f"拦截原因：包含违规诊断表述【{keyword}】"
                return result

        # 规则3：拦截虚假无依据疗法
        for therapy in self.fake_therapies:
            if therapy in ai_output:
                result["is_pass"] = False
                result["intercept_reason"] = f"拦截原因：包含虚假医疗疗法【{therapy}】"
                return result

        # 规则4：无答案时强制标准回复
        if self.standard_unknown not in ai_output and len(ai_output) < 5:
            result["final_output"] = self.standard_unknown

        return result


# 测试示例
if __name__ == '__main__':
    # 测试校验引擎
    validator = MedicalOutputValidator()
    # 测试1：正常合规回复
    test1 = "高血压一线用药推荐氨氯地平、缬沙坦"
    print(validator.validate(test1))
    # 测试2：虚假药品（幻觉内容）
    test2 = "高血压推荐服用氨氯西平"
    print(validator.validate(test2))
    # 测试3：违规诊断
    test3 = "你确诊患有高血压"
    print(validator.validate(test3))

import json

class MedicalWhiteList:
    """医疗AI白名单库：第三层防护，高频问题医师审核兜底"""
    def __init__(self, whitelist_path: str = "medical_whitelist.json"):
        self.whitelist_path = whitelist_path
        self.whitelist = self.load_whitelist()

    def load_whitelist(self) -> dict:
        """加载医师审核的白名单库"""
        # 模拟白名单数据：key=用户问题，value=医师审核标准答案
        default_whitelist = {
            "高血压吃什么药": "高血压一线用药包括氨氯地平、硝苯地平、缬沙坦、氯沙坦、美托洛尔（来源：国家卫健委高血压诊疗指南）",
            "糖尿病饮食注意什么": "糖尿病患者建议低糖、低脂、高纤维饮食，定时定量进餐（来源：中华医学会糖尿病学分会指南）",
            "感冒发烧怎么办": "普通感冒建议多喝水、多休息，体温超38.5℃可服用对乙酰氨基酚（来源：国家药监局药品说明书）"
        }
        # 保存到本地文件
        with open(self.whitelist_path, "w", encoding="utf-8") as f:
            json.dump(default_whitelist, f, ensure_ascii=False, indent=4)
        return default_whitelist

    def match_question(self, user_question: str) -> str | None:
        """
        匹配用户问题，返回白名单答案
        :param user_question: 用户问题
        :return: 匹配成功返回答案，失败返回None
        """
        # 精准匹配（可扩展为模糊匹配）
        if user_question in self.whitelist:
            return self.whitelist[user_question]
        return None

# 测试示例
if __name__ == '__main__':
    whitelist = MedicalWhiteList()
    # 匹配高频问题
    test_question = "高血压吃什么药"
    answer = whitelist.match_question(test_question)
    if answer:
        print(f"白名单匹配成功：{answer}")
    else:
        print("未匹配到白名单，进入大模型流程")

import datetime
import json

class FeedbackLoopSystem:
    """医疗AI反馈闭环系统：第四层防护，自动收集优化数据"""
    def __init__(self, feedback_path: str = "feedback_pool.json"):
        self.feedback_path = feedback_path
        self.feedback_pool = self.load_feedback_pool()

    def load_feedback_pool(self) -> list:
        """加载反馈优化池"""
        try:
            with open(self.feedback_path, "r", encoding="utf-8") as f:
                return json.load(f)
        except:
            return []

    def collect_feedback(self, user_question: str, ai_output: str, feedback_reason: str):
        """
        收集用户反馈数据，存入优化池
        :param user_question: 用户问题
        :param ai_output: AI回复
        :param feedback_reason: 反馈原因（不准确/虚假信息等）
        """
        feedback_data = {
            "timestamp": datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
            "user_question": user_question,
            "ai_output": ai_output,
            "feedback_reason": feedback_reason,
            "review_status": "未审核",  # 医师审核状态
            "hallucination_type": ""  # 幻觉类型
        }
        self.feedback_pool.append(feedback_data)
        # 保存到文件
        with open(self.feedback_path, "w", encoding="utf-8") as f:
            json.dump(self.feedback_pool, f, ensure_ascii=False, indent=4)
        print(f"反馈已收集，当前优化池数据量：{len(self.feedback_pool)}")

    def weekly_model_update(self):
        """每周模型迭代：基于反馈数据优化"""
        # 统计未审核数据
        unreviewed = [f for f in self.feedback_pool if f["review_status"] == "未审核"]
        print(f"本周待优化数据：{len(unreviewed)} 条")
        # 医师审核+模型微调（实战中对接大模型微调接口）
        for data in unreviewed:
            data["review_status"] = "已审核"
            data["hallucination_type"] = "虚假药品信息"
        # 保存更新后的数据
        with open(self.feedback_path, "w", encoding="utf-8") as f:
            json.dump(self.feedback_pool, f, ensure_ascii=False, indent=4)
        print("模型每周迭代完成，幻觉率持续优化")

# 测试示例
if __name__ == '__main__':
    feedback_system = FeedbackLoopSystem()
    # 收集用户反馈
    feedback_system.collect_feedback(
        user_question="高血压吃什么药",
        ai_output="推荐服用氨氯西平",
        feedback_reason="回复不准确，药品名称错误"
    )
    # 执行每周迭代
    feedback_system.weekly_model_update()