基于知识图谱的心血管疾病问答系统:Python构建与Neo4j实战
简介这是一套面向计算机、人工智能及相关专业学生与开发者的心血管疾病知识图谱问答系统项目资源适合用作毕业设计、课程设计、作业或项目立项演示也便于初学者进阶学习。资源包共197个文件约5.52MB以84个json与84个csv数据文件为核心覆盖心房颤动、冠心病、心力衰竭、心肌梗死、心律失常、心绞痛等常见心血管疾病的结构化知识另含15个txt说明、6个Python源码文件、4张png与1张gif示意图、1个md文档及1个js脚本构成从数据到问答逻辑的完整链路。已有58人学习关注。下载后可获得可运行的项目源码、配套文档说明与数据集帮助读者理解知识图谱构建、实体关系组织与问答匹配的实现思路并可在现有代码基础上修改扩展实现更多功能。1. 从一份心血管问答源码说起知识图谱到底解决了什么心血管疾病的指南、药品说明书、临床路径散落在几十个 PDF 和表格里患者问一句「房颤合并高血压能不能用某类抗凝药」传统关键词检索只能把含「房颤」「高血压」的段落全捞出来答不出「合并」背后的用药禁忌。这就是我最初做 Python《基于知识图谱的心血管疾病问答系统》的动机把疾病、症状、药品、检查、禁忌这些实体和它们之间的关系显式存下来让机器沿着关系推理而不是靠字面匹配。这套方案适合两类人一类是手里有科室数据、想验证知识图谱问答可行性的医疗信息化从业者另一类是正在找知识图谱构建 智能问答系统完整练手项目的 Python 学习者。项目源码、文档说明、数据集三件套的价值在于它把「本体建模 → 图谱构建 → 问答推理」这条链路完整跑通了一遍而不是只给你一个 Neo4j 截图。下面我按自己复现这类项目的顺序把每一步的参数、坑和验证方法讲清楚。2. 本体建模与数据准备心血管知识图谱的骨架怎么搭2.1 先定实体和关系别急着写代码知识图谱构建翻车最常见的原因是一上来就打开 Neo4j 建节点建到一半发现「并发症」和「并发症」之间还有层级只能推倒重来。我一般先做本体建模把心血管领域拆成几类实体和几类关系落到一张表里再动手。实体类型示例说明疾病冠心病、房颤、心力衰竭图谱核心节点症状胸痛、心悸、呼吸困难指向疾病药品阿司匹林、华法林含禁忌与相互作用检查心电图、心脏彩超用于确诊人群/禁忌孕妇、肝功能不全用药约束条件关系类型控制在 6 到 8 种就够用has_symptom、treated_by、contraindicated_with、diagnosed_by、complication_of、belongs_to。关系类型不是越多越好每多一种后面问答模板就要多写一套匹配逻辑。本体建模这一步的产出应该是一份实体-关系清单而不是代码。2.2 数据清洗把非结构化文本变成三元组数据集里通常是疾病百科段落、药品说明书片段这类半结构化文本。我的做法是先按标点切句再用规则 词典匹配抽三元组而不是一上来就上大模型。原因是心血管领域实体名相对固定词典匹配的准确率足够高且可解释。import re # 实体词典实际项目里从数据集或本体文件加载 disease_dict [房颤, 冠心病, 心力衰竭, 高血压] drug_dict [华法林, 阿司匹林, 美托洛尔] symptom_dict [胸痛, 心悸, 呼吸困难] # 关系触发词 - 关系名 relation_patterns { 表现为: has_symptom, 症状包括: has_symptom, 常用: treated_by, 治疗药物: treated_by, 禁用: contraindicated_with, } def extract_triples(sentence): triples [] for trigger, rel in relation_patterns.items(): if trigger not in sentence: continue left, _, right sentence.partition(trigger) head next((d for d in disease_dict if d in left), None) tail next((d for d in drug_dict symptom_dict if d in right), None) if head and tail: triples.append((head, rel, tail)) return triples text 房颤表现为心悸常用华法林治疗肝功能不全者禁用华法林。 for s in re.split(r[。], text): print(extract_triples(s))这段代码的逻辑是按标点切句后用关系触发词定位主语和宾语再从词典里回捞标准实体名。参数上relation_patterns是唯一需要按数据集调整的地方触发词覆盖不全就会漏三元组。注意partition只切第一个触发词一句话里出现两个关系时要先分句否则会丢关系。抽取完建议人工抽查 50 条统计准确率低于 85% 就回去补触发词别急着入库。2.3 三元组去重与格式统一抽出来的三元组会有重复和别名问题比如「心衰」和「心力衰竭」被当成两个实体。入库前必须做一次归一化建一张别名词表把同义实体映射到标准名再用集合去重。alias_map {心衰: 心力衰竭, 房颤: 心房颤动} def normalize(triples): seen set() result [] for h, r, t in triples: h, t alias_map.get(h, h), alias_map.get(t, t) key (h, r, t) if key not in seen: seen.add(key) result.append(key) return result归一化是知识图谱质量的分水岭。别名不统一后面问答时用户问「心衰」就查不到「心力衰竭」的边。这一步没有捷径只能靠领域词表加人工核对但一次投入长期受益。3. 用 Neo4j 构建知识图谱从 CSV 到可查询的图3.1 环境准备与 Neo4j 启动图谱存储我选 Neo4j原因是它的 Cypher 查询语言对多跳关系友好问答系统里「疾病的并发症用什么药」这种两跳问题写起来很直观。本地跑通最省事的方式是用 Docker避免装 JDK 和配环境的玄学问题。docker run -d \ --name cardio-kg \ -p 7474:7474 -p 7687:7687 \ -e NEO4J_AUTHneo4j/cardio123 \ neo4j:5参数说明7474是浏览器控制台端口7687是 Python 驱动连接的 Bolt 端口NEO4J_AUTH设初始账号密码首次登录会要求改密。启动后访问http://localhost:7474能看到控制台就说明成了。注意容器删了数据就没了正式用要挂载数据卷把/data映射到宿主机。3.2 批量导入三元组数据量小的时候可以直接用 Python 驱动逐条写但几千条以上就该走LOAD CSV速度快一个量级。先把三元组导出成 CSV表头固定为head,relation,tail。import csv from neo4j import GraphDatabase triples [ (心房颤动, has_symptom, 心悸), (心房颤动, treated_by, 华法林), (华法林, contraindicated_with, 肝功能不全), ] with open(triples.csv, w, newline, encodingutf-8) as f: writer csv.writer(f) writer.writerow([head, relation, tail]) writer.writerows(triples) driver GraphDatabase.driver(bolt://localhost:7687, auth(neo4j, cardio123)) with driver.session() as session: session.run( LOAD CSV WITH HEADERS FROM file:///triples.csv AS row MERGE (h:Entity {name: row.head}) MERGE (t:Entity {name: row.tail}) MERGE (h)-[:REL {type: row.relation}]-(t) ) driver.close()这里用统一的Entity标签加REL关系类型是为了导入简单查询时靠type属性区分关系。代价是 Cypher 写起来稍啰嗦但胜在导入脚本通用。LOAD CSV的路径是 Neo4j 容器内的/var/lib/neo4j/import宿主机文件要先拷进去否则会报找不到文件。导入后跑一句MATCH (n) RETURN count(n)确认节点数和 CSV 里的去重实体数对得上才算成功。3.3 验证图谱连通性导入完别急着写问答先验证图谱是不是「连得起来」。孤立节点太多说明关系抽取漏了。// 统计孤立节点 MATCH (n:Entity) WHERE NOT (n)--() RETURN count(n) AS isolated; // 看某个疾病的两跳邻居 MATCH (d:Entity {name: 心房颤动})-[r1]-(m)-[r2]-(x) RETURN d.name, type(r1), m.name, type(r2), x.name LIMIT 20;孤立节点占比超过 20% 就要回头查抽取规则。两跳查询能跑出结果说明图谱具备支撑多跳问答的基础这一步是后面问答能不能答对的前提。4. 问答系统实现从自然语言到 Cypher 的映射4.1 意图识别与实体链接问答系统的核心是把用户问题翻译成图谱查询。我的做法分两步先识别问题意图问症状、问用药、问禁忌再抽取问题里的实体并链接到图谱标准名。intent_rules { 症状: has_symptom, 表现: has_symptom, 吃什么药: treated_by, 用什么药: treated_by, 禁用: contraindicated_with, 禁忌: contraindicated_with, } def parse_question(question): intent None for kw, rel in intent_rules.items(): if kw in question: intent rel break entity next((e for e in disease_dict drug_dict if e in question), None) return entity, intent参数上intent_rules的关键词要覆盖用户口语表达比如「吃啥药」「能不能用」都得映射到treated_by或contraindicated_with。实体链接这里用了最简单的包含匹配实际项目里要接别名表否则用户说「心衰」就匹配不上。意图或实体任一为空就直接返回兜底话术别硬查。4.2 模板化生成 Cypher意图和实体确定后套模板生成查询语句。模板数量等于关系类型数量维护成本可控。cypher_templates { has_symptom: MATCH (d:Entity {name: $name})-[r:REL {type: has_symptom}]-(s) RETURN s.name AS answer , treated_by: MATCH (d:Entity {name: $name})-[r:REL {type: treated_by}]-(m) RETURN m.name AS answer , contraindicated_with: MATCH (d:Entity {name: $name})-[r:REL {type: contraindicated_with}]-(c) RETURN c.name AS answer , } def answer(question): entity, intent parse_question(question) if not entity or not intent: return 抱歉暂时无法理解这个问题。 with driver.session() as session: result session.run(cypher_templates[intent], nameentity) answers [record[answer] for record in result] return 、.join(answers) if answers else 图谱中没有相关记录。模板里用参数$name而不是字符串拼接一是防注入二是 Neo4j 能缓存执行计划。返回多条时用顿号连接符合中文阅读习惯。注意contraindicated_with的方向如果图谱里是「药品→禁忌」那问「房颤禁用什么」就要先查房颤的用药再查禁忌属于两跳模板要相应改成两段MATCH。4.3 多跳问题的处理单跳模板答不了「房颤的并发症用什么药」这类问题需要把两个模板串起来。def answer_multi_hop(question): entity, _ parse_question(question) if not entity: return 抱歉暂时无法理解这个问题。 with driver.session() as session: result session.run( MATCH (d:Entity {name: $name})-[:REL {type: complication_of}]-(c) -[:REL {type: treated_by}]-(m) RETURN c.name AS complication, m.name AS drug , nameentity) rows [f{r[complication]}可用{r[drug]} for r in result] return .join(rows) if rows else 图谱中没有相关记录。多跳查询的性能取决于图谱规模几万节点内响应都在毫秒级。坑在于关系方向必须和导入时一致方向写反会返回空结果排查时先用 Neo4j 控制台手写 Cypher 验证路径存在再回来看代码。5. 避坑与排查这套系统最容易翻车的五个地方5.1 实体别名没归一问答查不到现象用户问「心衰用什么药」系统返回「图谱中没有相关记录」但图谱里明明有「心力衰竭」的用药关系。原因是实体链接只做了包含匹配没接别名表。解决在parse_question前加一层别名映射把用户输入先转成标准名再匹配别名表从数据清洗阶段就维护起来。5.2 关系方向写反多跳查询全空现象单跳问答正常一问「并发症用什么药」就空。原因是 Cypher 里complication_of的方向和导入时相反。解决导入前把关系方向固定成「主体→客体」并写进文档查询模板严格照抄排查时先在 Neo4j 控制台跑MATCH (a)-[r:REL {type:complication_of}]-(b) RETURN a,b LIMIT 5看方向。5.3 LOAD CSV 报找不到文件现象导入脚本报Couldnt load the external resource。原因是LOAD CSV读的是 Neo4j 服务端路径不是宿主机路径。解决把 CSV 拷进容器的 import 目录或用docker cp复制路径写file:///triples.csv对应 import 根目录。5.4 意图关键词覆盖不全口语问法答不上现象用户问「房颤吃啥药」没反应问「房颤用什么药」正常。原因是intent_rules里没有「吃啥药」。解决把常见口语变体都加进规则或者引入同义词扩展上线前收集 100 条真实问法做回归测试命中率低于 90% 就继续补。5.5 图谱规模上来后查询变慢现象节点过万后多跳查询从毫秒变成秒级。原因是没建索引Neo4j 全表扫描。解决给实体名建索引CREATE INDEX FOR (n:Entity) ON (n.name)关系属性type也可以考虑建索引另外避免无方向的MATCH (a)--(b)方向明确才能走索引。6. 把问答准确率从能用推到好用两个具体技巧第一个技巧是给答案加置信度和来源。图谱问答最怕答错还理直气壮我的做法是每条答案附带它来自哪条边用户能顺着边自己核对。实现上就是在 Cypher 里多RETURN一个关系属性比如药品的说明书来源字段前端展示时折叠起来。这样即使答错用户也知道错在哪一环而不是对整个系统失去信任。第二个技巧是用测试集量化效果而不是凭感觉说「差不多能用」。我一般从数据集里抽 100 个问题人工标注标准答案跑一遍算准确率再按意图分类统计哪类问题最差。下面这个脚本就是干这个的。test_cases [ (房颤有什么症状, 心悸), (房颤用什么药, 华法林), (华法林禁用于什么, 肝功能不全), ] def evaluate(cases): correct 0 for q, gold in cases: pred answer(q) if gold in pred: correct 1 else: print(f错例: {q} | 期望: {gold} | 实际: {pred}) print(f准确率: {correct / len(cases):.2%}) evaluate(test_cases)参数上gold用包含匹配而不是全等是因为答案可能返回多个实体。错例打印出来要逐条看是实体没链上还是意图判错分类统计后针对性补规则。我自己的习惯是每加一批数据就重跑一次这个脚本准确率掉了就说明新数据引入了脏三元组先回查清洗环节。这套流程跑顺之后图谱问答的准确率能稳定在可交付的水平剩下的就是持续补数据、补别名、补口语问法。希望帮到你。本文还有配套的精品资源点击获取