Python医疗知识图谱问答系统：从本体设计到Neo4j落地

简介面向有一定Python基础并对知识图谱、医疗NLP感兴趣的开发者这是一套医疗领域知识图谱问答系统后台实现整合了基础数据爬取、知识图谱构建、自动问答三大模块完整代码与预处理数据均已打包配置好环境即可直接运行。系统使用Neo4j 3.2.2存储医学实体关系配合jieba分词、TensorFlow训练深度模型其中BiLSTM-CRF承担命名实体识别TextCNN承担问题分类图谱构建与问答解析代码分层清晰可作为课程设计或毕设项目基底。压缩包共57个文件约120.5MB主要包含14个Python源文件、16个txt数据与词典配置、2套TensorFlow模型文件checkpoint、meta、data、index、JSON医疗数据集、npy向量数据及可视化截图/动图等目录按数据、模型、图谱、问答等模块划分。已有4395人浏览学习。随包README对运行环境、版本匹配和训练/部署差异做了说明结合图片与动图可直观追踪爬虫、建图、实体识别、问题匹配到答案生成的整体链路省去重复搭建时间。1. 医疗知识图谱问答系统把「头痛挂什么科」变成一条可回溯的图查询医疗问答不能只追求“回答得像”更要“回答得有理有据”。基于 Python 的知识图谱医疗问答系统把症状、疾病、药物、科室之间的关联显式建模成实体—关系图用户每次提问都被翻译成一条可回溯的图路径查询。这一步恰恰是传统关键词检索和纯模型生成都难做到的——可解释、可修正、可离线部署。这个方向的落地形态以“完整代码 可直接运行数据”居多用最常见的技术栈搭一套从数据清洗、图谱构建到问答响应的闭环适合正在做医疗导诊、临床辅助决策的工程团队也适合想快速验证知识图谱路线的 Python 开发者。多花半小时把图谱建模想清楚后面问答的准确率就能高一大截。2. 医疗知识图谱的数据准备与本体设计先定实体关系再谈问题边界2.1 本体建模实体、关系与属性三层设计先画一张表医疗问答系统的核心在于“实体的边界”和“关系的语义”。初学者最容易犯的错是一上来就导数据结果发现“肺炎”既是疾病名又是检查报告里的诊断字段还是病历中的病种——同一个词在不同语境里指的东西完全不一样。所以第一件事不是写代码是画一张实体—关系表。常见做法是用 Excel 维护一份 schema因为团队里的业务同事也能参与评审。表格至少要有四列实体类型、实体示例、关系类型、关系方向。医疗领域最通用的本体是“疾病、症状、药物、科室、检查”五类实体关系包括“疾病表现为症状”“疾病就诊于科室”“疾病治疗用药物”“疾病需要检查”。这个规模足够撑起一个导诊和用药咨询的问答原型也符合工业场景下知识图谱设计的“先小后大”原则——先验证链路再扩规模。把这张表落到代码里就是常说的 schema 文件。它不只用来建图更用来约束后面所有查询模板的编写。schema 定义得越细问答系统的边界就越清晰——什么能答、什么答不了在建图之前就能判断。# 医疗知识图谱 schemadict 结构方便后续扩展属性 MEDICAL_SCHEMA { entity_types: { 疾病: [感冒, 肺炎, 高血压, 糖尿病], 症状: [头痛, 咳嗽, 发热, 头晕], 药物: [阿司匹林, 布洛芬, 头孢克肟], 科室: [呼吸内科, 神经内科, 心内科], 检查: [血常规, 胸片, 心电图] }, relations: [ (疾病, 表现为, 症状), (疾病, 就诊于, 科室), (疾病, 治疗用, 药物), (疾病, 需要检查, 检查), ] }这段结构里entity_types用 dict 做索引relations用元组列表存“头实体类型—关系名—尾实体类型”。写查询模板和校验数据时这两张表就是唯一事实来源。后续要加新实体比如“手术”“疫苗”在 schema 里加一项下游的建图脚本和查询脚本都要同步改——所以 schema 文件要单独放一个 .py 模块而不是塞在建图脚本里。参数取舍上实体名统一用name属性存带歧义的同义词全部合并进 aliases 列表比如“头疼”和“头痛”归一为同一实体。这一步归一化直接决定后面实体识别的准确率比任何模型调参都划算。2.2 用 Python 清洗医疗原始数据从 CSV 到三元组有了 schema第二步是处理原始数据。假设手头是一张导诊表列是“科室、常见疾病、典型症状、常用药物”行是一百来条常见病记录。这张表不是图谱需要的三元组形态要先拆。import pandas as pd import re # 读取原始导诊表列为: 科室, 疾病, 症状, 药物 df pd.read_csv(medical_data.csv, encodingutf-8-sig) triples [] for _, row in df.iterrows(): dept row[科室] disease row[疾病] symptoms [s.strip() for s in re.split(r[、,], row[症状]) if s.strip()] drugs [d.strip() for d in re.split(r[、,], row[药物]) if d.strip()] for sym in symptoms: triples.append((disease, 表现为, sym)) triples.append((sym, 属于症状, disease)) for drug in drugs: triples.append((disease, 治疗用, drug)) triples.append((disease, 就诊于, dept)) print(f共生成 {len(triples)} 条三元组)拆解逻辑很直白一行记录里如果有多个症状和多个药物就用正则拆开做笛卡尔积式的三元组展开。“属于症状”这个反向关系是为了在用户问“头痛挂什么科”时能从症状反查到疾病再找到科室等于给查询多留了一条通路。这里有一个关键参数分隔符正则[、,]。我见过有人只按中文顿号拆结果数据里混入英文逗号后丢了一半关系。清洗阶段宁可用宽松匹配后面校验时再删脏数据也不要一开始就漏掉关系。跑完上面的代码会得到几千到几万条三元组。接下来要做的是对同一实体名称做去重和归一化。from collections import defaultdict entity_names set() for h, r, t in triples: entity_names.add(h) entity_names.add(t) # 维护一份别名映射表覆盖导诊表里所有出现的叫法 synonyms {头疼: 头痛, 发烧: 发热, 心脏科: 心内科} normalized_triples [] for h, r, t in triples: h synonyms.get(h, h) t synonyms.get(t, t) normalized_triples.append((h, r, t)) unique_triples list(set(normalized_triples)) print(f归一化去重后剩余 {len(unique_triples)} 条三元组)这段代码暴露了整套系统最容翻车的地方——同义词归一化。如果“头疼”和“头痛”在图谱里是两个节点用户问“头疼”时匹配不到“头痛”下的任何关系问答效果直接归零。常见做法是建一张几十条的别名表把导诊表里出现的所有口语叫法收敛掉等数据量大了再用编辑距离或词向量做自动聚类但人工表永远是底线方案。这个阶段的产出物是一份干净的triples.csv三列分别是 head、relation、tail。到这一步数据准备完成可以进入图谱写入环节。3. Neo4j 写入与 Cypher 查询模板把三元组变成可检索的图3.1 建图策略按实体类型分批写入关系最后挂上一章产出的triples.csv是纯文本三元组。这一章的目标是把它导入 Neo4j让查询在毫秒级返回。常见做法是用 py2neo 连上图数据库然后“先节点、后关系”两阶段写入。为什么不是边读边写因为文本里同一个“高血压”会出现在几十行边读边写会产生大量重复节点。先建节点时做去重再遍历三元组挂关系这是做图谱落地的通用路线。from py2neo import Graph, Node, Relationship # 连接 Neo4j注意改成你本地的地址和密码 graph Graph(bolt://localhost:7687, auth(neo4j, 123456), namemedical_kg) # 根据 schema 里定义的实体类型确定每个实体属于哪个类型标签 entity_type_map {} for etype, names in MEDICAL_SCHEMA[entity_types].items(): for name in names: entity_type_map[name] etype def create_all_nodes(): tx graph.begin() # 用 merge 以 name 为唯一键保证同一实体只建一次 for name, etype in entity_type_map.items(): node Node(etype, namename) tx.merge(node, etype, name) graph.commit(tx) def create_all_relations(): # 分批次提交避免单事务过大 BATCH_SIZE 2000 for i in range(0, len(triples), BATCH_SIZE): batch triples[i:i BATCH_SIZE] tx graph.begin() for h, rel, t in batch: htype entity_type_map.get(h) ttype entity_type_map.get(t) if not htype or not ttype: continue head graph.nodes.match(htype, nameh).first() tail graph.nodes.match(ttype, namet).first() if head is not None and tail is not None: tx.create(Relationship(head, rel, tail)) graph.commit(tx)这段代码里能直接调的参数有两个BATCH_SIZE和节点反查方式。BATCH_SIZE设成 2000 是经验值Neo4j 单事务写入几千条关系比较稳妥太大容易触发事务内存上限太小浪费提交开销。如果你想加速可以把实体名到节点的映射维护在 Python 内存里避免每次关系写入都去数据库反查节点——在十万级三元组以下差别不大数据量上去后必须优化。另一个取舍是标签设计。这个版本用了 schema 实体类型做标签比如(:疾病 {name:高血压})查询时可以限定类型性能更好。如果同名实体跨类型不多也可以统一用(:Entity)查询更简单。二选一即可不建议混用否则查询模板要到处判断类型。建完图之后第一步不是写 Python而是在 Neo4j Browser 里人工验证几条路径。比如跑一句MATCH p(:疾病 {name:肺炎})-[:表现为]-(:症状) RETURN p看路径有没有连上。这一步能快速发现关系方向错误、实体名带空白字符、重复节点等脏数据问题比在 Python 里排错快得多。3.2 Cypher 查询模板用「参数化 反向关系」覆盖高频问法图谱验证通过后把高频问题归纳成 Cypher 模板。医疗问答最高频的有四类疾病有什么症状、症状该挂什么科、这病吃什么药、需要做什么检查。// 高频模板 1症状反查科室两跳路径 MATCH (s:症状 {name: $symptom})-[:表现为]-(d:疾病)-[:就诊于]-(c:科室) RETURN DISTINCT c.name AS department // 高频模板 2疾病查症状 MATCH (d:疾病 {name: $disease})-[:表现为]-(s:症状) RETURN DISTINCT s.name AS symptom // 高频模板 3疾病查药物 MATCH (d:疾病 {name: $disease})-[:治疗用]-(m:药物) RETURN DISTINCT m.name AS medicine注意模板 1 里症状到科室是两跳症状先反查到疾病疾病再到科室。建图时如果把症状到疾病的关系方向建反了或者没建反向关系这个模板就会返回空结果。图谱的正确性要靠查询来检验这也是我不厌其烦建议在 Browser 里先验路径的原因。关于参数化在 py2neo 的graph.run()方法里把参数放在第二个位置传入永远不要用 f-string 把用户输入拼进 Cypher。除了注入风险Neo4j 对参数化查询会缓存执行计划同一个模板第二次执行会快很多。# 正确姿势参数放第二个参数 result graph.run( MATCH (:症状 {name: $n})-[:表现为]-(d:疾病) -[:就诊于]-(c:科室) RETURN DISTINCT c.name AS department, n头痛 ).data()$n是占位符n头痛是参数。这样写的好处是模板与数据分离后续换实体名、换问题都不需要改字符串。有经验的团队会把所有模板集中放在一个queries.py里每加一个模板就在 Browser 里用真实数据验一遍而不是直接在 Python 里跑。4. 问答链路实现从原始问句到图谱答案的完整 Pipeline4.1 实体识别用词典匹配替代通用分词稳定性更高问答链路的第一环是实体识别。用户输入“头痛应该挂什么科”系统要先识别出“头痛”这个症状实体才能去查询。垂直医疗场景有一个比通用 NLP 有利的条件实体词大多是专名歧义少而且知识图谱本身就是一部词典——所有实体都能从 schema 或 Neo4j 里导出。import re # 从 Neo4j 导出所有实体词构建词典 def load_entity_dict(graph): entity_dict {} for etype in [疾病, 症状, 药物, 科室, 检查]: nodes graph.nodes.match(etype) names [n[name] for n in nodes] entity_dict[etype] names return entity_dict def recognize_entity(question, entity_dict): # 优先匹配最长实体词避免「胸口痛」里匹配出「口痛」 matched [] for etype, names in entity_dict.items(): for name in sorted(names, keylen, reverseTrue): if name in question: matched.append((etype, name)) question question.replace(name, * len(name)) return matched这段代码的核心逻辑是“最长优先”和“替换屏蔽”。先按词长从大到小排序能让“头痛”优先于“头”被匹配匹配成功后把原词替换为空格避免同一个片段被其他实体词二次命中。注意这里没有用 jieba。知识点来了在医疗专名词典完备的情况下词典匹配常比通用分词更稳。jieba 会把“上呼吸道感染”切成“上呼吸道”和“感染”而词典匹配不受这个影响。等词典覆盖不全了再接模型做补充这是“词典兜底 模型扩展”的常用工程路径。实体识别完成后还要把识别结果和问题类型对齐。如果一句话里识别出多种实体比如“感冒了咳嗽吃什么药”里“感冒”是疾病、“咳嗽”是症状需要结合下一节的意图分类来决定用哪个实体去查询。4.2 意图分类用规则匹配覆盖高频问法可解释可修正意图分类决定了查询模板的选择。常见实现有两种基于规则的关键词匹配和基于分类模型或 BERT 的语义分类。在医疗问答的垂直场景里我倾向于先用规则——意图数量不超过十个每个意图的关键词特征非常明显规则方案没有训练成本、可解释性强、改起来快。INTENT_RULES { query_symptom: [什么症状, 有哪些表现, 症状], query_department: [挂什么科, 去哪个科室, 科室], query_medicine: [吃什么药, 用什么药, 用药], query_check: [做什么检查, 检查项目, 查什么], } def classify_intent(question): # 多关键词命中时按 dict 顺序取第一个命中的意图 for intent, keywords in INTENT_RULES.items(): for kw in keywords: if kw in question: return intent return unknown规则匹配的粒度按需调整。比如“感冒了该挂什么科”里同时出现疾病名“感冒”和意图词“挂什么科”意图是query_department拿“感冒”这个实体去查科室。但“感冒咳嗽吃什么药”里没有明显意图词规则会落到unknown这时要做兜底处理优先使用疾病实体默认查询“疾病→药物”。兜底逻辑是问答系统里最容易被忽略的一环。我的经验是unknown 意图绝不能直接回“听不懂”而是按“疾病存在查药物、症状存在返科室”这种统计上的高频路径降级处理。用户不会因为一次答对而夸你但三次“不知道”足以让他不用这个系统。4.3 查询执行与答案组装多模板降级不空手而归识别出实体和意图后剩下的就是组装查询。第一个模板没命中时还要做降级查询——比如按疾病查科室没结果就换按症状查科室。def answer(question, graph, entity_dict): entities recognize_entity(question, entity_dict) intent classify_intent(question) if not entities: return 暂时没识别到疾病或症状信息请换个说法再问一次。 # 按实体类型和意图组装查询这里只展示核心分支 for etype, name in entities: if intent query_department: if etype 症状: result graph.run( MATCH (:症状 {name:$n})-[:表现为]-(d:疾病) -[:就诊于]-(c:科室) RETURN DISTINCT c.name AS dept, nname ).data() elif etype 疾病: result graph.run( MATCH (:疾病 {name:$n})-[:就诊于]-(c:科室) RETURN DISTINCT c.name AS dept, nname ).data() elif intent query_symptom: result graph.run( MATCH (:疾病 {name:$n})-[:表现为]-(s:症状) RETURN DISTINCT s.name AS symptom, nname ).data() # 命中了就直接返回避免后面实体覆盖答案 if result: return assemble_answer(intent, name, result) return 这个问题涉及的关系我还没覆盖试试问问症状挂什么科。assemble_answer是答案组装函数把图查询结果拼成一句完整的话。比如查询结果是[{dept: 神经内科}]拼成“根据已有医学知识建议你去神经内科就诊”。组装逻辑里最忌讳的是只回一个实体名至少要带上问句里的实体让用户能看出来这个答案是从哪个词出发找到的。到这里一条完整的问答链路就闭环了词典实体识别 → 规则意图分类 → 参数化 Cypher → 结果组装。这条链路没有用深度学习模型在常见病导诊场景的准确率能做到 85% 左右。如果后续准确率卡住不动说明问题出在数据覆盖和模板覆盖上而不是模型选型上。5. 医疗问答系统的常见坑现象、原因与处理5.1 子串误伤问「胸口痛」答「口痛」现象用户问“胸口痛挂什么科”系统匹配到“口痛”这个词典里的症状实体给出完全不相关的科室。原因词典匹配没有做最长优先处理。“胸口痛”里包含“口痛”而“口痛”恰好在症状词典里被抢先匹配了。解决匹配前对每个类型的实体表做按长度降序排序确保“胸口痛”先命中命中后把这段文字替换成空格屏蔽后续误匹配。另一个辅助手段是加最小匹配长度阈值比如长度低于两个字的实体词不参与首次匹配。5.2 同义词没归一化问「发烧」查不到「发热」现象图谱里只建了“发热”用户输入“发烧”时实体识别命中“发烧”但图里没有“发烧”节点查询结果为空。原因建图前的清洗阶段没做同义词映射两端各用各的叫法“发烧”和“发热”成了两个不相干的节点。解决清洗阶段手工维护别名表把口语叫法全部归一。注意要在两端同时归一——建图数据要处理用户问题在实体识别后也要再跑一遍synonyms.get(name, name)保证输入侧和存储侧用的是同一套命名。5.3 Neo4j 返回顺序不稳定同样的问题答案每次不同现象同一个问题问两遍返回的科室列表顺序变了用户以为系统“抽风”。原因Neo4j 的MATCH在没有ORDER BY时不保证返回顺序结果顺序受执行计划影响会有波动。解决所有面向用户的查询模板在RETURN前统一加ORDER BY。多结果时按实体名排序比如ORDER BY c.name。这一步不改变答案内容但会让系统行为可预期排障时也不容易误判。5.4 py2neo 事务过大建图写入时内存溢出现象一次性把两万条关系放进一个事务提交py2neo 直接抛内存错误整个建图脚本中断。原因Neo4j 单事务的内存上限受配置约束事务内的写入操作和索引维护都在事务内存里完成数据量过大会撑爆。解决按批提交BATCH_SIZE设置在 2000 到 5000 之间。我的经验是 2000 最稳配合batch[:i BATCH_SIZE]切片。另外每批之间稍作停顿给后端一点刷盘时间建图脚本会慢一点但不会半夜翻车。5.5 空值数据进入图谱出现「nan」节点现象图谱里出现了名为“nan”的实体节点查询结果里也经常返回莫名奇妙的 nan。原因CSV 里的空单元格被 pandas 读成NaN清洗脚本没有过滤把 “nan” 当字符串写进了三元组。解决在做三元组拆分前先df.dropna(subset[疾病, 症状])药物这种允许为空的字段单独处理。更彻底的做法是在读数据后跑一行断言发现有“nan”字样的实体直接抛异常终止建图宁可中断也不要污染图谱。提示以上五条属于知识图谱问答里最常踩的坑。每条都可以在本地复现修复成本都不高但漏掉任何一条都会让问答系统的体验断崖式下跌。6. 验证与进阶迭代先测准准确率再谈换模型当整个链路跑通后下一步不是急着加模型而是建一套回归测试集。常见做法是准备 100 条问题覆盖科室查询、症状查询、药物查询和跨类型兜底四种情况每条都手工标好预期答案。跑一遍系统统计三个指标实体识别准确率、意图分类准确率、最终答案命中率。这三个指标的诊断意义不同。实体识别准确率差优先补词典和同义词表意图分类差加规则关键词两者都正常但答案命中率低问题出在 Cypher 模板覆盖不全比如只做了疾病查科室没做症状查科室。按这个顺序迭代效率最高。进阶方向有两个按需求选。一个是把规则意图分类换成 BERT 意图分类器适合问法多变、规则维护成本高的场景另一个是给实体识别接上基于 BERT 的命名实体识别模型解决词典外新词的问题。这步升级的代价是引入训练数据和推理延迟但换来的是对口语问法的泛化能力。我的习惯是先跑一个月规则版本收集真实问题日志把高频 miss 案例整理成新规则或新数据再决定是否上模型。这套方案做完你会拥有一套完全可控、可解释、可离线跑的医疗问答原型。数据换一换实体类型改一改同样一套代码也能复用到法律、电商客服等领域。我自己的教训是图谱问答的瓶颈从来不在算法而在数据是否干净、模板是否覆盖真实问题。先把这两件事做扎实再谈优化。希望帮到你。本文还有配套的精品资源点击获取