基于Python与Neo4j构建心血管知识图谱问答系统实战

简介：这套Python项目围绕心血管疾病构建基于知识图谱的问答系统，提供完整项目源码、说明文档与数据集，面向计算机、人工智能、电子信息等专业的在校学生和老师，可用于毕设、课设、项目初期立项演示，也适合想入门知识图谱问答的开发者自学。压缩包共197个文件、约5.52MB：json与csv文件分别存放图谱结构数据和心血管疾病相关知识语料/实体关系表，py脚本是系统核心实现，txt、md为运行说明与文档，png、gif可直观展示界面或流程。目前已有58人浏览学习。下载并阅读README后即可运行项目，还能基于源码修改疾病类型、扩展问答逻辑或丰富前端交互。整体代码已验证可运行，能帮助读者理解知识图谱构建、实体识别与问答检索的完整流程，是一份实战性较强的参考资料。

1. Python 知识图谱问答系统：心血管疾病场景先解决什么问题

「高血压不能吃什么」「心绞痛要做哪些检查」这类问题交给程序回答，最稳的做法不是堆大模型接口，而是把医学知识组织成图谱，再让问答逻辑在图上检索。基于知识图谱的心血管疾病问答系统，是一条「知识结构化→问句解析→图谱查询→答案生成」的流水线，几百到几万条三元组都能跑，是 Python 里同时涉及 NLP、图数据库与 Web 服务的典型组合。

核心价值是可控：知识来自可审计数据集，答案由 Cypher 查询产出，出错能沿着图谱一步步回查。相比基于 DeepSeek 等大模型的问答系统，图谱方案没有推理成本、可离线演示，答案还能直接指向数据来源，适合院内随访辅助、健康宣教演示和课程设计这类要求可解释的场景。

适合想从零搭一个可演示问答系统的 Python 开发者，也适合想搞懂知识图谱构建完整链路的学生。下文按建图、解析、查询、验证四步展开，每一节都有可以直接抄下来改的代码和命令。

2. 心血管知识图谱构建：本体设计、数据集清洗与 Neo4j 批量导入

知识图谱不是「先把数据全塞进图数据库再说」，而是先定清楚有哪些节点、哪些关系。这一步叫本体设计，做错了后面所有查询模板都得跟着返工。

2.1 本体设计先行：六类实体与七种关系，决定后面所有查询方向

常见做法是参考公开医学知识图谱项目的 schema，把心血管场景收敛成六类实体。不要一开始就把「基因」「手术方式」都加进来，实体类型越多，规则问答的维护成本涨得越快。

实体定完后，关系要统一方向。我一般规定「主语在前、宾语在后」，比如疾病指向症状、疾病指向药品，这样后面写查询模板时不用每句都猜方向。

关系数量控制在七条左右，正好对应下一章要做的七类意图。关系超过十五条时，规则模板会开始互相打架，那时候就不是加规则能解决的问题了。

2.2 数据集清洗：宽表转三元组，顿号和空值一起处理

下载下来的数据集大多是宽表，一行一个疾病，列是 symptom、drug、food 这些，一个格子里塞了多个值，用顿号或逗号分隔。写库之前必须先把宽表转成三元组列表，否则导入脚本会写出一堆脏边。

import pandas as pd raw = pd.read_csv("data/relation.csv", encoding="utf-8") raw = raw.dropna(subset=["disease"]) # 丢掉疾病名为空的行 raw["disease"] = raw["disease"].astype(str).str.strip() triples = [] for _, row in raw.iterrows(): d = row["disease"] # col 是宽表列名，rel 是对应关系名 for rel, col in [ ("HAS_SYMPTOM", "symptom"), ("RECOMMEND_DRUG", "drug"), ("NO_EAT_FOOD", "no_eat"), ("NEED_CHECK", "check"), ("BELONG_TO_DEPARTMENT", "department"), ]: val = row.get(col) if pd.notna(val): for item in str(val).split("、"): # 单元格内多个值用顿号分隔 item = item.strip() if item: triples.append((d, rel, item)) pd.DataFrame(triples, columns=["head", "rel", "tail"]).to_csv( "data/triples.csv", index=False, encoding="utf-8" ) print(len(triples), triples[:5])

这里有三个参数细节值得注意。dropna(subset=["disease"])只过滤主键为空的行，其他列的空值交给循环里的pd.notna处理，避免把「不知道是什么」误写成空字符串边。split("、")是清洗的关键，多数医学数据集的列表分隔符是中文顿号，只按逗号拆会漏掉一半数据。输出成 triples.csv 后，后续导入脚本和实体词典生成都从这份文件读，保证单一数据源。

