大模型应用：医疗行业大模型：从生成前校验到生成后审计的应用实践.73

# 导入所需库
import re
from fuzzywuzzy import fuzz

# ====================== 第一步：构建基础医疗知识库和规则引擎 ======================
REQUIRED_INFO = {
    "电子病历": ["姓名", "性别", "年龄", "主诉", "现病史"],
    "用药建议": ["疾病名称", "年龄", "过敏史"]
}

# 隐私脱敏：仅对高确定性字段脱敏（手机号、身份证、住院号）
# 暂不自动脱敏“姓名”（因正则易误杀医学术语），依赖输入规范或后续泛化
PRIVACY_PATTERNS = {
    "手机号": re.compile(r"1[3-9]\d{9}"),
    "身份证号": re.compile(r"\d{17}[\dXx]"),
    "住院号": re.compile(r"住院号[:：]?\s*\d+|住院编号[:：]?\s*\d+")
}

MEDICAL_KNOWLEDGE = {
    "疾病名称": ["高血压", "2型糖尿病", "冠心病", "肺炎", "支气管炎"],
    "药物名称": ["阿司匹林", "硝苯地平", "二甲双胍", "阿莫西林", "沙丁胺醇"],
    "标准剂量": {
        "硝苯地平": "每日2次，每次10-20mg，每日最大剂量不超过60mg",
        "阿司匹林": "每日1次，每次100mg，饭后服用"
    }
}

FORMAT_RULES = {
    "电子病历": ["包含主诉", "包含现病史", "无乱码", "排版清晰"],
    "用药建议": ["包含药物名称", "包含服用剂量", "包含过敏提醒", "结尾标注仅供参考"]
}

# ====================== 第二步：生成前校验（语义化检查） ======================
def pre_generate_validate(input_content: str, scene: str) -> tuple[bool, str, str]:
    is_pass = True
    tip_info = "校验通过"
    processed_content = input_content

    if scene not in REQUIRED_INFO:
        return False, f"不支持的医疗场景，支持场景：{list(REQUIRED_INFO.keys())}", processed_content

    missing_info = []

    # === 语义化完整性校验 ===
    if scene == "用药建议":
        # 检查疾病（标准 + 口语 + 错别字）
        disease_terms = set(MEDICAL_KNOWLEDGE["疾病名称"]) | {"血压高", "血糖高", "心口疼", "高雪压"}
        if not any(term in processed_content for term in disease_terms):
            missing_info.append("疾病名称")

        # 检查年龄
        if not re.search(r"\d+\s*岁", processed_content):
            missing_info.append("年龄")

        # 检查过敏史
        if "过敏" not in processed_content:
            missing_info.append("过敏史")

    elif scene == "电子病历":
        if not re.search(r"患者|某[男女]|[男女]性", processed_content):
            missing_info.append("患者标识")
        if not re.search(r"[男女]性", processed_content):
            missing_info.append("性别")
        if not re.search(r"\d+\s*岁", processed_content):
            missing_info.append("年龄")
        if not re.search(r"主诉|因.*?就诊", processed_content):
            missing_info.append("主诉")
        if not re.search(r"现病史|近.*?天", processed_content):
            missing_info.append("现病史")

    # === 隐私脱敏（仅安全字段）===
    for privacy_type, pattern in PRIVACY_PATTERNS.items():
        if pattern.search(processed_content):
            processed_content = pattern.sub(f"[{privacy_type}]", processed_content)
            tip_info += f" | 已脱敏{privacy_type}"

    # === 标准化处理 ===
    typo_correction = {"高雪压": "高血压", "阿斯匹林": "阿司匹林", "二甲双瓜": "二甲双胍"}
    for typo, correct in typo_correction.items():
        if typo in processed_content:
            processed_content = processed_content.replace(typo, correct)
            tip_info += f" | 已修正错别字：{typo}→{correct}"

    oral_to_standard = {"血压高": "高血压", "血糖高": "2型糖尿病", "心口疼": "胸痛（冠心病待查）"}
    for oral, standard in oral_to_standard.items():
        if oral in processed_content:
            processed_content = processed_content.replace(oral, standard)
            tip_info += f" | 已标准化术语：{oral}→{standard}"

    if missing_info:
        is_pass = False
        tip_info = f"输入内容缺少必要信息：{', '.join(missing_info)}，请补充"

    return is_pass, tip_info, processed_content

