简介这是一份基于Neo4j图数据库的医疗知识图谱智能问答机器人Python项目源码专为计算机专业毕业设计、期末大作业及知识图谱实战学习者设计。项目涵盖了医疗数据清洗、图谱构建、问句意图识别、CQL查询生成、答案匹配等完整流程难度适中可帮助读者快速掌握医疗知识图谱问答系统的开发思路。压缩包共36个文件主要包括8个Python源码文件、5个编译后的pyc文件、10个txt说明文档另有前端展示所需的HTML、CSS、JS与图表资源整体约15.34MB结构清晰便于按模块学习。项目源码均已在本地编译运行通过并附带详细使用说明能有效降低环境搭建与调试门槛。目前已有296人学习使用适合需要完整可运行项目作为参考的开发者。1. 医疗知识图谱问答的落地路径比模型更重要的是图谱和查询设计把“高血压吃什么药”这类问题交给一个医疗问答机器人表面上考验的是模型实际跑通后你会发现瓶颈通常不在模型而在知识表达和数据质量。一个基于Neo4j的医疗知识图谱问答系统核心链路是“问句 - 意图识别 实体链接 - Cypher 查询 - 答案组装”其中实体链接和查询模板设计直接决定回答准不准。这类项目也常作为 Python 知识图谱课程设计和毕业设计的选题因为源码结构清晰、效果可感且每一步都能独立验证。这个项目适合三类人想掌握知识图谱工程化落地方式的 Python 工程师需要把医学词典或临床数据组织成可查询结构的开发人员以及正在做智能问答相关课程设计的学生。你不需要在大模型上烧钱用规则加词典就能覆盖相当比例的垂直领域问法关键是建模和模板要稳。接下来顺着“建模 - 导入 - 识别 - 查询 - 对话”这条线完整走一遍。2. 医疗知识图谱建模实体关系设计要能支撑问答查询2.1 实体类型和关系类型先想清楚要被怎么查医疗知识图谱的建模不是把数据塞进图里就结束而是反推“哪些问法要能被回答”。常见问法包括疾病介绍、症状反查疾病、疾病用药、药品用法、疾病饮食禁忌和检查项目。针对这些问法我一般会定义四类核心实体以及它们之间的五类关系。标签核心属性数据来源Diseasename、icd10、departmentICD-10字典、临床路径Symptomname症状标准化词表Medicinename、usage、dosage、note药品说明书、处方集CheckItemname、desc检查项目列表关系类型起点终点关键属性HAS_SYMPTOMDiseaseSymptomweight 症状相关权重HAS_DRUGDiseaseMedicinegroup 一线/二线用药CHECK_OFDiseaseCheckItemlevel 检查目的FORBID_FOODDiseaseFoodreason 禁忌原因INTERACTIONMedicineMedicinedesc 相互作用描述建模时要特别注意同义概念的归一化。“高血压”和“原发性高血压”在百科里可能是两个词条但在问答里应合并到同一个 Disease 节点用alias数组属性存放别名而不是建两个节点。否则实体链接会命中多个节点查询结果重复且无法消歧。另一个常见错误是把“糖高”这种口语化简称直接当实体结构正确的做法是放在同义词表里在导入前统一映射到标准实体名。2.2 通过 CSV 批量导入 Neo4j约束、索引和 LOAD CSV 的顺序数据准备阶段我会把清洗好的节点和关系分别导出为 CSV 文件再通过LOAD CSV导入 Neo4j。先建立约束和索引再导数据这样可以确保MERGE不会产生重复节点。Neo4j 5.x 的约束语法如下CREATE CONSTRAINT disease_name IF NOT EXISTS FOR (d:Disease) REQUIRE d.name IS UNIQUE; CREATE INDEX symptom_name IF NOT EXISTS FOR (s:Symptom) ON (s.name);参数说明IF NOT EXISTS保证脚本可重复执行REQUIRE d.name IS UNIQUE是 5.x 的写法4.x 用的是ASSERT d.name IS UNIQUE两个版本的语法不兼容升级版本时这是最常见的报错点。索引针对的是查询入口属性凡是MATCH里用等值匹配的字段都应建索引否则数据量上万后每条查询都会全库扫描。节点导入使用MERGE而不是CREATE目的就是配合唯一约束做幂等写入USING PERIODIC COMMIT 500 LOAD CSV WITH HEADERS FROM file:///disease.csv AS row MERGE (d:Disease {name: trim(row.name)}) SET d.icd10 row.icd10, d.department row.department, d.summary row.summary;USING PERIODIC COMMIT是旧版分批提交的写法在 4.4 之后已不推荐但很多教程仍在使用。实际上 Neo4j 4.0 之后LOAD CSV默认就是在独立事务里提交不需要手动指定。保留这个子句对数据一致性没有破坏只是不再有实际性能提升。关系导入的核心是先定位两端的实体节点再做MERGE。这里最容易踩的坑是 CSV 里的实体名和库中节点名有空格或大小写差异所以在节点导入时用trim()清洗在关系导入时也保持一致LOAD CSV WITH HEADERS FROM file:///rel_symptom.csv AS row MATCH (d:Disease {name: trim(row.disease)}) MATCH (s:Symptom {name: trim(row.symptom)}) MERGE (d)-[r:HAS_SYMPTOM {weight: toFloat(row.weight)}]-(s);参数说明toFloat把 CSV 里的字符串权重转成数值类型便于后续排序MATCH找不到节点时该行会被跳过不会报错但会导致关系缺失。所以建议导入完成后用MATCH (d:Disease)-[r:HAS_SYMPTOM]-(s:Symptom) RETURN count(r)核对关系数量与 CSV 行数对比判断是否有实体名没对上。如果数据源是 Excel 或数据库直接用 Python 做预处理再导出 CSV。这里常见的做法是用 pandas 清洗字段并剔除重复列。注意导出时 CSV 编码建议用 UTF-8并去掉 BOM 头否则第一列列名多出不可见字符LOAD CSV WITH HEADERS会报列名不匹配。import pandas as pd df pd.read_excel(medical_data.xlsx) rel df[[disease, symptom]].drop_duplicates() rel.to_csv(rel_symptom.csv, indexFalse, header[disease, symptom])2.3 Neo4j Browser 只显示 25 个标签不代表数据没导入很多人在 Neo4j Browser 里执行MATCH (n) RETURN n LIMIT 50发现画布上只有 25 个节点就误以为导入失败。这不是数据问题而是 Browser 在可视化时的默认渲染上限。确认数据是否完整应该执行统计查询而不是看画布MATCH (n:Disease) RETURN count(n) AS disease_count;Neo4j Browser 的设置里有一个“Max number of nodes”默认是 25 或 50可以调大但节点变多后浏览器渲染会明显卡顿。实际做图谱探索时更建议围绕一个疾病中心查询其 2 跳以内的子图既能看到完整关联又不会一次性加载全库。想要更专业的可视化效果可以接 Neo4j Bloom 或 Gephi但那偏向展示场景日常问答开发中用 Cypher 查询就够了。3. 智能问答的意图识别和实体链接先规则后模型3.1 意图体系设计把问法归类成有限集合医疗问答的意图不需要太多控制在六个以内反而准确率更高。意图设计过多会导致训练语料稀疏规则也难以覆盖。我的建议是分为疾病介绍、症状查疾病、疾病查用药、药品用法、疾病饮食禁忌、疾病检查六类。每类意图对应一类查询模板同时配合一组典型的触发词组。意图 ID触发问法示例查询侧重点disease_intro什么是高血压 / 讲讲糖尿病疾病定义、科室symptom_to_disease头晕是什么病 / 哪些病会发热症状反查疾病disease_to_drug高血压吃什么药疾病到药品drug_usage阿司匹林怎么吃药品用法、剂量disease_food糖尿病不能吃什么疾病饮食禁忌disease_check高血压需要做什么检查疾病检查项目意图识别的实现上先做关键词命中再做正则兜底最后才考虑模型。比如问句包含“什么药”“吃什么”“用药”时优先归入 disease_to_drug包含“怎么吃”“用法”时归入 drug_usage。这两类容易混淆需要靠位置信息区分如果问句中先出现疾病实体再出现“什么药”走疾病查药如果先出现药品实体再出现“怎么吃”走药品用法。3.2 用 AC 自动机做词典实体链接实体链接是整个问答链路中最影响结果准确率的模块。医疗领域实体词表通常有几千到几万条线性匹配每条问句的效率太低。我会用pyahocorasick构建 AC 自动机一次扫描完成所有词典条目的匹配。import ahocorasick def build_automaton(entities): automaton ahocorasick.Automaton() for idx, entity in enumerate(entities): automaton.add_word(entity, (idx, entity)) automaton.make_automaton() return automaton def extract_entities(text, automaton): hits [] for end_index, (_, entity) in automaton.iter(text): start_index end_index - len(entity) 1 hits.append((start_index, end_index, entity)) return hits代码逻辑说明add_word将每个词条加入自动机make_automaton构建失败指针iter在文本上做线性扫描返回每个匹配的结束位置和词条内容。这个方案的时间复杂度是 O(n)词表规模对性能影响很小。运行结果会出现重叠匹配比如词典里同时有“高血压”和“高血压病”用户输入“高血压病”时两个词条都命中。这里需要做最长匹配过滤对同一个结束位置只保留最长的实体再按起始位置去重。逻辑如下def merge_hits(hits): hits.sort(keylambda x: (x[1], x[0])) merged [] for h in hits: if not merged: merged.append(h) continue # 当前命中被上一个覆盖则跳过 if h[0] merged[-1][0] and h[1] merged[-1][1]: continue if h[1] merged[-1][1] and h[0] merged[-1][0]: continue merged.append(h) return merged这里最关键的是实体类型的判定。因为不同意图需要识别不同类型的实体比如 symptom_to_disease 需要从问句中提取症状实体disease_to_drug 需要提取疾病实体。如果同一段文本命中两个实体例如“高血压”既是 Disease 又是 Symptom因为高血压本身也是症状词就需要靠实体类型字典做一次过滤优先匹配当前意图对应的类型。3.3 同义词和别名医疗场景的准确率关键医疗问法的口语化程度很高“血压高”“高血压”“HBP”指向同一实体“糖尿病”的别名有“消渴症”。我维护一张同义词映射表在实体链接前先对问句做替换归一化再进入 AC 自动机匹配。synonym_map { 血压高: 高血压, 血糖高: 高血糖, 消渴症: 糖尿病, 阿司匹林肠溶片: 阿司匹林 } def normalize_question(question): for alias, standard in synonym_map.items(): question question.replace(alias, standard) return question参数说明同义词替换要在实体匹配之前执行否则“血压高”无法命中词典中的“高血压”。替换的顺序也有讲究长词优先替换比如先替换“阿司匹林肠溶片”再替换“阿司匹林”避免“阿司匹林”被先替换后导致长词无法匹配。同义词表还会影响查询结果的组织方式。当某个疾病节点通过别名命中后答案里应该指出标准名和别名之间的关系避免用户觉得答非所问。比如回答“高血压又称原发性高血压的常用药物包括……”。3.4 实体无法命中时的兜底模糊相似度匹配总会有词表覆盖不到的问法。对于未登录词我会先用编辑距离或序列相似度在候选实体中找最接近的但阈值要收得比较紧否则会把无关实体链接进来。这里用rapidfuzz的fuzz.ratio做简单兜底from rapidfuzz import fuzz def fuzzy_match_entity(question, candidate_entities, threshold80): best (, 0) for entity in candidate_entities: score fuzz.partial_ratio(question, entity) if score best[1]: best (entity, score) if best[1] threshold: return best[0] return None这个兜底方案有问题需要说明候选实体集合如果很大计算量是 O(n) 次相似度计算只适合在 AC 自动机命中结果为空时使用。可以把候选集限制为高频实体前 200 条或者按问句里的关键词先粗筛一遍再算相似度。模糊匹配得到的实体需要在下游查询里验证如果 Cypher 返回空结果就不要把这个实体当作有效链接结果输出。4. 把意图变成 Cypher 查询模板并组装答案4.1 查询模板配置表把 Cypher 和代码解耦实体链接完成后下一步是把“意图 实体”组合成可执行的 Cypher 查询。这里的设计原则是查询模板不写在 Python 代码里而是放在独立的 JSON 配置文件中每个意图对应一条或多条模板模板中用参数占位符代替实体值。{ disease_to_drug: { template: MATCH (d:Disease {name: $entity})-[r:HAS_DRUG]-(m:Medicine) RETURN m.name AS drug, m.usage AS usage, m.dosage AS dosage ORDER BY r.group, answer: {entity}的常用药物有{drug}{dosage}用法{usage}。 }, symptom_to_disease: { template: MATCH (d:Disease)-[r:HAS_SYMPTOM]-(s:Symptom {name: $entity}) RETURN d.name AS disease, r.weight AS weight ORDER BY r.weight DESC LIMIT 5, answer: 出现{entity}可能相关的疾病有{disease}。 } }参数说明$entity是 Cypher 参数占位符运行查询时通过参数传入实体值。这样做能防止用户输入直接拼进查询字符串导致注入风险同时避免拼接时转义特殊符号的问题。4.2 Python 端执行查询并组装返回文本使用官方驱动neo4j连接图数据库。每个问答请求创建一个 session 并在使用结束后自动关闭。模板中的$entity参数通过session.run的第二个参数传入。from neo4j import GraphDatabase class MedicalQA: def __init__(self, uri, user, password): self.driver GraphDatabase.driver(uri, auth(user, password)) def run_template(self, cypher, entity): with self.driver.session() as session: result session.run(cypher, entityentity) return [record.data() for record in result] def close(self): self.driver.close()使用说明session.run的第二个参数entityentity对应模板中的$entity字段名必须完全一致。record.data()会返回一个 Python 字典键为 RETURN 子句中的别名。如果要返回多列数据直接通过字典取值组装答案。4.3 答案去重、多候选合并与空结果处理查询返回多条记录时直接拼接会让答案变得冗长。常见的处理方式是按字段去重并对同类结果做合并。def format_answer(records, answer_template, entity): seen set() items [] for record in records: key tuple(record.values()) if key not in seen: seen.add(key) items.append(record) if not items: return 抱歉暂时没有找到与{}相关的信息。.format(entity) text for record in items: text answer_template.format(entityentity, **record) return text.rstrip()关于空结果直接回答“没有找到”对用户体验很差。我会加一层兜底查询先用精确匹配查一次无结果时用CONTAINS做模糊匹配。但要注意CONTAINS无法利用普通索引数据量大时会变慢。更稳妥的做法是给实体名建全文索引。CREATE FULLTEXT INDEX disease_fulltext IF NOT EXISTS FOR (n:Disease) ON EACH [n.name];全文索引创建后可以用db.index.fulltext.queryNodes做分词匹配适合处理问句中实体不完整的情况。比如用户说“高血”而不是“高血压”全文索引能返回包含“高血”的节点名称。4.4 查询模板的调试技巧开发阶段最耗时的就是 Cypher 模板写错。我通常先在 Neo4j Browser 里用固定实体值跑通模板再复制到 JSON 配置里。一个必须养成的习惯是模板里的属性名必须与导入时的属性名完全一致比如导入时用了name模板里写title就会返回空。另外模板配置做热更新或者版本控制不要直接在数据库客户端里改完就忘。把 JSON 配置文件纳入仓库每次修改记录对应答效果的影响。模板里出现多跳路径时先确认每一跳的方向和关系类型。5. 多轮对话中的指代消解与问答回归验证多轮对话是本项目最容易被忽略但提升明显的部分。用户第一轮问“高血压吃什么药”第二轮追问“孕妇能吃吗”这里“这药”指代上一轮回答中的药品或者“能吃吗”省略了主语。简单的指代消解不需要模型只需要维护当前会话的上下文实体状态。每个会话对象保存上一次生效的实体和实体类型当新一轮问句没有命中任何实体时自动继承会话上下文中的实体并把意图重新判定为承接上一轮。具体实现是在问答类的查询入口加一个会话级缓存字段。实体链接结果为空时检查当前会话的context_entitydef answer(self, question, session_id): entity self.extract(question) context self.session_store.get(session_id, {}) if not entity and context.get(entity): entity context[entity] # 命中实体后更新会话上下文 self.session_store[session_id] {entity: entity}这个方案要设置有效期比如会话超过 5 分钟未活动就清空上下文避免不同用户共用同一状态。多轮对话中还有一个细节如果问句里又明确出现了新实体必须用新实体替换旧上下文不能继续沿用上一轮的。问答系统的验证不能只看 demo。我会构造一组回归测试集覆盖每个意图 10 到 20 种问法脚本跑完统计两个指标意图识别准确率和回答有效率。回答有效率指系统最终是否返回了非空结果它比意图准确率更能反映整体链路问题。import json import csv test_cases [ {question: 什么是高血压, intent: disease_intro, entity: 高血压}, {question: 头晕是什么病, intent: symptom_to_disease, entity: 头晕}, {question: 糖尿病可以吃什么药, intent: disease_to_drug, entity: 糖尿病}, ] def run_evaluation(qa): rows [] for case in test_cases: answer qa.answer(case[question], session_idtest) rows.append((case[question], case[entity] in answer if answer else False)) return rows运行后将结果追加写入 CSV 文件每次修改模板或同义词表后重新执行通过对比差异就能快速定位是哪一条规则改坏了。这套方法比肉眼测试更可靠也适合在课程设计中作为项目效果的数据支撑。本文还有配套的精品资源点击获取