2.3 用 Python 驱动批量写 Neo4j：约束、MERGE 与批次事务

写库前先确认环境：Neo4j 4.4 或 5.x 社区版，本地装好 Python 3.8 以上，pip install neo4j。连接串默认是bolt://localhost:7687，认证账号密码在 Neo4j 启动时设置。

from neo4j import GraphDatabase URI = "bolt://localhost:7687" AUTH = ("neo4j", "change_me") driver = GraphDatabase.driver(URI, auth=AUTH) NODE_LABELS = ["Disease", "Symptom", "Drug", "Food", "Check", "Department"] def ensure_schema(session): for label in NODE_LABELS: session.run( f"CREATE CONSTRAINT IF NOT EXISTS " f"FOR (n:{label}) REQUIRE n.name IS UNIQUE" ) def upsert_node(tx, label, name): tx.run(f"MERGE (n:{label} {{name:$name}})", name=name) def upsert_rel(tx, head, head_label, rel, tail, tail_label): # label 和 rel 来自代码内的白名单，name 才来自数据，必须参数化 cypher = ( f"MATCH (h:{head_label} {{name:$head}}), " f"(t:{tail_label} {{name:$tail}}) " f"MERGE (h)-[r:{rel}]->(t)" ) tx.run(cypher, head=head, tail=tail) def import_triples(session, triples, batch_size=500): for i in range(0, len(triples), batch_size): with session.begin_transaction() as tx: for head, rel, tail in triples[i:i + batch_size]: hl, tl = rel.split("_")[0], "_".join(rel.split("_")[1:]) # 由关系名推导类型，需自行校准 upsert_node(tx, hl, head) upsert_node(tx, tl, tail) upsert_rel(tx, head, hl, rel, tail, tl)

这段代码有三个设计点。第一，CREATE CONSTRAINT ... REQUIRE n.name IS UNIQUE给每个实体类型建唯一约束，重复执行不会报错，后续 MERGE 才能保证不产生重复节点。第二，MERGE 比 CREATE 慢一点，但幂等，断点重跑不会把图写花，数据量在十万条以内完全可接受。第三，事务按 500 条一批提交，避免单事务太大导致 Neo4j 内存压力；超过十万条三元组建议改用neo4j-admin database import离线导入，那套方式不走驱动，速度能快一个数量级。

需要注意标签和关系名是用 f-string 拼进去的，安全前提是它们来自代码里的白名单，绝不能把用户输入拼到这个位置。实体名称这类数据值一律走$name、$head参数绑定，这既是防注入，也是避免名称里的引号把语句弄坏。

2.4 导入后的验证查询与三个高频坑

导入完先跑两条查询确认图结构。一条看每个类型的节点数，一条看核心关系的分布：

MATCH (n) RETURN labels(n)[0] AS label, count(*) AS cnt ORDER BY cnt DESC; MATCH (d:Disease)-[r:HAS_SYMPTOM]->(s:Symptom) RETURN d.name AS disease, count(r) AS cnt, collect(s.name)[..5] AS sample ORDER BY cnt DESC LIMIT 5;

collect(s.name)[..5]是 Cypher 的列表切片语法，只看每个疾病前五个症状，确认关系不是全挤在同一个节点上。

三个坑里最隐蔽的是「知识图谱只显示 25 个标签」这个现象。Neo4j Browser 左侧的 Label 列表在实体类型较多或标签数量大时有显示上限，看起来像图谱没建全，其实数据都在。验证别靠浏览器侧栏，要用CALL db.labels();看全量标签。

第二个坑是关系方向不一致。清洗阶段有人把「药品→疾病」也当成正向写进去，查询(d:Disease)-[:RECOMMEND_DRUG]->(dr:Drug)就会漏数据。排查时用无方向查询MATCH (a)-[r:HAS_SYMPTOM]-(b)看看到底是没导入还是方向反了，根治手段是在 2.2 的清洗脚本里统一方向。

第三个坑是属性类型不统一。「是否遗传」这类列在 CSV 里是「是/否/未知」字符串，写入时如果不统一转成布尔或固定字符串，后面查询里WHERE n.hereditary = true永远查不到结果。凡是有枚举语义的列，写库前先value_counts()看一遍取值集合。

3. 问句解析与意图识别：词典实体识别 + 规则模板的落地写法

问答系统能不能用，一半看知识图谱，另一半看问句解析。医疗问句的句式其实非常收敛，把意图和实体拆出来，规则方案足够撑起一个能演示的系统。

3.1 先定七个意图标签，规则方案才能收敛

对真实患者问题做统计，头部的七类问法占了八成以上。意图标签要和上一章的关系一一对应，否则意图识别出来了，查询模板却不知道查什么。

