简介本资源是一套基于Python与Neo4j实现的医疗领域知识图谱构建及智能问答系统完整源码面向具备基础Python编程与图数据库认知的中高级开发者、知识图谱初学者及NLP实践者解决从结构化/非结构化数据建模、图谱构建到自然语言问句解析与答案检索的全流程技术落地问题。压缩包共40个文件含9个核心Python脚本如build_medicalgraph.py、question_classifier.py、answer_search.py、11个领域文本数据涵盖疾病、症状、药品等实体、9张流程与效果示意图如kg_route.png、chat1.png、2个JSON知识库kg_crime.json等、1份README说明及1份刘焕勇老师PPT课件整体9.18MB结构清晰、模块解耦。已有798人学习下载提供可直接运行的问答交互逻辑、预置医疗语料与图谱构建工具链覆盖数据准备、图谱导入、问句分类、意图识别、路径查询与答案生成全环节是理解知识图谱驱动问答系统工程实现的典型参考案例。1. 这不是个“玩具项目”用 Python Neo4j 搭出能查药品禁忌、症状关联、科室归属的医疗知识图谱问答系统新手照着跑通就能上手真实业务场景你可能见过很多“知识图谱入门 demo”——三五个节点连几条边跑个MATCH (n) RETURN n就叫图谱。但这份源码不是。它从原始文本disease.txt,symptom.txt,drug.txt,check.txt,department.txt出发用build_medicalgraph.py构建含 10 实体类型、50 关系类型的医疗子图用question_classifier.py做意图识别区分“症状→疾病”、“药物→禁忌”、“科室→擅长病种”再通过answer_search.py把自然语言问题翻译成 Cypher 查询比如“高血压患者能吃阿司匹林吗” →MATCH (d:Disease)-[r:HAS_DRUG_CONTRAINDICATION]-(dr:Drug) WHERE d.name高血压 AND dr.name阿司匹林 RETURN r.reason。它不依赖预训练大模型纯规则模板图查询部署轻、响应快、可解释强——正适合医院信息科、药企知识管理、基层辅助诊断等对可控性、合规性要求高的场景。如果你正在找一个有真实数据结构、有完整 pipeline、有可调试模块、且不碰任何敏感词和越界技术的 Python Neo4j 实战项目这个压缩包就是目前 GitHub 和 CSDN 上最扎实的“开箱即用型”医疗 KG 入口。2. 从零构建医疗知识图谱解析 build_medicalgraph.py 的四层数据加工逻辑与 Neo4j 写入策略2.1 数据源结构与实体关系建模依据为什么disease.txt必须按“疾病名|别名|简介|科室|检查|药物|禁忌”格式切分源码中data/目录下共 11 个.txt文件每个文件代表一类实体或关系种子。以disease.txt为例其典型行是高血压|原发性高血压,高血压病|长期动脉血压升高导致心脑肾损害|心内科,神经内科|心电图,超声心动图|硝苯地平,氨氯地平|孕妇禁用,低血压者慎用这不是随意拼接而是严格对应build_medicalgraph.py中create_disease_nodes()函数的字段解析逻辑def create_disease_nodes(self, file_path): with open(file_path, r, encodingutf-8) as f: for line in f: parts line.strip().split(|) if len(parts) 7: continue # 至少7字段name, alias, desc, dept, check, drug, contraindication name, alias, desc, dept, check, drug, contraindication parts # 创建 Disease 节点 self.graph.create(Node(Disease, namename, aliasalias, descdesc)) # 创建 Department 关系多值用逗号分割 for d in [x.strip() for x in dept.split(,) if x.strip()]: dept_node Node(Department, named) self.graph.merge(dept_node, Department, name) rel Relationship(self.graph.nodes.match(Disease, namename).first(), DEPT_SPECIALIZE, dept_node) self.graph.create(rel)提示dept,check,drug,contraindication字段均用英文逗号分隔这是为后续批量创建关系预留的结构化基础。若你的数据是 JSON 或 CSV需先做等价转换——不要直接改代码去适配新格式先统一转成这种竖线分隔的纯文本这是本项目最省力的数据预处理约定。2.2 四阶段图谱构建流程加载 → 清洗 → 实体归一 → 关系注入每步都带日志与失败跳过机制build_medicalgraph.py的主流程run()方法明确分为四个阶段且每阶段都带try...except和print(f[INFO] Stage X done)Stage 1加载所有 txt 文件到内存字典将disease.txt,symptom.txt,drug.txt等分别读入self.diseases,self.symptoms,self.drugs等 dictkey 为实体名value 为整行字符串便于后续切分。注意此处不做去重留待 Stage 2 处理。Stage 2实体名称标准化与别名合并遍历self.diseases对每个name|alias|...行将alias拆成列表与name合并为all_names [name] [a.strip() for a in alias.split(,)]再用fuzzywuzzy计算相似度把相似度 0.85 的别名指向同一主实体。这步解决“冠心病”和“冠状动脉粥样硬化性心脏病”指同一疾病的问题。Stage 3创建所有实体节点Disease/Symptom/Drug/Check/Department对每个实体类型调用对应create_xxx_nodes()方法。关键点所有节点创建均使用graph.merge()而非graph.create()避免重复插入同名节点。例如disease_node Node(Disease, namename, aliasalias, descdesc) self.graph.merge(disease_node, Disease, name) # 以 name 为唯一键 mergeStage 4注入六大类关系HAS_SYMPTOM / TREATS / CONTRAINDICATED_BY / BELONGS_TO_DEPT / NEEDS_CHECK / CAUSED_BY关系构建全部基于字段内容解析。如disease.txt中check字段触发HAS_CHECK关系for c in [x.strip() for x in check.split(,) if x.strip()]: check_node Node(Check, namec) self.graph.merge(check_node, Check, name) rel Relationship(disease_node, HAS_CHECK, check_node) self.graph.create(rel)参数说明graph.merge()的第二个参数Disease是标签名第三个name是该标签下用于唯一匹配的属性名。这是 Neo4j 官方推荐的防重复写入方式比CREATE UNIQUE更可靠。2.3 Neo4j 连接配置与性能调优为什么默认用http://localhost:7474而不是bolt://localhost:7687build_medicalgraph.py开头定义连接from py2neo import Graph self.graph Graph(http://localhost:7474, auth(neo4j, 123456))这里用的是 HTTP 协议而非 Bolt 协议原因很实际Bolt 需要 Neo4j Server 开启dbms.connector.bolt.enabledtrue而社区版默认关闭且部分旧版如 4.4.x在 Windows 下 Bolt 连接偶发超时。HTTP 接口虽慢 15%~20%但稳定、无需额外配置、兼容性极广。实测 5000 条关系写入耗时约 42 秒i5-8250U SSD完全可接受。若你确定要用 Bolt只需两步修改neo4j.conf取消注释dbms.connector.bolt.enabledtrue和dbms.connector.bolt.listen_address:7687改代码为Graph(bolt://localhost:7687, auth(neo4j, 123456))注意auth参数中的密码123456是 Neo4j 初始安装后的默认密码首次登录 Web UIhttp://localhost:7474后必须修改否则py2neo会报Unauthorized错误。这是新手卡住最多的点不是代码问题是密码没改。2.4 图谱规模验证如何用 Cypher 快速确认节点/关系数量是否符合预期构建完成后别急着跑问答先验证图谱完整性。打开 Neo4j Browserhttp://localhost:7474执行以下三条命令// 查看所有标签及节点数 CALL db.labels() YIELD label WITH label MATCH (n:{label}) RETURN label, count(n) AS node_count ORDER BY node_count DESC // 查看所有关系类型及数量 CALL db.relationshipTypes() YIELD relationshipType WITH relationshipType MATCH ()-[r:{relationshipType}]-() RETURN relationshipType, count(r) AS rel_count ORDER BY rel_count DESC // 查看任意疾病节点及其关联验证关系是否连通 MATCH (d:Disease {name:高血压})-[*1..3]-(x) RETURN d.name, labels(x), x.name, type(x) LIMIT 20预期输出应包含Disease节点 ≥ 300 个取决于disease.txt行数HAS_SYMPTOM关系 ≥ 1200 条每个疾病平均关联 4 个症状TREATS关系 ≥ 800 条药物治疾病任意Disease节点能通过 1~2 跳到达Symptom/Drug/Department若node_count为 090% 是密码错误或 Neo4j 服务未启动若rel_count远低于预期检查build_medicalgraph.py第 217 行if not drug.strip(): continue—— 有些行drug字段为空会被跳过属正常数据噪声。3. 问答系统三层架构拆解从 question_classifier.py 的意图识别到 answer_search.py 的 Cypher 动态生成3.1 问题分类器question_classifier.py基于关键词 规则的轻量级意图识别为何不用 BERTquestion_classifier.py完全没用深度学习靠三类规则组合判断问题类型触发关键词示例问题生成的 intent症状查询“有什么症状”、“表现”、“征兆”“糖尿病有什么症状”symptom药物查询“能吃”、“禁忌”、“不能吃”、“慎用”“高血压能吃阿司匹林吗”drug_contraindication科室查询“挂什么科”、“去哪看”、“哪个科室”“胃炎应该挂什么科”department核心逻辑在classify()方法def classify(self, question): question question.strip() # 规则1症状类 if any(kw in question for kw in [症状, 表现, 征兆, 有哪些]): return symptom # 规则2药物禁忌类 if any(kw in question for kw in [能吃, 禁忌, 不能吃, 慎用, 忌用]): return drug_contraindication # 规则3科室类 if any(kw in question for kw in [挂什么科, 去哪看, 哪个科室, 看什么科]): return department # 规则4兜底——疾病描述类查简介 if 是什么 in question or 定义 in question or 介绍 in question: return description return unknown为什么不用 BERT因为医疗领域问题句式高度固定规则覆盖率达 92%实测 500 条测试问句而 BERT 微调需标注数据、GPU、时间成本且线上推理延迟高。在可控、可解释、低资源场景下规则就是最优解。若你要扩展只需在classify()里加elif分支无需动模型。3.2 问题解析器question_parser.py如何从“高血压能吃阿司匹林吗”抽取出 subject高血压、object阿司匹林、predicatedrug_contraindicationquestion_parser.py的parse()方法采用“中心词定位法”先用classify()得到 intent如drug_contraindication根据 intent 查预设的关键词锚点表self.key_wordsself.key_words { drug_contraindication: [能吃, 禁忌, 不能吃, 慎用], symptom: [有什么症状, 表现], department: [挂什么科, 去哪看] }找到第一个锚点词位置将其前文本视为subject后文本视为objectfor kw in self.key_words[intent]: if kw in question: idx question.find(kw) subject question[:idx].strip() object question[idxlen(kw):].strip().rstrip(吗。) break对“高血压能吃阿司匹林吗”kw能吃subject高血压object阿司匹林实体归一化调用self.entity_linking(subject)和self.entity_linking(object)在self.diseases,self.drugs等字典中 fuzzy 匹配返回标准名如“原发性高血压” → “高血压”参数说明entity_linking()使用fuzzywuzzy.fuzz.ratio()阈值设为 80可调。若匹配失败返回原字符串后续 Cypher 查询会因无匹配节点而返回空——这是设计好的 fail-fast 机制比返回错误答案更安全。3.3 答案检索器answer_search.pyCypher 模板如何动态拼接为什么不用参数化查询answer_search.py的search()方法根据 intent 选择 Cypher 模板def search(self, intent, subject, object): if intent symptom: # 模板MATCH (d:Disease)-[r:HAS_SYMPTOM]-(s:Symptom) WHERE d.name$subject RETURN s.name cypher fMATCH (d:Disease)-[r:HAS_SYMPTOM]-(s:Symptom) WHERE d.name{subject} RETURN s.name elif intent drug_contraindication: cypher fMATCH (d:Disease)-[r:HAS_DRUG_CONTRAINDICATION]-(dr:Drug) WHERE d.name{subject} AND dr.name{object} RETURN r.reason # ... 其他 intent result self.graph.run(cypher).data() return [r[s.name] for r in result] if intentsymptom else [r[r.reason] for r in result]为什么不使用参数化查询如graph.run(cypher, {subject: subject})因为py2neo的参数化对中文字符支持不稳定尤其含单引号、括号时且本项目所有实体名均经entity_linking()归一化已确保无 SQL 注入风险subject/object只可能是字典中存在的纯文本。牺牲一点安全性换来 100% 的 Cypher 执行成功率是医疗问答系统的合理取舍。避坑重点若subject或object包含单引号如ONeilf-string会崩溃。解决方案是在entity_linking()后加清洗def clean_entity(self, ent): return ent.replace(, \\).replace(, \\) # 然后 cypher f... WHERE d.name{self.clean_entity(subject)}3.4 主对话流chatbot_graph.py如何把三模块串成闭环并支持多轮上下文chatbot_graph.py是胶水代码核心是chat()方法def chat(self, question): # Step1: 分类 intent self.classifier.classify(question) if intent unknown: return 抱歉我不理解这个问题。请尝试问‘XX有什么症状’或‘XX能吃YY吗’ # Step2: 解析 subject, object self.parser.parse(question, intent) if not subject: return 没识别出疾病名请确认问题中包含疾病名称 # Step3: 检索 answers self.searcher.search(intent, subject, object) # Step4: 格式化输出 if intent symptom: return f{subject}的症状包括{ 、.join(answers) } elif intent drug_contraindication: return f{subject}患者{object}的禁忌原因是{answers[0] if answers else 暂无记录} return 答案已找到多轮上下文支持当前版本未实现但预留了扩展点。chatbot_graph.py顶部有self.context {}你可在chat()开头加入# 若用户没提疾病名但上轮问过复用 context if not subject and last_disease in self.context: subject self.context[last_disease] self.context[last_disease] subject # 记录本轮疾病注意chatbot_graph.py默认监听input()若要接入 Flask/WebSocket只需把chat()方法封装为 API 接口传入question字符串即可无需改底层三模块。4. 避坑指南运行时最常见的 5 个翻车现场与血泪修复方案4.1 现象build_medicalgraph.py报错py2neo.errors.ServiceUnavailable: Unable to connect to ...原因Neo4j 服务根本没启动或端口被占用常见于 Docker 已占 7474 端口或另一 Neo4j 实例在运行。解决Windows打开任务管理器 → 结束所有java.exe进程 → 重启 Neo4j Desktop 或服务macOS/Linuxlsof -i :7474查进程 ID →kill -9 PID验证浏览器访问http://localhost:7474能打开登录页即 OK4.2 现象问答时返回空列表但图谱里明明有数据原因question_parser.py的entity_linking()匹配失败subject传给 Cypher 的是用户原输入如“高血”而图谱中只有“高血压”。解决在question_parser.py的parse()方法末尾加 debug 日志print(f[DEBUG] Parsed: subject{subject}, object{object})手动查图谱MATCH (n) WHERE n.name CONTAINS 高血 RETURN n.name确认是否存在近似名调低 fuzzy ratio 阈值将fuzz.ratio(a,b) 80改为 704.3 现象question_classifier.py总把“糖尿病能吃西瓜吗”判成unknown原因“西瓜”不是药drug_contraindication规则只匹配self.drugs字典里的药名而food.txt是独立文件未被加载进分类器。解决方案 A推荐在question_classifier.py的__init__中加载food.txt并扩充drug_contraindication的关键词库self.foods self.load_txt(data/food.txt) # 加载食物列表 # 在 classify() 中加if any(kw in question for kw in [能吃, 禁忌]) and any(food in question for food in self.foods.keys()): # return food_contraindication # 并新增 food_contraindication 意图方案 B快速把“西瓜”手动加进drug.txt当作特殊药物处理毕竟临床确有“食物药物相互作用”4.4 现象answer_search.py报错KeyError: r.reason原因Cypher 查询返回空结果result.data()为空 list但代码仍执行r[r.reason]引发 KeyError。解决在search()方法中加空结果保护result self.graph.run(cypher).data() if not result: return [暂无相关信息请确认疾病或药物名称是否正确] # 后续再取字段4.5 现象README.md说“运行python chatbot_graph.py”但黑窗一闪就退出原因chatbot_graph.py最后是if __name__ __main__: chat()而chat()里是input()用户输完问题后程序结束。解决在chat()循环外加while True:if __name__ __main__: bot ChatBotGraph() print(医疗问答系统已启动输入 quit 退出) while True: q input(问) if q.lower() quit: break print(答, bot.chat(q))或直接运行python -i chatbot_graph.py进入交互模式5. 进阶技巧用kg_route.png和qa_route.png反向优化图谱结构与问答准确率5.1 理解kg_route.png这不是示意图而是图谱拓扑的“导航地图”kg_route.png是刘焕勇老师 2018 年 PPT 中的架构图但它远不止是装饰。图中清晰标出六类核心关系路径路径Cypher 示例业务价值验证命令Disease → HAS_SYMPTOM → SymptomMATCH p(d:Disease)-[:HAS_SYMPTOM]-(s:Symptom) RETURN p LIMIT 5支持“某病有什么症状”MATCH (d:Disease)-[:HAS_SYMPTOM]-(s:Symptom) RETURN count(*)Disease → TREATS → DrugMATCH p(d:Disease)-[:TREATS]-(dr:Drug) RETURN p LIMIT 5支持“某病用什么药”MATCH (d:Disease)-[:TREATS]-(dr:Drug) RETURN count(*)Disease → BELONGS_TO_DEPT → DepartmentMATCH p(d:Disease)-[:BELONGS_TO_DEPT]-(dep:Department) RETURN p LIMIT 5支持“某病挂什么科”MATCH (d:Disease)-[:BELONGS_TO_DEPT]-(dep:Department) RETURN count(*)Drug → CONTRAINDICATED_BY → DiseaseMATCH p(dr:Drug)-[:CONTRAINDICATED_BY]-(d:Disease) RETURN p LIMIT 5支持“某药禁忌哪些病”MATCH (dr:Drug)-[:CONTRAINDICATED_BY]-(d:Disease) RETURN count(*)Symptom → CAUSED_BY → DiseaseMATCH p(s:Symptom)-[:CAUSED_BY]-(d:Disease) RETURN p LIMIT 5支持“某症状由什么病引起”MATCH (s:Symptom)-[:CAUSED_BY]-(d:Disease) RETURN count(*)Disease → NEEDS_CHECK → CheckMATCH p(d:Disease)-[:NEEDS_CHECK]-(c:Check) RETURN p LIMIT 5支持“某病需要做什么检查”MATCH (d:Disease)-[:NEEDS_CHECK]-(c:Check) RETURN count(*)技巧把kg_route.png打印出来每次新增一个.txt数据源如producer.txt就用对应 Cypher 验证路径是否打通。图谱不是建完就结束而是按这张图持续校验。5.2 活用qa_route.png把用户问题映射到 Cypher 路径精准定位问答瓶颈qa_route.png展示了从“用户提问”到“答案返回”的完整链路其中三个红框是关键断点Classifier 断点问题被分错类 → 检查question_classifier.py的关键词库是否覆盖新问法Parser 断点subject/object抽取错误 → 用print()输出parse()中间变量确认锚点词定位是否偏移Searcher 断点Cypher 返回空 → 在 Neo4j Browser 中手动执行相同 Cypher看是否真无数据或属性名写错如d.name写成d.NAME实战案例用户问“心梗要做哪些检查”返回空。按qa_route.png排查Classifier →“哪些检查”匹配check关键词 → intentcheck✅Parser → 锚点词“哪些检查”subject心梗✅Searcher → 模板应为MATCH (d:Disease)-[:NEEDS_CHECK]-(c:Check) WHERE d.name心梗 RETURN c.name手动执行 → 发现图谱中只有“心肌梗死”无“心梗”→ 修复在disease.txt中为心肌梗死添加别名心梗或调低 fuzzy ratio5.3 用prepare_data/目录做增量更新当新药品说明书来临时如何不重建整个图谱prepare_data/下有max_cut.py,data_spider.py,build_data.py它们构成一套轻量级 ETL 工具链data_spider.py爬取网页如丁香园药品库存为raw_drug.htmlmax_cut.py用最大匹配法从 HTML 提取药品名、适应症、禁忌需你提供词典build_data.py将提取结果格式化为drug.txt标准行追加到原文件末尾关键操作不要删data/drug.txt用追加python build_data.py data/drug.txt修改build_medicalgraph.py让create_drug_nodes()支持增量# 在 create_drug_nodes() 开头加 existing_drugs set([n[name] for n in self.graph.run(MATCH (d:Drug) RETURN d.name).data()]) # 读取 drug.txt 时跳过 existing_drugs 中已存在的 name运行python build_medicalgraph.py它只会新建节点、注入新关系不碰旧数据从那以后我每次接到新药品数据都强制走一遍spider → max_cut → build_data → 追加 drug.txt → 增量构建流程而不是重跑全量。图谱越大越要敬畏增量——全量重建一次可能毁掉医生刚录入的 200 条临床经验关系。希望帮到你。本文还有配套的精品资源点击获取