简介:一套基于Python与Neo4j图数据库的医疗知识图谱智能问答机器人毕业设计项目,面向计算机相关专业正在准备毕业设计的学生,以及需要完整实战练习的编程学习者,也可直接用于课程设计或期末大作业。项目代码经过严格调试,可正常运行。压缩包共36个文件,约15.34MB,核心为8个Python源码文件,覆盖问题分析、医疗图谱构建、查询生成、答案检索等完整问答流程;另附10个txt说明文档、7张图片以及配套前端样式与脚本,方便对照理解与二次开发。当前已有701人学习下载。代码附带超详细中文注释与使用说明,重点模块的图谱构建、问答主流程均清晰标注处理逻辑,适合参考其图数据库建模思路和意图识别设计,并可直接在此基础上扩展药品、疾病、症状等知识节点。
1. 医疗知识图谱问答机器人:先建图,再让问句找到答案
拿到这份 Python 基于 Neo4j 图数据库的医疗知识图谱智能问答机器人源码,最容易犯的错误是一上来就研究问答代码。把问答拆开看,七成工作量在建图,三成在匹配,真正跑通之后你会发现:图建对了,问答就成功了一半。它解决的是医疗信息查询里“回答可解释、可溯源”的问题——比如患者问“糖尿病能不能吃西瓜”,系统不是甩出一段搜索摘录,而是给出“糖尿病—禁忌食物—西瓜”这条关系路径,答案自带依据。适合正在做知识图谱练手或毕设、想从普通增删改查转向图查询的 Python 开发者,也适合医疗信息化场景做技术验证。下文按“建模→导入→问句解析→查询生成→排错→验证”的顺序把整条链路说透。
2. 医疗知识图谱建模与 Neo4j 落地:先把图建对,再谈问答
建图不是把数据塞进 Neo4j 就完事,实体怎么分、关系朝哪个方向、别名怎么归一,直接决定后面问答代码的复杂程度。顺序反了,后面每写一个查询模板都要给前面的设计买单,这是新手最容易翻车的地方。
2.1 实体与关系怎么定:按“用户会怎么问”取舍本体
医疗知识图谱本体不需要做到教科书级别,够用就好。常见的最小实体集是:疾病 Disease、症状 Symptom、药品 Medicine、科室 Department、检查项目 CheckItem,按业务需要再加食物 Food。每个实体至少带 name(唯一)、alias_list(别名列表)、desc(简述)三个属性,来源字段看场景加。
关系按“问句里会出现的主语和宾语”来定义,方向统一才能让模板好写:
- (Disease)-[:HAS_SYMPTOM]->(Symptom):症状寻因和疾病查症共用一条边
- (Medicine)-[:TREAT]->(Disease):治什么病
- (Medicine)-[:CONTRAINDICATED]->(Disease):禁忌,必须单独建关系,不能复用 TREAT 反向
- (Disease)-[:VISIT_DEPT]->(Department):挂什么科
- (Disease)-[:FORBID_FOOD]->(Food):忌口什么
同义词归一化是医疗图谱特有的前置工作。“糖尿病”“2型糖尿病”“消渴症”在口语里可能指同一个病,“拜糖平”和“阿卡波糖”是同一成分的不同商品名。常见做法是每个实体维护一个 alias 属性,导入时把别名合并到节点上,问答层统一查标准名,匹配层负责把口语映射过去。这里有一个注意点:关系方向全篇保持一致,不要“药品治疾病”和“疾病被药品治”混着建,否则 4.1 的模板会非常难写。
属性取舍上只导查询和展示会用的字段,别把整库数据搬进来。医疗数据源经常有大段的说明文本,真需要全文展示就放 desc 字段,其他中间字段一律不建索引,减少导入时间和存储占用。
2.2 数据导入的两条路线:LOAD CSV 与 Python 驱动
小规模数据(几百到几万节点)用 LOAD CSV 最省事,在 Neo4j Browser 或 cypher-shell 里就能直接执行。先给两张表:disease.csv 放节点字段,relation.csv 放关系三元组。
LOAD CSV WITH HEADERS FROM 'file:///disease.csv' AS row MERGE (d:Disease {name: row.name}) SET d.alias = row.alias, d.desc = row.desc, d.source = row.source; LOAD CSV WITH HEADERS FROM 'file:///relation.csv' AS row MATCH (d:Disease {name: row.disease_name}) MATCH (s:Symptom {name: row.symptom_name}) MERGE (d)-[:HAS_SYMPTOM]->(s);逻辑说明:LOAD CSV 是 Neo4j 内置的批量导入命令,WITH HEADERS 让首行成为字段名;MERGE 按 name 判重,重复执行不会堆出重复节点;第二条语句先定位两个端点再建关系,避免用 CREATE 造成重复关系。文件默认放在 Neo4j 的 import 目录下,file:/// 后面跟相对路径即可。
参数说明:文件编码必须 UTF-8,Excel 另存的 CSV 大概率是 GBK,直接导会乱码(见 5.2);name 字段建议用医疗标准术语,别名放 alias 列;MERGE 依赖唯一约束,最好在建表前先加约束,否则数据量上去后判重性能会退化。
要走定时更新或和爬虫结果联动,就用 Python 官方驱动做自动化导入:
from neo4j import GraphDatabase URI = "bolt://localhost:7687" AUTH = ("neo4j", "password") class KGImporter: def __init__(self): self.driver = GraphDatabase.driver(URI, auth=AUTH) def close(self): self.driver.close() def make_relation(self, med_name, disease_name): with self.driver.session() as session: session.run( """ MATCH (m:Medicine {name: $med}) MATCH (d:Disease {name: $dis}) MERGE (m)-[:TREAT]->(d) """, med=med_name, dis=disease_name, ) if __name__ == "__main__": imp = KGImporter() imp.make_relation("阿卡波糖", "2型糖尿病") imp.close()逻辑说明:neo4j 驱动是官方库,用 $ 占位符传参,避免字符串拼接;session 内部自动管理事务,每次 make_relation 做两次匹配、一次 MERGE,适合增量修正某条关系。
参数说明:URI 里的 7687 是 Bolt 协议默认端口,浏览器访问的 HTTP 端口是 7474,两者别搞混;AUTH 元组里是初始账号密码,第一次启动 Neo4j 时会要求修改;如果入库脚本跑得频繁,可以给 driver 加 max_connection_pool_size 限制连接数,防止连接池占满。
2.3 建完图先验证:三条查询确认结构与质量
问答代码还没写,先把图“验货”。下面三条 Cypher 从规模、路径、空结果三个角度检查。
// 1. 统计各标签节点数量 MATCH (n) RETURN labels(n)[0] AS label, count(*) AS cnt ORDER BY cnt DESC; // 2. 抽查一条医疗路径:糖尿病的症状 MATCH (d:Disease {name: '糖尿病'})-[:HAS_SYMPTOM]->(s:Symptom) RETURN s.name LIMIT 10; // 3. 查糖尿病的禁忌食物,返回空不一定错 MATCH (d:Disease {name: '糖尿病'})-[:FORBID_FOOD]->(f:Food) RETURN f.name LIMIT 10;逻辑说明:第一条确认导入没有漏表,哪个标签数量异常一眼可见;第二条验证关系方向,如果这里返回空,多半是导入脚本里把 HAS_SYMPTOM 建反了;第三条空返回有两种可能,图谱没这个关系,或者问法本身在数据里就没有答案,先别急着改代码。
接下来建唯一约束,防止后续导入脚本重复执行时节点翻倍:
CREATE CONSTRAINT ON (d:Disease) ASSERT d.name IS UNIQUE;参数说明:这个约束语法在 Neo4j 3.x 和部分早期 4.x 版本可用,新版改成了 CREATE CONSTRAINT FOR (d:Disease) REQUIRE d.name IS UNIQUE,版本差异会在 5.1 展开讲。跑完约束后再用 Neo4j Browser 的可视化视图抽查两条带关系的路径,箭头方向能直接暴露导入脚本里的反向错误,这一步比看日志直观得多。
3. 问句解析与实体识别:把“糖尿病能不能吃西瓜”拆成能检索的零件
问答链路里的黑匣子不是 Cypher,而是“从一句话里找到实体并判断意图”。用户问“高血压能不能吃阿司匹林”,句子里有两个实体,系统得知道主实体是阿司匹林、意图是禁忌查询,这步错了,后面图再准也白搭。
3.1 词典+AC自动机先于NER模型的三条理由
医疗问答场景里,词典匹配通常比训练 NER 模型更划算。第一,医疗实体几乎都是专有名词,常见病、常用药几千条别名就能覆盖业务大头,词典覆盖率足够高;第二,标注语料很贵,医疗 NER 要处理超长实体和嵌套实体,冷启动成本不是小项目能接受的;第三,NER 模型输出的是概率,词典输出的是证据——命中哪个词、对应哪个标准名,出了问题可以直接查。
用 pyahocorasick 做多模式匹配,一次遍历就能命中所有词典词条:
import ahocorasick import csv class MedicalNER: def __init__(self, dict_csv): self.auto = ahocorasick.Automaton() with open(dict_csv, encoding="utf-8") as f: for row in csv.DictReader(f): alias = row["alias"].strip() self.auto.add_word( alias, (len(alias), row["entity_type"], row["std_name"]) ) self.auto.make_automaton() def extract(self, question): hits = [] for end_idx, (length, etype, std_name) in self.auto.iter(question): start_idx = end_idx - length + 1 hits.append((start_idx, length, etype, std_name)) hits.sort(key=lambda x: (x[1], -x[0]), reverse=True) return hits逻辑说明:add_word 把每个别名加入自动机,value 里存词长、实体类型、标准名;extract 遍历问句拿到所有命中,按“长词优先”排序,排序后由主流程做重叠过滤,保证“高血压”命中时不会同时保留“高血”这个残缺匹配。
参数说明:dict_csv 至少要有 alias、entity_type、std_name 三列,别名一行一个;如果同一个别名出现两次,后写的会覆盖先写的,导入词典前先做去重;Aho-Corasick 对中文同样适用,不需要额外分词,但要注意别名里带括号或空格时,匹配前最好把问句和词典做同样的归一化处理。
3.2 意图分类:先抓否定词,再按关键词模板分层
意图分类不需要上模型,短文本里关键词互斥性很强,规则模板足够稳。先把意图切成五类:症状查病、疾病查药、药品禁忌、疾病查科室、药品用法。分类顺序很重要,否定和禁忌意图必须最先判断。
NEG_KEYWORDS = ["不能", "不可以", "禁忌", "忌", "禁用", "慎用"] def classify(question, entities): # 否定/禁忌意图优先于治疗意图 for w in NEG_KEYWORDS: if w in question: return "DRUG_CONTRADICTION" if "挂什么科" in question or "挂科" in question or "科室" in question: return "DISEASE_TO_DEPT" if "症状" in question or "表现" in question or "什么病" in question: return "SYMPTOM_TO_DISEASE" if any(t == "Medicine" for _, _, t, _ in entities): if "怎么用" in question or "怎么吃" in question or "剂量" in question: return "DRUG_USAGE" return "DISEASE_TO_MEDICINE"逻辑说明:顺序即优先级,禁忌 > 科室 > 症状 > 药品用法 > 默认治疗。先处理否定词,防止“高血压不能吃什么药”落到治疗意图;句子里带 Medicine 实体且出现用法词时走 DRUG_USAGE,否则默认落到疾病治疗查询,空结果由 4.2 的兜底逻辑处理。
参数说明:NEG_KEYWORDS 要按业务持续维护,漏掉“忌口”比多写一个“不能”代价更大,这类词宁可多写;分类结果建议打进日志,问答回放时能直接看出意图判断错在哪一步。
多实体场景下还要定主实体:一句里出现“高血压”和“阿司匹林”时,意图是 DRUG_CONTRADICTION,主实体取药品而非疾病,因为禁忌关系的起点是药品。常见做法是定义各意图的主实体类型优先级,比如含药品时优先选 Medicine,否则取 Disease 或 Symptom,这个逻辑在 4.1 的 pick_primary_entity 里落地。
4. 从意图到 Cypher 再到答案:模板映射与结果兜底
意图和实体都齐了,下一步是把它们翻译成图查询。这一步最容易写出“能跑但不敢上线”的代码——字符串拼 Cypher、不设 LIMIT、空结果直接抛异常,都是初学者常踩的坑。
4.1 意图模板表与参数化查询,别再拼字符串
每个意图对应一个 Cypher 模板,$entity 指的是标准化后的实体名,也就是 3.1 里 std_name 字段的值。初版模板保持一跳关系就够,扩展多跳留给后面的优化阶段。
| 意图 | 问句示例 | Cypher 模板 | 返回 |
|---|---|---|---|
| SYMPTOM_TO_DISEASE | 头晕伴多饮是什么病 | MATCH (d:Disease)-[:HAS_SYMPTOM]->(s:Symptom) WHERE s.name = $entity RETURN d.name, d.desc | 最多 10 种疾病及简介 |
| DISEASE_TO_MEDICINE | 2型糖尿病吃什么药 | MATCH (m:Medicine)-[:TREAT]->(d:Disease) WHERE d.name = $entity RETURN m.name | 最多 10 种药品 |
| DRUG_CONTRADICTION | 高血压能吃阿司匹林吗 | MATCH (m:Medicine {name:$entity})-[:CONTRAINDICATED]->(d:Disease) RETURN d.name | 禁忌疾病列表 |
| DISEASE_TO_DEPT | 心悸挂什么科 | MATCH (d:Disease {name:$entity})-[:VISIT_DEPT]->(dept:Department) RETURN dept.name | 科室名称 |
| DRUG_USAGE | 阿卡波糖怎么吃 | MATCH (m:Medicine {name:$entity}) RETURN m.usage, m.dosage | 用法用量 |
Python 端把模板存成字典,执行时统一走参数化查询:
TEMPLATES = { "SYMPTOM_TO_DISEASE": ( "MATCH (d:Disease)-[:HAS_SYMPTOM]->(s:Symptom) " "WHERE s.name = $entity RETURN d.name AS name, d.desc AS desc LIMIT 10" ), "DISEASE_TO_MEDICINE": ( "MATCH (m:Medicine)-[:TREAT]->(d:Disease) " "WHERE d.name = $entity RETURN m.name AS name LIMIT 10" ), "DRUG_CONTRADICTION": ( "MATCH (m:Medicine {name:$entity})-[:CONTRAINDICATED]->(d:Disease) " "RETURN d.name AS name LIMIT 10" ), } def query_answer(question, entities, intent): std_name = pick_primary_entity(entities, intent) if not std_name: return [] cypher = TEMPLATES[intent] with driver.session() as session: result = session.run(cypher, entity=std_name) return list(result)逻辑说明:所有实体值全部走 $ 占位符,由驱动安全序列化,药品名里出现引号或百分号也不会破坏查询结构,这是参数化查询对字符串拼接的降维打击;query_answer 按意图选模板,取主实体后执行,主实体为空时直接返回,不发 Cypher。
参数说明:LIMIT 统一控制在 10,防止把几千条关系一次拉回;RETURN 里的 AS 别名方便后续按字段名取值;pick_primary_entity 的规则在 3.2 讲过,药品优先,其次疾病,最后症状,意图不同优先级也不同。
4.2 三态结果与答案包装:有、无、太多
查询结果只有三种情况,分别处理才是完整问答。有结果时做格式化输出;无结果时先区分“图谱缺数据”还是“实体识别错”;结果太多时截断展示并给用户追问空间。
MAX_DISPLAY = 5 def render(records, intent): if not records: return "记录里暂时查不到这个答案,换个说法或检查名称后重试。" names = [r["name"] for r in records[:MAX_DISPLAY]] text = "、".join(names) if len(records) > MAX_DISPLAY: text += f" 等{len(records)}种" return text + "。以上结果仅供查询参考,具体以临床医生意见为准。"逻辑说明:展示层面截断到 5 条,多余的用总数说明,避免答案列表失控;医疗问答必须带免责后缀,这是底线,不能省。
参数说明:MAX_DISPLAY 设 5 是因为中文列表一行放 3 到 4 项,5 条差不多一屏;如果一条查询经常返回 10 条以上,说明图的连接太密,应该从模板侧收敛而不是在展示侧硬撑。
无结果的排查要可操作:拿 std_name 再发一条MATCH (x) WHERE x.name = $entity RETURN labels(x),如果节点不存在,说明词典和图的实体没有对齐;如果节点存在但关系为空,说明模板写的路径或关系方向有问题。这个判断逻辑可以直接写进问答日志,作为诊断字段,排错时少走弯路。
5. 医疗问答项目的五个常见问题与排查记录:版本、编码与实体命中
这一章是给同类项目排障后整理的踩坑记录,每一条都按“现象 → 原因 → 解决”写,你能直接对照自己的报错信息定位。
5.1 Neo4j 版本不一致:3.5 的写法在 5.x 里报错
现象:不少这类项目把 Neo4j 锁在 3.5 版本,而你机器上新装的是桌面版或社区版 5.x。按照旧说明执行 LOAD CSV 或创建约束时,直接报语法错误,提示 unknown function 或 procedure not found。
原因:Neo4j 跨大版本语法不兼容。3.x 时代的CREATE CONSTRAINT ON (d:Disease) ASSERT d.name IS UNIQUE在 4.x 后改成了CREATE CONSTRAINT FOR (d:Disease) REQUIRE d.name IS UNIQUE;另外 3.5 时代有些脚本依赖 APOC 扩展存储过程,新版不会随安装包附带,需要单独装。
解决:锁版本是最省事的路径。Neo4j 官方 Archives 页面能选历史版本,Docker 镜像也保留着对应 tag,按项目说明的版本部署就不会有语法迁移问题。验证型项目不值得把时间耗在升级语法上,这是踩过之后才想明白的后悔药。如果非要跑新版,把查询里的约束语法、存储过程调用统一按新版改写,并确认 APOC 是否已安装。
5.2 CSV 中文变乱码:Excel 保存的 GBK 与 Neo4j 的 UTF-8
现象:LOAD CSV 完成后,中文名称全是方块或乱码,按中文名查询关系全部落空。
原因:Excel 另存的 CSV 默认 ANSI/GBK 编码,Neo4j 按 UTF-8 读取,字节错位导致乱码。
解决:导入前统一转成 UTF-8,用 Python 几行代码搞定。
with open("disease_gbk.csv", encoding="gbk") as f: text = f.read() with open("disease_utf8.csv", "w", encoding="utf-8", newline="") as f: f.write(text)逻辑说明:先用 gbk 把 Excel 导出的文件读回内存,再按 utf-8 写新文件,顺手把表头也一起规范一下。不要在 Neo4j 端硬解编码,导入层统一最省心。
参数说明:newline="" 防止 Windows 下写入时空行被插入;如果源文件本身就是 UTF-8,用 errors="replace" 快速定位坏字节比逐行猜编码更高效。
5.3 实体识别命中“高血”而不是“高血压”
现象:问“高血压注意什么”,实体列表里出现“高血”或“血压”这种子串,按错误实体进图查询返回空结果。
原因:词典里多个别名存在前缀重叠,匹配逻辑只做了单词命中,没有做最长匹配优先。
解决:在 3.1 的 extract 返回后加一个重叠过滤,排序后从左往右扫,重叠区间只保留最长的词条。
best, last_end = [], 0 for start, length, etype, std_name in hits: if start < last_end: continue best.append((std_name, etype)) last_end = start + length逻辑说明:按长词优先排序后,取第一个未重叠的命中,后面与区间重叠的字段全部跳过,保证“高血压”覆盖“高血”。
参数说明:last_end 随采纳词条的结束位置更新,漏掉这一步会出现间隔性跳过错误,比如两个相邻实体只保留第一个。
5.4 一次问答返回上万条路径:图查询没有 LIMIT
现象:单个问题响应时间超过两秒,Neo4j Browser 可视化卡死,日志里频繁出现查询超时告警。
原因:模板里漏写 LIMIT,数据库把相关路径全部展开,多跳查询更明显,路径数呈指数级增长。
解决:所有模板统一加 LIMIT,这是 4.1 已经做好的事;Neo4j 配置里打开事务超时,常见做法是把 dbms.transaction.timeout 设为 5s;Python 驱动连接池上限调小,GraphDatabase.driver 加 max_connection_pool_size=8。多跳查询确实需要时,先用 PROFILE 看 Expand 数量是否异常,再决定要不要加索引。
5.5 重复导入让节点翻倍:MERGE 和唯一约束是两件套
现象:定时导入脚本每次运行后,疾病节点数量翻倍,同名节点散落多处,关系也成对重复。
原因:导入了 CREATE 而不是 MERGE;或者用了 MERGE 但没建唯一约束,并发环境下仍可能产生重复。
解决:导入语句统一用 MERGE;对 name 建唯一约束,语法按 5.1 的版本差异来选;管道跑完做一次幂等校验。
MATCH (d:Disease) WITH d.name AS n, count(*) AS c WHERE c > 1 RETURN n, c;逻辑说明:重复计数是检验导入脚本是否健壮的第一道门,返回空说明节点唯一性没问题;注意唯一约束建立后,更新实体用 SET 改属性,不要删了重建,否则挂在旧节点上的关系会全部断开。
6. 用验证集把问答从“能跑”推到“可信”:回放日志与三个扩展
判断一套问答系统能不能用,标准不是几个测试问句答得好,而是能稳定知道“什么时候答不出”。我一般先准备 50 到 100 条问答对,覆盖五个意图和边界场景,比如否定问法、双实体问法、空结果问法。标好 expect 列后用评估脚本批量跑,正确率、兜底率、错答分布一次算清。
import csv import logging def evaluate(qa_csv, answer_fn): ok = 0 for row in csv.DictReader(open(qa_csv, encoding="utf-8")): expect = row["expect"] got = answer_fn(row["question"]) if expect in got: ok += 1 logging.info("PASS %s", row["question"]) else: logging.warning("FAIL q=%s expect=%s got=%s", row["question"], expect, got) return ok逻辑说明:用隶属判断而不是全等判断,因为回答是列表拼接文本,标准答案只写关键实体或短语即可;logging 同时记录失败样本,方便反查是哪一环出了问题。
回放日志是迭代的核心工具。每一条问答把 question、intent、std_name、返回条数、耗时写一行,每周统计兜底率。兜底率超过三成,优先回看是词典缺词还是模板缺类型;2.1 里提到的同义词问题,会在回放日志里直接暴露成一批“查不到”的 query,按日志补词条比凭空想快得多。
三个扩展方向按性价比排序:第一个是把实体识别接上大模型做增强,词典保底处理高频,大模型处理长尾口语,但知识结果仍以图数据库为准,不要让大模型直接生成三元组入库,医疗场景经不起幻觉;第二个是做多轮追问,拿到 TOP5 疾病后反问症状,用追加实体收窄结果,这是单轮问答走向多轮对话的关键路径;第三个是增量更新管道,从权威药品说明书和临床指南里结构化抽取新增药品与禁忌,定期导入,配合 5.5 的幂等校验保证数据不翻倍。
这个项目我做得不算顺利,最初把大量时间压在 NER 模型选型上,最后发现把词典、模板和回放日志当成一个闭环来迭代,效果提升反而最快。工具会过时,图谱加日志的迭代习惯不会过时。希望帮到你。
本文还有配套的精品资源,点击获取