意图控制在十个以内时，正则和关键词模板是最划算的，命中率能到九成，调试也直观。超过十个意图，规则之间会互相覆盖，那时候再考虑 fastText 或 BERT 分类器，但那是另一个量级的工程了。

3.2 实体识别用词典最长匹配，而不是 Jieba 裸跑

医疗词表里「心绞痛」「心肌梗死」这类词，Jieba 默认词典经常切碎。给 Jieba 加自定义词典能缓解，但词频权重偶尔还会把它切开。更稳的做法是用 flashtext 这类纯词典匹配库，做最长匹配，顺便把别名统一成标准名。

from flashtext import KeywordProcessor def build_entity_index(dict_path): kp = KeywordProcessor(case_sensitive=False) with open(dict_path, encoding="utf-8") as f: for line in f: line = line.rstrip("\n") if not line or "\t" not in line: continue name, aliases = line.split("\t", 1) for alias in [a.strip() for a in aliases.split(",") if a.strip()]: kp.add_keyword(alias, name) # 别名 -> 标准名 kp.add_keyword(name, name) # 标准名自身也要能命中 return kp # entity_dict.tsv 每行: 标准名<TAB>别名1,别名2 kp = build_entity_index("data/entity_dict.tsv") text = "高血压患者能吃柚子吗？" print(kp.extract_keywords(text)) # ['高血压', '柚子']

add_keyword(alias, name)把别名映射到标准名，抽取结果天然完成实体归一化，比如「原发性高血压」这个别名会统一变成「高血压」。实体词典最好直接从第 2 章的图谱导出，保证识别和查询共用同一份词表：

MATCH (n) WHERE n:Disease OR n:Drug OR n:Food RETURN n.name AS name, coalesce(n.alias, '') AS aliases

coalesce处理没有别名的节点，避免导出文件里出现空字段把解析脚本搞挂。

3.3 意图识别用正则模板，命中顺序就是优先级

意图识别用正则列表实现，顺序就是优先级。规则引擎最重要的一条经验：否定意图永远排在肯定意图前面。

import re RULES = [ (r"(什么|哪些|啥).{0,8}(症状|表现|感觉)", "SYMPTOM_OF_DISEASE"), (r".{0,4}(症状|表现).{0,8}(什么|哪些|啥).{0,4}病", "DISEASE_OF_SYMPTOM"), (r"(什么|哪些|啥).{0,8}(药|药物|药品|药片)", "DRUG_OF_DISEASE"), (r"(不能|不可|不宜|别|忌).{0,6}(吃|食用|喝|摄入)", "FOOD_FORBIDDEN"), (r"(能|可以|宜|适合|应该).{0,6}(吃|食用|喝|摄入)", "FOOD_RECOMMEND"), (r"(什么|哪些|啥).{0,8}(检查|检验|项目)", "CHECK_OF_DISEASE"), (r"(挂|去|看).{0,4}(什么|哪个|啥).{0,3}(科|科室)", "DEPARTMENT_OF_DISEASE"), ] def match_intent(question: str) -> str: for pattern, intent in RULES: if re.search(pattern, question): return intent return "DEFAULT"

正则里的.{0,8}是容错设计，允许「高血压平时都有哪些症状」这类插入语气词的问法命中。FOOD_FORBIDDEN必须在FOOD_RECOMMEND之前，因为「不能吃」里也含「能」字。疾病反向问句「什么病会胸痛」的 pattern 同时覆盖了句首和句中两种位置，这类边界在写规则时要逐个用例验证。

3.4 否定词、多实体与口语噪音：三个实测高频问题

第一个问题是「高血压不能吃柚子吗」这种句子，「能」和「不能」同时出现，正则按顺序会先匹配到 FOOD_RECOMMEND。解决方式是先做否定词扫描，命中「不能、不可以、不宜、忌」时强制走 FOOD_FORBIDDEN：

NEG_WORDS = ("不能", "不可以", "不宜", "禁忌", "忌") def refine_intent(question: str, intent: str) -> str: if any(w in question for w in NEG_WORDS): return "FOOD_FORBIDDEN" return intent

第二个问题是多实体。「心梗和脑梗哪个更严重」会同时抽出两个疾病，规则系统处理不了对比逻辑。我一般的做法是检测到多个疾病实体时，先返回对比模板的提示语，或者只取第一个实体并追加「您是否想问...」，不要让下游查询静默丢掉实体。

第三个问题是口语噪音。「我爸爸得了高血压平时要注意什么」里的「爸爸」「得了」会污染后面的相似度兜底。在建词典阶段维护一份医疗停用词表，实体抽取前先做一轮过滤，能让兜底检索的准确率明显上升。