# ====================== 第三步：生成后审计 ======================
def post_generate_audit(output_content: str, scene: str) -> tuple[bool, str, str]:
    is_pass = True
    tip_info = "审计通过"
    processed_output = output_content

    if scene not in FORMAT_RULES:
        return False, f"不支持的医疗场景：{scene}", processed_output

    # === 格式合规检查 ===
    format_rules_list = FORMAT_RULES[scene]
    detected_rules = []
    if "药物名称" in format_rules_list and any(drug in processed_output for drug in MEDICAL_KNOWLEDGE["药物名称"]):
        detected_rules.append("包含药物名称")
    if "服用剂量" in format_rules_list and re.search(r"\d+\s*mg|每次|每日", processed_output):
        detected_rules.append("包含服用剂量")
    if "过敏提醒" in format_rules_list and ("过敏" in processed_output or "避免" in processed_output):
        detected_rules.append("包含过敏提醒")
    if "结尾标注仅供参考" in format_rules_list and "仅供参考" in processed_output:
        detected_rules.append("结尾标注仅供参考")

    missing_format = [rule for rule in format_rules_list if rule not in detected_rules]
    if missing_format:
        is_pass = False
        tip_info = f"输出格式不合规，缺少：{', '.join(missing_format)}"
        if "结尾标注仅供参考" in missing_format:
            processed_output += "\n\n注：本文内容仅供医疗参考，不构成最终诊疗建议，具体请遵医嘱。"
            tip_info += " | 已自动补充“仅供参考”标注"

    # === 内容准确性审计 ===
    medical_errors = []
    for drug, std_dose in MEDICAL_KNOWLEDGE["标准剂量"].items():
        if drug in processed_output and std_dose not in processed_output:
            medical_errors.append(f"药物'{drug}'剂量未按标准表述（标准：{std_dose}）")

    if medical_errors:
        is_pass = False
        tip_info += f" | 内容准确性存疑：{', '.join(medical_errors)}，建议人工复核"

    # === 隐私二次审计（仅高风险字段）===
    for privacy_type, pattern in PRIVACY_PATTERNS.items():
        if pattern.search(processed_output):
            processed_output = pattern.sub(f"[{privacy_type}]", processed_output)
            tip_info += f" | 已脱敏输出中的{privacy_type}信息"
            is_pass = False

    # === 强制匿名化：移除任何可能的真实姓名引用 ===
    # 简单策略：将“患者XXX”替换为“患者”
    processed_output = re.sub(r"患者\s*[\u4e00-\u9fa5]{2,4}", "患者", processed_output)

    return is_pass, tip_info, processed_output

# ====================== 第四步：完整测试流程 ======================
if __name__ == "__main__":
    # 模拟用户输入（含真实姓名、手机号、错别字）
    user_input = """
    患者张三，手机号13812345678，年龄55岁，过敏史：青霉素。
    患有高雪压，需要生成用药建议，推荐阿斯匹林和硝苯地平。
    """
    medical_scene = "用药建议"

    print("=" * 60)
    print("【系统提示】正在处理医疗请求，请稍候...")
    print("（注意：手机号等高风险信息将被脱敏；姓名由系统泛化处理）")

    # 步骤1：生成前校验
    pre_pass, pre_tip, pre_processed_input = pre_generate_validate(user_input, medical_scene)
    print("\n" + "=" * 60)
    print("🔹 步骤1：生成前校验")
    status = "√ 校验通过" if pre_pass else "× 校验未通过"
    print(status)
    # print(f"√ 通过：{pre_pass}")
    print(f"！提示：{pre_tip}")
    print(f"脱敏并标准化后的输入内容：\n{pre_processed_input.strip()}")

    if pre_pass:
        print("\n" + "=" * 60)
        print("🔹 步骤2：大模型生成（模拟）")
        # 模拟 LLM 基于脱敏输入生成内容（使用泛化表述）
        llm_output = """
        针对高血压患者，推荐用药：阿司匹林、硝苯地平。
        阿司匹林：每日1次，每次200mg。
        硝苯地平：每日3次，每次20mg。
        过敏提醒：避免使用青霉素类药物。
        """
        print("大模型原始输出：")
        print(llm_output.strip())

        # 步骤3：生成后审计
        post_pass, post_tip, post_processed_output = post_generate_audit(llm_output, medical_scene)
        print("\n" + "=" * 60)
        print("🔹 步骤3：生成后审计")
        # print(f"√ 通过：{post_pass}")
        status = "√ 校验通过" if post_pass else "× 校验未通过"
        print(status)
        print(f"！提示：{post_tip}")
        print(f"最终交付内容（安全、匿名、合规）：\n{post_processed_output.strip()}")

        print("\n" + "=" * 60)
        print("√ 流程结束。输出已脱敏、匿名化，并符合医疗规范要求。")
    else:
        print("\n" + "=" * 60)
        print("× 流程终止：生成前校验未通过，无法继续。")