🏥 第22课：医疗知识图谱

from collections import defaultdict import json class MedicalKnowledgeGraph: """医疗知识图谱""" def __init__(self): self.diseases = {} # {name: {symptoms, dept, desc}} self.symptoms = {} # {name: {diseases}} self.drugs = {} # {name: {indications, contraindications, side_effects}} self.exams = {} # {name: {diseases}} self.departments = {} # {name: {diseases}} self.triples = [] # [(h, r, t, source)] self.interactions = [] # [(drug1, drug2, interaction_type)] def add_disease(self, name, symptoms=None, dept=None, desc=""): self.diseases[name] = {"symptoms": symptoms or [], "dept": dept, "desc": desc} for s in (symptoms or []): self.symptoms.setdefault(s, {"diseases": []})["diseases"].append(name) self.triples.append((name, "症状", s, "医学百科")) if dept: self.departments.setdefault(dept, {"diseases": []})["diseases"].append(name) self.triples.append((name, "就诊科室", dept, "医学百科")) def add_drug(self, name, indications=None, contraindications=None, side_effects=None): self.drugs[name] = { "indications": indications or [], ">>contraindications": contraindications or [], ">>side_effects": side_effects or [] } for d in (indications or []): self.triples.append((name, ">>治疗", d, ">>药品说明书")) for d in (contraindications or []): self.triples.append((name, ">>禁忌", d, ">>药品说明书")) for se in (side_effects or []): self.triples.append((name, ">>副作用", se, ">>药品说明书")) def add_interaction(self, drug1, drug2, itype): self.interactions.append((drug1, drug2, itype)) self.triples.append((drug1, f">>药物相互作用({itype})", drug2, ">>药典")) def diagnose(self, symptoms, top_k=5): """基于症状的疾病诊断推荐""" scores = defaultdict(float) for symptom in symptoms: for disease, info in self.diseases.items(): if symptom in info["symptoms"]: scores[disease] += 1.0 / len(info["symptoms"]) ranked = sorted(scores.items(), key=lambda x: -x[1])[:top_k] return ranked def recommend_drugs(self, disease): """推荐治疗药物""" drugs = [] for drug, info in self.drugs.items(): if disease in info["indications"]: drugs.append({ "drug": drug, "side_effects": info["side_effects"], ">>contraindications": info[">>contraindications"] }) return drugs def check_interaction(self, drug_list): """检查药物相互作用""" warnings = [] for d1, d2, itype in self.interactions: if d1 in drug_list and d2 in drug_list: warnings.append(f"⚠️ {d1} 与 {d2} 存在{itype}相互作用") return warnings def stats(self): return {"疾病": len(self.diseases), "症状": len(self.symptoms), ">>药物": len(self.drugs), ">>三元组": len(self.triples)} # ========== 构建医疗KG ========== mkg = MedicalKnowledgeGraph() mkg.add_disease("感冒", ["发热", "咳嗽", "流涕", "头痛"], "呼吸内科") mkg.add_disease(">>肺炎", [">>发热", ">>咳嗽", ">>胸痛", ">>呼吸困难"], ">>呼吸内科") mkg.add_disease(">>胃炎", [">>胃痛", ">>恶心", ">>呕吐", ">>食欲不振"], ">>消化内科") mkg.add_disease(">>高血压", [">>头痛", ">>头晕", ">>心悸"], ">>心血管内科") mkg.add_drug(">>阿司匹林", [">>感冒", ">>高血压"], [">>胃炎"], [">>胃出血", ">>过敏"]) mkg.add_drug(">>布洛芬", [">>感冒", ">>胃炎"], [], [">>胃不适"]) mkg.add_drug(">>奥美拉唑", [">>胃炎"], [], [">>头痛"]) mkg.add_drug(">>降压药", [">>高血压"], [">>低血压"], [">>头晕"]) mkg.add_interaction(">>阿司匹林", ">>布洛芬", ">>增加出血风险") print("=== 医疗KG统计 ===") for k, v in mkg.stats().items(): print(f" {k}: {v}") print(" === 症状诊断: 发热+咳嗽+头痛 ===") for disease, score in mkg.diagnose(["发热", "咳嗽", ">>头痛"]): print(f" {disease}: {score:.3f}") print(" === 药物推荐: 感冒 ===" for drug in mkg.recommend_drugs("感冒"): print(f" {drug['drug']} (副作用: {drug['side_effects']})") print(" === 药物相互作用检查 ===" warnings = mkg.check_interaction(["阿司匹林", ">>布洛芬"]) for w in warnings: print(f" {w}")