简介这是一套基于Python实现的医疗知识图谱自动问答系统源码面向人工智能、自然语言处理与智慧医疗方向的学习者和开发者旨在解决医疗领域专业问题的语义理解与精准检索难题。资源包含26个文件涵盖8个Python核心模块如question_classifier.py、chatbot_graph.py、answer_search.py等、5个XML配置与数据定义文件、8个TXT医学词典含疾病、症状、药物、科室等实体、2个JSON知识图谱数据文件及1张界面演示图整体压缩包大小为15.54MB结构清晰、模块职责明确便于快速理解问答流程与图谱构建逻辑。已有184人学习下载适合希望掌握知识图谱建模、医疗NLP实战、Neo4j图查询与Flask轻量Web集成的中级开发者。读者可直接运行调试完整复现从数据预处理、实体关系抽取、Cypher查询到答案生成的全流程获得可拓展的医疗问答系统基线代码与工程组织范式。1. 这不是另一个“AI问诊Demo”一个能跑通实体链接关系推理多跳查询的医疗知识图谱问答源码专治“问得对但查不出”的黑匣子问题你有没有试过——把“高血压患者能吃阿司匹林吗”喂给一个医疗问答系统它精准识别出“高血压”“阿司匹林”却卡在“禁忌症”和“药物相互作用”这两个关键关系上最后返回一句“请咨询医生”这不是模型能力弱而是知识没连成网。这份QAMedicalKG-master源码包本质是一套可落地、可调试、可验证的医疗知识图谱问答流水线它不依赖大语言模型兜底而是用规则图谱轻量NLP组合拳在 Neo4j 图数据库里完成从中文问句→实体消歧→关系路径生成→Cypher 多跳查询→结构化答案组装的全链路。它适合两类人一是想真正搞懂“知识图谱怎么服务真实医疗场景”的算法/工程同学二是需要快速搭建一个有明确知识边界、可解释、可审计的内部辅助问答模块的医疗信息化团队。它不承诺替代医生但能帮你把“疾病-症状-检查-药物-禁忌”这五层关系织成一张可追溯、可增删、可 debug 的网——这才是医疗 AI 落地最硬的那块砖。2. 从 raw JSON 到 Neo4j 图谱数据准备与图谱构建的四个必过环节这个项目的数据流非常清晰原始文本.txt和结构化 JSONmedical.json,medical2.json→ 清洗与标准化 → 实体-关系三元组抽取 → Neo4j 导入。但每一步都有隐性门槛尤其对没碰过医疗领域数据的同学。我拆开prepare_data/和build_medicalgraph.py把关键动作落到可执行命令和参数上。2.1 数据清洗为什么food.txt和drug.txt必须人工校验项目里dict/下的 8 个.txt文件是领域词典基础格式为纯文本每行一个术语如高血压、阿司匹林肠溶片。但实际使用中你会发现drug.txt里混有商品名拜阿司匹灵和通用名阿司匹林而disease.txt中存在同义词未归一心肌梗死vs急性心肌梗塞。prepare_data/max_cut.py用 jieba 做最大匹配分词但它不会自动合并同义词。必须手动做三件事用pandas加载所有.txt去重并按 Unicode 排序对drug.txt和disease.txt建立映射表{“拜阿司匹灵”: “阿司匹林”, “急性心肌梗塞”: “心肌梗死”}存为dict/synonym_map.json修改build_data.py中的load_dict()函数强制加载该映射表并做标准化替换。# build_data.py 第 45 行附近插入以下逻辑 def load_dict(filepath, synonym_map_pathdict/synonym_map.json): with open(filepath, r, encodingutf-8) as f: terms [line.strip() for line in f if line.strip()] if os.path.exists(synonym_map_path): with open(synonym_map_path, r, encodingutf-8) as f: synonym_map json.load(f) # 标准化将原文本中的键替换为值 terms [synonym_map.get(t, t) for t in terms] return list(set(terms)) # 去重提示synonym_map.json是你后续扩展知识边界的锚点。比如新增“二甲双胍缓释片”→“二甲双胍”只需改这里无需动分词或图谱构建逻辑。2.2 三元组抽取medical.json的 schema 解析与build_medicalgraph.py的字段映射medical.json是核心知识源格式为 JSONL每行一个 JSON 对象典型样例如下{ name: 糖尿病, cause: [遗传因素, 肥胖], symptom: [多饮, 多尿, 体重下降], drug: [二甲双胍, 胰岛素], check: [空腹血糖, 糖化血红蛋白], department: [内分泌科] }注意它不是 RDF 三元组而是以疾病为中心的属性聚合。build_medicalgraph.py的任务就是把它“打散”成(疾病)-[HAS_CAUSE]-(病因)这样的边。关键在create_graph()函数中对字段的硬编码映射# build_medicalgraph.py 第 120 行起 field_to_relation { cause: HAS_CAUSE, symptom: HAS_SYMPTOM, drug: TREATS, # 注意这里反向疾病TREATS药物逻辑错误 check: REQUIRES_CHECK, department: BELONGS_TO_DEPT }血泪经验drug: TREATS是典型逻辑翻车点。正确应为(疾病)-[TREATED_BY]-(药物)或(药物)-[TREATS]-(疾病)。我选择后者统一主语为“药物”便于后续问“什么药治糖尿病”时直接 MATCH(d:Drug)-[:TREATS]-(dis:Disease)。因此必须修改field_to_relation { cause: HAS_CAUSE, symptom: HAS_SYMPTOM, drug: TREATS, # 保持(Drug)-[:TREATS]-(Disease) check: REQUIRES_CHECK, # (Disease)-[:REQUIRES_CHECK]-(Check) department: BELONGS_TO_DEPT } # 同时调整 create_node_and_rel() 中的节点创建顺序 # 先建 Disease 节点再建 Drug 节点最后建 TREATS 关系2.3 Neo4j 导入用neo4j-admin import替代 Python 驱动批量写入build_medicalgraph.py默认用py2neo逐条写入10 万节点关系时耗时超 2 小时且易断连。生产环境必须切到离线导入。项目已生成data/nodes.csv和data/relations.csv但格式需微调nodes.csv必须含:ID,name:string,:LABEL三列relations.csv必须含:START_ID,:END_ID,:TYPE三列所有字符串字段用双引号包裹逗号转义为,。我写了一个fix_csv_for_neo4j.py脚本预处理# fix_csv_for_neo4j.py import pandas as pd import csv def fix_csv(infile, outfile, is_relationFalse): df pd.read_csv(infile, dtypestr, keep_default_naFalse) if is_relation: df[:START_ID] df[:START_ID] df[:END_ID] df[:END_ID] df[:TYPE] df[:TYPE] else: df[:ID] df[:ID] df[name] df[name] df[:LABEL] df[:LABEL] df.to_csv(outfile, indexFalse, quotingcsv.QUOTE_NONE, escapechar\\) fix_csv_for_neo4j(data/nodes.csv, data/nodes_fixed.csv) fix_csv_for_neo4j(data/relations.csv, data/relations_fixed.csv, is_relationTrue)然后终端执行Neo4j 5.x# 停止 Neo4j 服务 sudo systemctl stop neo4j # 清空原数据库谨慎 sudo rm -rf /var/lib/neo4j/data/databases/graph.db # 离线导入路径按你的 Neo4j 安装位置调整 sudo /var/lib/neo4j/bin/neo4j-admin import \ --nodes/path/to/data/nodes_fixed.csv \ --relationships/path/to/data/relations_fixed.csv \ --id-typeSTRING \ --databasegraph.db # 启动 sudo systemctl start neo4j提示--id-typeSTRING是关键参数。医疗实体名如“高血压”是中文字符串若不指定导入会失败并报错ID must be numeric。3. 从问句到 Cypher问题解析器question_parser.py的三层解耦设计问答效果好不好70% 取决于question_parser.py是否能把“孕妇能吃布洛芬吗”这种模糊问句稳定拆解成(Drug)-[:CONTRAINDICATED_FOR]-(Population)这样的可查询模式。它不是端到端黑盒而是分三层意图识别 → 实体抽取 → 关系路径生成。每一层都可独立测试、替换、debug。3.1 意图分类器question_classifier.py用规则TF-IDF 而非 BERT 的务实选择项目用sklearn.feature_extraction.text.TfidfVectorizerLinearSVC训练了一个轻量意图分类器支持 6 类disease_symptom,disease_cause,disease_drug,drug_disease,disease_check,disease_department。训练数据在data/question_intent_train.csv格式为question,intent。但原始代码没提供训练脚本。我补全了train_classifier.py并强调两个关键实践停用词必须医疗定制通用停用词“的”、“了”会干掉关键信息。我在question_classifier.py的get_stopwords()中加入[能, 可以, 是否, 怎么, 什么]—— 因为“能吃吗”里的“能”是意图信号不是噪音TF-IDF 的 ngram_range(1,2)单字如“高”血压和双字“高血压”都保留避免切碎疾病名。# train_classifier.py from sklearn.feature_extraction.text import TfidfVectorizer from sklearn.svm import LinearSVC import jieba def custom_tokenizer(text): # 先用jieba分词再过滤停用词 words jieba.lcut(text) stopwords get_stopwords() # 返回定制停用词列表 return [w for w in words if w not in stopwords and len(w) 1] vectorizer TfidfVectorizer( tokenizercustom_tokenizer, ngram_range(1, 2), # 关键捕获“高血压”和“高”“血压” max_features5000 )训练后保存模型import joblib joblib.dump(vectorizer, model/tfidf_vectorizer.pkl) joblib.dump(classifier, model/intent_classifier.pkl)3.2 实体链接器question_parser.py中的extract_entity()如何对抗歧义医疗实体歧义极强“钙”可能是元素、药物钙片、检查血钙“克”可能是单位mg、药物名克林霉素。extract_entity()用词典匹配 位置优先级解决步骤1在dict/所有.txt中全文匹配记录所有命中位置步骤2按长度降序排序长词优先“阿司匹林肠溶片” “阿司匹林”步骤3剔除被长词完全包含的短词“阿司匹林”在“阿司匹林肠溶片”中则丢弃步骤4对剩余词查synonym_map.json做标准化。# question_parser.py 第 88 行 def extract_entity(question): candidates [] for dict_name in [disease, drug, symptom, check]: dict_path fdict/{dict_name}.txt if not os.path.exists(dict_path): continue with open(dict_path, r, encodingutf-8) as f: terms [line.strip() for line in f if line.strip()] # 用正则找所有匹配保留起始位置 for term in terms: for match in re.finditer(re.escape(term), question): candidates.append((match.start(), match.end(), term, dict_name)) # 按长度降序再按起始位置升序长词优先左优先 candidates.sort(keylambda x: (-(x[1]-x[0]), x[0])) # 去重跳过被前面长词覆盖的 selected [] covered set() for start, end, term, dtype in candidates: if any(i in covered for i in range(start, end)): continue selected.append((start, end, term, dtype)) covered.update(range(start, end)) return selected注意此逻辑假设用户提问是线性文本。对“布洛芬和对乙酰氨基酚哪个更适合儿童”这种并列结构它会抽到两个药但不会自动识别“和”是并列关系——这是后续关系路径生成要解决的。3.3 关系路径生成generate_cypher()如何把“孕妇能吃布洛芬吗”翻译成可执行查询这是整个问答链最精妙的一环。generate_cypher()不是拼字符串而是根据意图实体类型查一个预定义的 Cypher 模板库意图主实体类型从实体类型Cypher 模板drug_contraindicationDrugPopulationMATCH (d:Drug {name:$drug})-[:CONTRAINDICATED_FOR]-(p:Population {name:$pop}) RETURN p.namedisease_drugDiseaseDrugMATCH (dis:Disease {name:$disease})-[:TREATS]-(d:Drug) RETURN d.name但原始代码只支持单跳。我扩展了generate_cypher()支持两跳当问“高血压会引起什么症状”时意图是disease_symptom但medical.json中symptom字段是直接挂在疾病下的所以是单跳而问“吃阿司匹林会导致胃出血吗”需走Drug-TREATS-Disease-HAS_COMPPLICATION-Symptom这就是两跳。# question_parser.py 新增 multi_hop_templates multi_hop_templates { drug_to_symptom_via_disease: MATCH (d:Drug {{name:$drug}}) -[:TREATS]-(dis:Disease) -[:HAS_COMPPLICATION]-(s:Symptom {{name:$symptom}}) RETURN s.name } def generate_cypher(intent, entities): if intent drug_contraindication and len(entities) 2: # 假设 entities[0] 是 drug, entities[1] 是 population return cypher_templates[drug_contraindication].format( drugentities[0][2], popentities[1][2] ) elif intent drug_to_symptom and len(entities) 2: # entities[0]: drug, entities[1]: symptom return multi_hop_templates[drug_to_symptom_via_disease].format( drugentities[0][2], symptomentities[1][2] ) # ... 其他分支4. 避坑五个让新手卡三天的真实问题与当场解决法部署这个项目时90% 的时间花在看似 trivial 的细节上。以下是我在三台不同环境Ubuntu 22.04 / macOS 14 / Windows WSL2反复验证过的坑每一条都附带现象 → 原因 → 解决的闭环。4.1 现象build_medicalgraph.py运行时报UnicodeDecodeError: gbk codec cant decode byte 0xad原因Windows 系统默认用 GBK 编码读取.txt文件但dict/下文件是 UTF-8 无 BOM 格式。open()不指定encoding时Python 在 Windows 上 fallback 到 GBK。解决全局搜索open(全部改为open(..., encodingutf-8)。重点文件build_data.py,question_parser.py,answer_search.py。4.2 现象Neo4j Web UI 能打开但chatbot_graph.py连接时报ServiceUnavailable: Cannot connect to ...原因Neo4j 5.x 默认关闭 HTTP 端口7474只开 Bolt7687且py2neo默认尝试 HTTP。chatbot_graph.py中Graph(http://localhost:7474, auth(neo4j, password))已失效。解决修改 Neo4j 配置/etc/neo4j/neo4j.conf取消注释dbms.connector.http.enabledtruedbms.connector.http.listen_address:7474重启 Neo4jsudo systemctl restart neo4j或更推荐改用 Bolt 协议在chatbot_graph.py中Graph(bolt://localhost:7687, auth(neo4j, password))4.3 现象问“糖尿病有什么症状”返回空但medical.json明确有symptom: [多饮, 多尿]原因build_medicalgraph.py中HAS_SYMPTOM关系的:END_ID写成了symptom_id但nodes.csv里symptom节点的:ID列名是symptom_nameID 值却是多饮这样的字符串导致关系找不到终点。解决检查data/nodes.csv确保symptom节点的:ID列值与:LABEL一致即:ID多饮,:LABELSymptom并在build_medicalgraph.py的write_relations()中写关系时:END_ID必须严格等于nodes.csv中该 symptom 的:ID值。4.4 现象question_classifier.py训练时报ValueError: X has 0 samples原因data/question_intent_train.csv是空文件或只有表头。项目 zip 包里该文件为空需自己构造至少 20 条样本。解决新建data/question_intent_train.csv内容示例question,intent 高血压有哪些症状,disease_symptom 糖尿病的病因是什么,disease_cause 阿司匹林能治疗什么病,drug_disease至少每类意图 3 条共 18 条以上。4.5 现象Flask 启动后访问http://localhost:5000显示Internal Server Error日志报KeyError: disease原因demo.jpg是静态资源但app.py中send_from_directory()路径写死为static而项目结构里图片在根目录。templates/index.html中img src{{ url_for(static, filenamedemo.jpg) }}会 404但 Flask 默认不暴露静态文件错误。解决创建static/目录把demo.jpg移进去或修改app.py中app.route(/static/path:filename)路由指向根目录app.route(/static/path:filename) def static_files(filename): return send_from_directory(., filename) # . 表示当前目录5. 问答效果调优用answer_search.py的debug_mode抓住每一次查询的“决策黑箱”answer_search.py是问答链最后一环也是最容易被忽略的调优入口。它接收question_parser.py生成的 Cypher执行查询并对结果做后处理去重、排序、生成自然语言答案。但原始代码没有 debug 开关你永远不知道 Cypher 是什么、查到了什么、为什么没答案。我给它加了debug_modeTrue参数让每次问答都在终端打印完整决策链。5.1 开启 debug 模式四层日志直击问题根源在chatbot_graph.py的search_answer()函数调用处传入debugTrue# app.py 第 62 行 answer, debug_info searcher.search_answer(question, debugTrue) print(DEBUG INFO:, debug_info) # 关键打印到终端 return jsonify({answer: answer})debug_info是一个 dict包含四层信息字段说明示例值intent分类意图disease_drugentities抽取的实体及类型[[糖尿病, disease], [二甲双胍, drug]]cypher生成的 Cypher 查询MATCH (dis:Disease {name:糖尿病})-[:TREATS]-(d:Drug) RETURN d.nameraw_resultsNeo4j 返回的原始结果[{d.name: 二甲双胍}, {d.name: 胰岛素}]提示raw_results是真相。如果cypher正确但raw_results为空说明图谱里没建这条边如果cypher错了比如TREATS写成TREAT那就是question_parser.py的模板 bug。5.2 结果后处理为什么answer_search.py的format_answer()必须支持多答案聚合医疗问答常有多解“糖尿病吃什么药”答案可能是[二甲双胍, 胰岛素, 格列美脲]。原始format_answer()只返回第一个。我重构了它支持三种模式def format_answer(raw_results, intent, modelist): values [r[list(r.keys())[0]] for r in raw_results] if mode list: return 、.join(values) # 二甲双胍、胰岛素、格列美脲 elif mode numbered: return .join([f{i1}.{v} for i, v in enumerate(values)]) elif mode confidence: # 假设 values 是 (entity, confidence_score) 元组 sorted_vals sorted(values, keylambda x: x[1], reverseTrue) return .join([f{v[0]}置信度{v[1]:.2f} for v in sorted_vals])在chatbot_graph.py中调用时指定answer format_answer(raw_results, intent, modenumbered)5.3 真实案例用 debug 日志定位“高血压不能吃什么药”查不到答案某次测试问“高血压不能吃什么药”debug_info显示{ intent: disease_contraindication, entities: [[高血压, disease], [布洛芬, drug]], cypher: MATCH (dis:Disease {name:高血压})-[:CONTRAINDICATED_FOR]-(d:Drug {name:布洛芬}) RETURN d.name, raw_results: [] }问题很清晰图谱里没有(高血压)-[:CONTRAINDICATED_FOR]-(布洛芬)这条边。但medical.json里高血压的contraindication字段是空的。解决方案有两个补数据编辑medical.json给高血压加上contraindication: [布洛芬, 麻黄碱]重新运行build_medicalgraph.py扩关系在build_medicalgraph.py中增加一条规则如果某 drug 的contraindication字段包含 disease 名则反向建CONTRAINDICATED_FOR边。我选了方案2因为它不破坏原始数据源且可复用。在build_medicalgraph.py的create_graph()循环内加# 遍历所有 drug for drug_data in drug_json: drug_name drug_data.get(name) contras drug_data.get(contraindication, []) for dis_name in contras: # 确保 dis_name 在 disease 节点中存在 if dis_name in disease_names: rel_writer.writerow([fDisease:{dis_name}, fDrug:{drug_name}, CONTRAINDICATED_FOR])这样只要drug.json里写了contraindication: [高血压]就会自动生成边。从那以后我每次新增一个 drug 或 disease都强制走一遍debug_mode看cypher和raw_results是否符合预期。这比盲调模型参数靠谱十倍。希望帮到你。本文还有配套的精品资源点击获取