4. 答案检索与生成：Cypher 模板路由、多跳查询与兜底检索

意图和实体都拿到了，接下来就是把它们翻译成 Cypher 查询，拿到结果再组装成自然语言答案。这一章是整个系统的出口，翻车概率也最高。

4.1 意图到 Cypher 模板的映射表

每个意图对应一条固定模板，查询语句里只有实体值是变量，结构完全由后端白名单控制。

注意反向关系的写法。DISEASE_OF_SYMPTOM用的是(s:Symptom {name:$name})<-[:HAS_SYMPTOM]-(d:Disease)，箭头方向是查询语义的一部分，写反了就查不到任何数据。

4.2 模板引擎实现：参数绑定与结果组装

TEMPLATES = { "SYMPTOM_OF_DISEASE": ( "MATCH (d:Disease {name:$name})-[:HAS_SYMPTOM]->(s:Symptom) " "RETURN s.name AS answer LIMIT 10" ), # 其余意图按 4.1 的表补齐 } def query_answer(driver, intent, entity): cypher = TEMPLATES.get(intent) if not cypher: return [] try: with driver.session() as session: records = session.run(cypher, name=entity).data() except Exception: return [] answers = [r["answer"] for r in records] # dict.fromkeys 去重且保持顺序 return list(dict.fromkeys(answers))

session.run(cypher, name=entity)里的name=entity就是参数绑定，值是第 3 章归一化后的标准实体名。LIMIT 10是答案长度上限，防止「高血压」这种中心节点挂了几百条症状把接口拖慢。try/except包住查询是有意的：图谱数据不完整时查询可能抛异常，这时候返回空列表走兜底，比把异常抛给前端友好得多。

4.3 多跳查询：禁忌判断与跨实体过滤

单跳查询只能回答「有什么症状」「吃什么药」这类简单问题。真实问法里有一类必须多跳，比如「房颤患者能不能吃阿司匹林」，要同时查推荐关系和禁忌关系：

MATCH (d:Disease {name:$name})-[:RECOMMEND_DRUG]->(dr:Drug {name:$drug}) RETURN '可以' AS answer UNION MATCH (d:Disease {name:$name})-[:FORBID_DRUG]->(dr:Drug {name:$drug}) RETURN '不可以' AS answer

UNION把两条独立查询的结果合并，应用层根据返回的字符串直接出答案，不用写两段查询逻辑再拼结果。这类模板在实现时要注意：RECOMMEND_DRUG和FORBID_DRUG在同一条路径上同时存在时，UNION会返回两行，需要在组装层做「禁忌优先」的排序规则。

另一个常用多跳是并发症查询。第 2 章如果加了COMPLICATION关系，一条查询就能覆盖「高血压会引起哪些并发症」：

MATCH (d:Disease {name:$name})-[:COMPLICATION]->(c:Disease) RETURN c.name AS answer

4.4 兜底检索：图谱查不到时的 Jaccard 相似度

实体没抽出来、模板查不到、意图落进 DEFAULT，这三种情况都要走兜底。最常见的兜底是基于分词后集合的 Jaccard 相似度：

import jieba def _token_set(text: str): return set(jieba.lcut(text.lower())) def jaccard(a: str, b: str) -> float: A, B = _token_set(a), _token_set(b) if not A or not B: return 0.0 return len(A & B) / len(A | B) def fallback_answer(question: str, candidates, topk=3, threshold=0.1): scored = [(c, jaccard(question, c)) for c in candidates] scored.sort(key=lambda x: x[1], reverse=True) return [c for c, s in scored[:topk] if s >= threshold]

candidates是所有疾病名和别名的集合，可以用第 3 章导出的实体词典直接生成。阈值threshold是最需要调的参数：0.1 偏低，会把「高血压怎么预防」错误拉回「高血压」然后给一段不太相关的答案；0.3 偏高，正常问法也进不了兜底。我一般准备两百条测试问句，在 0.05 到 0.3 之间扫一遍，选准确率拐点处的值。

兜底也要兜不住的时候。阈值过滤后仍然为空，答案层必须返回「暂时无法回答，建议咨询心内科医生」这类话术，医疗问答系统不能为了答而答。

4.5 答案组织的三个细节

原始查询结果是字符串列表，直接返回前端很难看，组装层要按意图加前缀、做截断：

PREFIX = { "SYMPTOM_OF_DISEASE": "常见症状包括", "DRUG_OF_DISEASE": "临床常用药物有", "FOOD_FORBIDDEN": "饮食上需要注意，忌口", "CHECK_OF_DISEASE": "需要做的检查有", } def format_answer(intent: str, items) -> str: if not items: return "暂时无法回答，建议咨询心内科医生" body = "、".join(items[:10]) if len(items) > 10: body += f" 等 {len(items)} 项" return f"{PREFIX.get(intent, '相关知识')}：{body}"

三个细节分别是：前缀文案要让答案读起来像完整句子；超过十条截断并标注总数，避免同一疾病的几百条症状把响应撑爆；空结果文案固定成一句就医建议，这是医疗问答的底线。

5. 源码目录、数据集对齐与回归验证：把问答准确率变成可量化的指标

下载下来的工程包能不能跑起来，先看的不是代码，而是目录和数据的对应关系。

5.1 拿到源码包先对齐三张表，再谈跑通

一个标准的最小工程目录长这样：

cardiovascular_kg_qa/ ├── data/ │ ├── disease.csv # 疾病主表：名称、别名、病因、预防等属性 │ ├── relation.csv # 宽表：disease 与 symptom/drug/food/check 对应 │ └── entity_dict.tsv # 实体词典：标准名<TAB>别名1,别名2 ├── kg/ │ ├── build_graph.py # 第 2 章的写库脚本 │ └── query.py # 第 4 章的 Cypher 模板与查询封装 ├── nlp/ │ ├── entity.py # flashtext 实体识别 │ └── intent.py # 正则意图规则 ├── server/ │ └── app.py # Flask 接口，POST /qa 返回答案 ├── tests/ │ └── qa_cases.json # 回归用例 └── requirements.txt

这类工程八成跑不通的根源只有一个：build_graph.py里的关系名和relation.csv的列名对不上。拿到数据后先用三行代码摸底，不要急着启动 Neo4j：

import pandas as pd rel = pd.read_csv("data/relation.csv", encoding="utf-8") print(rel.columns.tolist()) print(rel["disease"].nunique(), len(rel))

列名列表和节点数一出来，就能判断数据规模够不够支撑演示。数据集下载后如果编码是 GBK，encoding="utf-8"会直接报错，改成"gbk"重读即可。

5.2 回归测试：让问答准确率不再靠肉眼

问答系统改规则特别容易改 A 坏 B，所以测试用例要提前备好。qa_cases.json里每一条同时卡意图、实体和答案三段：

[ { "question": "高血压有什么症状", "intent": "SYMPTOM_OF_DISEASE", "entity": "高血压", "must_contain": ["头晕"] }, { "question": "冠心病吃什么药", "intent": "DRUG_OF_DISEASE", "entity": "冠心病", "must_contain": ["阿司匹林"] } ]

配套的 pytest 脚本按三个断言逐层检查：

import json import pytest from server.app import answer CASES = json.load(open("tests/qa_cases.json", encoding="utf-8")) @pytest.mark.parametrize("case", CASES, ids=[c["question"] for c in CASES]) def test_qa(case): resp = answer(case["question"]) assert resp["intent"] == case["intent"], "意图识别错误" assert case["entity"] in resp["entities"], "实体抽取遗漏" for word in case["must_contain"]: assert word in resp["answer"], f"答案缺少关键词: {word}"

三个断言分别卡住链路的三段：意图错了说明正则顺序有问题，实体漏了说明词典覆盖不足，答案缺词说明图谱数据或查询模板有缺口。每修一个 badcase，先往qa_cases.json里补一条，再改代码，避免无意识回归。

5.3 三个能明显提分的具体技巧

第一个技巧是别名归一化和实体词典必须同源。实体识别用的entity_dict.tsv如果和图谱里的节点不是同一份数据，会出现「抽取到了别名，图谱里却按标准名查不到」的尴尬情况。我一般把导出脚本直接写进构建流程，建完图自动重新生成词典，两个文件永远一起变更。

第二个技巧是兜底阈值不要拍脑袋。准备三百条问句，把阈值从 0.05 到 0.3 按 0.05 步进各跑一遍，画出 top1 命中率曲线，取曲线拐点。这个数字每个数据集都不一样，抄别人的参数往往不适配自己的数据分布。

第三个技巧是评测指标别只报 top1。top1 不中但 top3 命中，说明兜底在起作用，只是排序差一步；只看 top1 会把这类 progress 判成零分。用 MRR（平均倒数排名）能更细地看到每次修改带来的真实涨跌，改动规则时对比 MRR 比对比单条用例更可靠。

把这套回归脚本固定下来，换数据集、加实体类型时先跑一遍，准确率变化立刻变成可对比的数字，比凭感觉调规则靠谱得多。

本文还有配套的精品资源，点击获取