简介这套Python项目围绕心血管疾病构建基于知识图谱的问答系统提供完整项目源码、说明文档与数据集面向计算机、人工智能、电子信息等专业的在校学生和老师可用于毕设、课设、项目初期立项演示也适合想入门知识图谱问答的开发者自学。压缩包共197个文件、约5.52MBjson与csv文件分别存放图谱结构数据和心血管疾病相关知识语料/实体关系表py脚本是系统核心实现txt、md为运行说明与文档png、gif可直观展示界面或流程。目前已有58人浏览学习。下载并阅读README后即可运行项目还能基于源码修改疾病类型、扩展问答逻辑或丰富前端交互。整体代码已验证可运行能帮助读者理解知识图谱构建、实体识别与问答检索的完整流程是一份实战性较强的参考资料。1. Python 知识图谱问答系统心血管疾病场景先解决什么问题「高血压不能吃什么」「心绞痛要做哪些检查」这类问题交给程序回答最稳的做法不是堆大模型接口而是把医学知识组织成图谱再让问答逻辑在图上检索。基于知识图谱的心血管疾病问答系统是一条「知识结构化→问句解析→图谱查询→答案生成」的流水线几百到几万条三元组都能跑是 Python 里同时涉及 NLP、图数据库与 Web 服务的典型组合。核心价值是可控知识来自可审计数据集答案由 Cypher 查询产出出错能沿着图谱一步步回查。相比基于 DeepSeek 等大模型的问答系统图谱方案没有推理成本、可离线演示答案还能直接指向数据来源适合院内随访辅助、健康宣教演示和课程设计这类要求可解释的场景。适合想从零搭一个可演示问答系统的 Python 开发者也适合想搞懂知识图谱构建完整链路的学生。下文按建图、解析、查询、验证四步展开每一节都有可以直接抄下来改的代码和命令。2. 心血管知识图谱构建本体设计、数据集清洗与 Neo4j 批量导入知识图谱不是「先把数据全塞进图数据库再说」而是先定清楚有哪些节点、哪些关系。这一步叫本体设计做错了后面所有查询模板都得跟着返工。2.1 本体设计先行六类实体与七种关系决定后面所有查询方向常见做法是参考公开医学知识图谱项目的 schema把心血管场景收敛成六类实体。不要一开始就把「基因」「手术方式」都加进来实体类型越多规则问答的维护成本涨得越快。实体类型实例说明Disease 疾病高血压、冠心病、心房颤动全图的核心节点Symptom 症状胸痛、心悸、气短患者主诉入口Drug 药品阿司匹林、硝苯地平用药推荐与禁忌Food 食物柚子、海带、咸菜饮食指导Check 检查心电图、冠脉造影辅助检查项目Department 科室心内科、急诊科就诊导诊实体定完后关系要统一方向。我一般规定「主语在前、宾语在后」比如疾病指向症状、疾病指向药品这样后面写查询模板时不用每句都猜方向。关系名方向语义HAS_SYMPTOMDisease → Symptom疾病有哪些症状RECOMMEND_DRUGDisease → Drug推荐用药FORBID_DRUGDisease → Drug禁忌用药RECOMMEND_FOODDisease → Food宜吃食物NO_EAT_FOODDisease → Food忌口食物NEED_CHECKDisease → Check需要做的检查BELONG_TO_DEPARTMENTDisease → Department就诊科室关系数量控制在七条左右正好对应下一章要做的七类意图。关系超过十五条时规则模板会开始互相打架那时候就不是加规则能解决的问题了。2.2 数据集清洗宽表转三元组顿号和空值一起处理下载下来的数据集大多是宽表一行一个疾病列是 symptom、drug、food 这些一个格子里塞了多个值用顿号或逗号分隔。写库之前必须先把宽表转成三元组列表否则导入脚本会写出一堆脏边。import pandas as pd raw pd.read_csv(data/relation.csv, encodingutf-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, indexFalse, encodingutf-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, authAUTH) NODE_LABELS [Disease, Symptom, Drug, Food, Check, Department] def ensure_schema(session): for label in NODE_LABELS: session.run( fCREATE CONSTRAINT IF NOT EXISTS fFOR (n:{label}) REQUIRE n.name IS UNIQUE ) def upsert_node(tx, label, name): tx.run(fMERGE (n:{label} {{name:$name}}), namename) def upsert_rel(tx, head, head_label, rel, tail, tail_label): # label 和 rel 来自代码内的白名单name 才来自数据必须参数化 cypher ( fMATCH (h:{head_label} {{name:$head}}), f(t:{tail_label} {{name:$tail}}) fMERGE (h)-[r:{rel}]-(t) ) tx.run(cypher, headhead, tailtail) def import_triples(session, triples, batch_size500): 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 先定七个意图标签规则方案才能收敛对真实患者问题做统计头部的七类问法占了八成以上。意图标签要和上一章的关系一一对应否则意图识别出来了查询模板却不知道查什么。意图问法示例对应图谱关系SYMPTOM_OF_DISEASE高血压有什么症状HAS_SYMPTOMDISEASE_OF_SYMPTOM胸痛可能是什么病HAS_SYMPTOM 反向DRUG_OF_DISEASE冠心病吃什么药RECOMMEND_DRUGFOOD_RECOMMEND高血压吃什么好RECOMMEND_FOODFOOD_FORBIDDEN高血压不能吃什么NO_EAT_FOODCHECK_OF_DISEASE心绞痛要做哪些检查NEED_CHECKDEPARTMENT_OF_DISEASE房颤挂哪个科BELONG_TO_DEPARTMENT意图控制在十个以内时正则和关键词模板是最划算的命中率能到九成调试也直观。超过十个意图规则之间会互相覆盖那时候再考虑 fastText 或 BERT 分类器但那是另一个量级的工程了。3.2 实体识别用词典最长匹配而不是 Jieba 裸跑医疗词表里「心绞痛」「心肌梗死」这类词Jieba 默认词典经常切碎。给 Jieba 加自定义词典能缓解但词频权重偶尔还会把它切开。更稳的做法是用 flashtext 这类纯词典匹配库做最长匹配顺便把别名统一成标准名。from flashtext import KeywordProcessor def build_entity_index(dict_path): kp KeywordProcessor(case_sensitiveFalse) with open(dict_path, encodingutf-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 aliasescoalesce处理没有别名的节点避免导出文件里出现空字段把解析脚本搞挂。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_FORBIDDENNEG_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 模板的映射表每个意图对应一条固定模板查询语句里只有实体值是变量结构完全由后端白名单控制。意图Cypher 模板SYMPTOM_OF_DISEASEMATCH (d:Disease {name:$name})-[:HAS_SYMPTOM]-(s:Symptom) RETURN s.name AS answerDISEASE_OF_SYMPTOMMATCH (s:Symptom {name:$name})-[:HAS_SYMPTOM]-(d:Disease) RETURN d.name AS answerDRUG_OF_DISEASEMATCH (d:Disease {name:$name})-[:RECOMMEND_DRUG]-(dr:Drug) RETURN dr.name AS answerFOOD_RECOMMENDMATCH (d:Disease {name:$name})-[:RECOMMEND_FOOD]-(f:Food) RETURN f.name AS answerFOOD_FORBIDDENMATCH (d:Disease {name:$name})-[:NO_EAT_FOOD]-(f:Food) RETURN f.name AS answerCHECK_OF_DISEASEMATCH (d:Disease {name:$name})-[:NEED_CHECK]-(c:Check) RETURN c.name AS answerDEPARTMENT_OF_DISEASEMATCH (d:Disease {name:$name})-[:BELONG_TO_DEPARTMENT]-(dep:Department) RETURN dep.name AS answer注意反向关系的写法。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, nameentity).data() except Exception: return [] answers [r[answer] for r in records] # dict.fromkeys 去重且保持顺序 return list(dict.fromkeys(answers))session.run(cypher, nameentity)里的nameentity就是参数绑定值是第 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 answerUNION把两条独立查询的结果合并应用层根据返回的字符串直接出答案不用写两段查询逻辑再拼结果。这类模板在实现时要注意RECOMMEND_DRUG和FORBID_DRUG在同一条路径上同时存在时UNION会返回两行需要在组装层做「禁忌优先」的排序规则。另一个常用多跳是并发症查询。第 2 章如果加了COMPLICATION关系一条查询就能覆盖「高血压会引起哪些并发症」MATCH (d:Disease {name:$name})-[:COMPLICATION]-(c:Disease) RETURN c.name AS answer4.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, topk3, threshold0.1): scored [(c, jaccard(question, c)) for c in candidates] scored.sort(keylambda x: x[1], reverseTrue) 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的列名对不上。拿到数据后先用三行代码摸底不要急着启动 Neo4jimport pandas as pd rel pd.read_csv(data/relation.csv, encodingutf-8) print(rel.columns.tolist()) print(rel[disease].nunique(), len(rel))列名列表和节点数一出来就能判断数据规模够不够支撑演示。数据集下载后如果编码是 GBKencodingutf-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, encodingutf-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 比对比单条用例更可靠。把这套回归脚本固定下来换数据集、加实体类型时先跑一遍准确率变化立刻变成可对比的数字比凭感觉调规则靠谱得多。本文还有配套的精品资源点击获取