简介这是一份基于Python的医疗知识图谱问答系统完整源码主要面向毕业设计、期末大作业和课程设计学生也适合想了解知识图谱落地应用的开发者。项目将医疗数据构建为知识图谱支持自然语言问诊式查询覆盖意图识别、命名实体识别、知识抽取与查询匹配等关键环节并包含构建图谱、意图服务、NER服务等子模块代码注释详细思路清晰新手也能快速读懂。资源共70个文件以py源码为主配以json配置、pkl模型、docx文档和bat一键启动脚本压缩包容量51.62MB目录按nlu、knowledge_extraction、build_kg等模块划分便于检索和二次开发。目前已有149人学习下载。下载后参考使用说明与依赖清单即可完成部署系统界面简洁、功能完善既可直接用于答辩展示也可作为医疗问答方向的实战学习样本。1. 医疗知识图谱问答系统不是ChatGPT套壳是结构化知识检索当你在python里调用一个大模型接口问“头痛伴发热吃什么药”得到一段流畅但可能编造来源的回答时你就知道纯粹的生成式大模型在医疗场景里有多危险。医疗知识图谱问答系统是另一条路先构建一张由疾病、症状、药物、科室、检查等实体和它们之间关系组成的图再用python把用户问句解析成结构化查询最后从图里捞出确定性的答案。它不能写诗但能回答“高血压患者禁用哪类药”时给出可追溯的实体路径。适合做医疗导诊、用药提醒、病历结构化检索的从业者也适合想用neo4j python做垂直领域问答的开发者。本文从零拆解一套可落地的实现方案。2. 医疗问答背后的技术栈拆解知识图谱为什么比关键词搜索更可靠2.1 医疗场景对问答系统的三个硬约束医疗问答不是“搜到带关键词的网页”就结束。第一个硬约束是答案必须可解释。患者问“糖尿病可以吃西瓜吗”系统不能只给一个“可以”或“不可以”得能回溯到“糖尿病-禁忌食物-西瓜”这条关系链让医生或患者看到推理依据。第二个硬约束是实体名称极度不规范。同一药物有商品名、通用名、别名同一疾病有俗称和标准名“高血压”和“hypertension”指向同一个实体。第三个硬约束是错误代价高。答错一个菜谱问题顶多是口味问题答错用药禁忌可能是医疗事故。所以系统必须设计拒答机制宁可说“我不确定”也不要强行生成。这三个约束决定了医疗问答不能走纯文本匹配必须构建一个显式的、结构化的知识表示层。2.2 知识图谱 vs 向量库 vs 传统数据库选型依据先看传统关系型数据库。你有一张疾病表和一张症状表通过外键关联。问“什么病会头疼发热”SQL要写多个JOIN而且随着关系维度增加药物禁忌、检查指标、手术方式表数量爆炸查询模板越来越难维护。向量库适合语义相似检索但它的输出是“文本片段”不是“实体关系”无法回答“这个药和那个药是否冲突”这类需要关系推理的问题。知识图谱用节点和边建模把“疾病-症状-药物-科室”天然表达为图结构查询路径就是答案路径。医疗领域的实体关系非常稠密一个疾病可能关联几十个症状、十几种药、多个并发症图模型最贴合这种网状结构。实际项目里常见做法是用Neo4j存知识图谱用Elasticsearch或jieba词典做实体召回再用python逻辑层做意图识别和查询组装。向量库可以作为辅助对长尾开放性问题做答案候选召回但最终答案必须经过图谱验证。选型时还要考虑团队技术栈Neo4j的Cypher查询语言对python开发者来说学习曲线平缓官方py2neo和neo4j驱动都成熟这也是医疗图谱项目里Neo4j使用率高的原因。2.3 最小系统架构从问句到答案的数据流一套最小可运行系统由四个模块组成问句理解、实体链接、图谱查询、答案生成。用户问句 - 问句理解意图分类 槽位提取 - 实体链接识别“高血压”指向图谱中哪个节点 - 图谱查询根据意图生成Cypher查出子图 - 答案生成把节点和关系拼成自然语言回答问句理解这一步我一般不引入复杂的BERT模型而是用规则词典先跑通。医疗问句的句式相对固定“XX的症状有哪些”“XX不能吃什么药”“XX挂什么科”用正则和关键词就能覆盖80%的意图。实体链接用Aho-Corasick算法匹配词典配合双向最大匹配解决歧义。图谱查询是核心Cypher模板按意图分类预先写好比如意图为“symptom_query”就执行匹配疾病节点并返回其症状关系。答案生成把图查询结果填进预制的自然语言模板。这个架构的好处是每一层都能单独测试。实体链接错了查Neo4j日志能看到抽出了什么词Cypher写错了报错直接定位答案生成了可以反查路径验证是否正确。相比一个黑匣子式的深度学习模型这种白盒结构在医疗场景里更容易通过审核。3. 构建医疗知识图谱数据清洗、实体关系设计与存储落地3.1 医疗数据源与实体关系设计疾病-症状-药物-科室开源的医疗数据源有不少常见的有CCKS评测提供的医疗实体标注数据、中文症状库、药品说明书结构化数据、公开的疾病百科数据。这些数据质量参差不齐有的字段缺失有的实体名和标准名不一致。我一般先把数据统一成三元组格式(头实体, 关系, 尾实体)例如(糖尿病, 禁忌食物, 西瓜)、(头痛, 表现为, 偏头痛)。实体类型至少要建四类疾病Disease、症状Symptom、药物Drug、科室Department。再加上两类辅助实体检查Exam和食物Food。关系设计要面向问答需求关系头实体尾实体问句示例表现为疾病症状高血压有哪些症状治疗用药疾病药物高血压用什么药禁忌药物疾病药物高血压不能吃什么药所属科室疾病科室头痛挂什么科推荐检查疾病检查糖尿病要做什么检查禁忌食物疾病食物糖尿病不能吃什么这张表就是图谱的Schema。设计时要注意关系方向统一比如“治疗用药”头实体是疾病、尾实体是药物反过来查询“这个药治什么病”时用Cypher的逆向匹配(d:Disease)-[:治疗用药]-(drug:Drug)并不需要额外建一条反向关系。用属性relation_type标记关系类型比给每个关系单独建类型更灵活但查询时会多一次属性过滤数据量在十万级以下性能差别不大建议直接用单独的关系类型可读性好。3.2 Neo4j落地实体、关系、属性的Cypher写法数据清洗完成后用Cypher把三元组写入Neo4j。先创建约束保证实体唯一再逐条创建关系。以下是核心脚本// 创建唯一约束避免重复节点 CREATE CONSTRAINT FOR (d:Disease) REQUIRE d.name IS UNIQUE; CREATE CONSTRAINT FOR (s:Symptom) REQUIRE s.name IS UNIQUE; CREATE CONSTRAINT FOR (dr:Drug) REQUIRE dr.name IS UNIQUE; CREATE CONSTRAINT FOR (dep:Department) REQUIRE dep.name IS UNIQUE; // 创建疾病节点 MERGE (d:Disease {name: 糖尿病}) ON CREATE SET d.description 以高血糖为特征的代谢性疾病; // 创建症状节点并建立“表现为”关系 MERGE (s:Symptom {name: 多饮}) MERGE (d:Disease {name: 糖尿病}) MERGE (d)-[:表现为]-(s);这段代码里有两个关键操作MERGE是“有则匹配、无则创建”配合唯一约束可以防止重复节点。ON CREATE SET只在节点首次创建时写入属性更新数据时不会覆盖已有描述。批量导入数据时如果不用约束MERGE频繁扫描会让写入慢几十倍所以约束一定要先建。关系创建后先用几个查询验证图的连通性。查某个疾病的所有症状MATCH (d:Disease {name: 糖尿病})-[:表现为]-(s:Symptom) RETURN s.name;3.3 数据导入的两种方式CSV批量导入与Python逐条写入数据量在几万条级别用Python驱动逐条写入即可代码简单易调试。数据量超过几十万条推荐使用Neo4j的neo4j-admin import或LOAD CSV。两种方式的适用场景不同# 方式一Python逐条写入数据量10万时推荐 from neo4j import GraphDatabase class MedicalGraphBuilder: def __init__(self, uri, user, password): self.driver GraphDatabase.driver(uri, auth(user, password)) def add_relation(self, head_type, head_name, rel, tail_type, tail_name): cypher ( fMERGE (h:{head_type} {{name: $head_name}}) fMERGE (t:{tail_type} {{name: $tail_name}}) fMERGE (h)-[:{rel}]-(t) ) with self.driver.session() as session: session.run(cypher, head_namehead_name, tail_nametail_name) def close(self): self.driver.close() if __name__ __main__: builder MedicalGraphBuilder(bolt://localhost:7687, neo4j, your_password) # 格式头实体类型, 头实体名, 关系, 尾实体类型, 尾实体名 triples [ (Disease, 糖尿病, 表现为, Symptom, 多饮), (Disease, 糖尿病, 禁忌食物, Food, 西瓜), (Disease, 头痛, 所属科室, Department, 神经内科), ] for head_type, head_name, rel, tail_type, tail_name in triples: builder.add_relation(head_type, head_name, rel, tail_type, tail_name) print(f已写入: {head_name} -{rel}- {tail_name}) builder.close()这段代码用f-string拼接Cypher注意实体名和关系名绝不能直接接受用户输入否则有注入风险。上面代码里的实体名是通过参数传的$head_name只作参数传递但关系名rel是直接拼接的所以关系名必须来自白名单。我把关系名做了一层映射REL_MAP {表现为: 表现为, 治疗用药: 治疗用药, ...}传入时先校验。方式二用LOAD CSV批量导入先把三元组存成CSV文件每条记录是head_name,rel_name,tail_name,head_type,tail_typeLOAD CSV WITH HEADERS FROM file:///medical_triples.csv AS row MERGE (h:Disease {name: row.head_name}) MERGE (t:SYM {name: row.tail_name}) MERGE (h)-[:REL {type: row.rel_name}]-(t);注意LOAD CSV的文件路径是Neo4j服务端目录不是python脚本所在目录。我之前在这个坑上卡了半小时文件明明存在却报找不到后来把CSV放到$NEO4J_HOME/import目录才解决。另外CSV里的中文最好转为UTF-8无BOM格式否则第一条记录会带上不可见字符导致匹配不上。4. 问答核心流程实现意图识别、实体链接与答案生成的Python代码4.1 问句预处理与意图分类用户的问句进来先做基础清洗去除空格、统一大小写、替换全角标点。然后做意图分类。医疗问句意图可以分成这几类症状查询、用药查询、禁忌查询、科室推荐、检查推荐。常见做法是维护一个意图模板表import re INTENT_PATTERNS { disease_symptom: [r(.?)的?症状(?:是|有|包括)?(?:哪些|什么)?, r(.?)(?:有哪些|有什么)症状], disease_drug: [r(.?)用(?:什么|哪些)(?:药|药物), r(.?)(?:吃什么|吃啥)药], disease_drug_forbid: [r(.?)不能吃(?:什么|哪些)药, r(.?)禁用(?:什么|哪些)药], disease_department: [r(.?)挂(?:什么|哪个)科, r(.?)(?:应该|该)去(?:什么|哪个)科室], disease_exam: [r(.?)做(?:什么|哪些)(?:检查|项目)], } def parse_intent(question: str) - str: for intent, patterns in INTENT_PATTERNS.items(): for pattern in patterns: match re.search(pattern, question) if match: return intent, match.group(1) return unknown, question这里有个小坑正则模板的优先级很关键。不能吃什么药如果先匹配了吃什么药就会把“不能”漏掉所以禁忌类意图必须放在普通用药意图之前。我在字典序上把disease_drug_forbid放在disease_drug前面同时把不能吃这类否定词做前置校验NEGATIVE_WORDS [不能, 不可以, 禁忌, 禁止, 不要] def parse_intent(question: str) - str: has_neg any(w in question for w in NEGATIVE_WORDS) for intent, patterns in INTENT_PATTERNS.items(): if has_neg and intent ! disease_drug_forbid: continue # 包含否定词时先只试禁忌类 for pattern in patterns: match re.search(pattern, question) if match: return intent, match.group(1) return unknown, question意图识别做到这个程度已经能覆盖大部分模板化问句。遇到无法匹配的问句不要急着硬猜直接返回unknown进入拒答流程。4.2 实体链接从问句抽疾病名和症状名意图识别拿到了问句中的关键短语比如“高血压的症状有哪些”提取出“高血压”。但要链接到图谱里的实体还需要处理“高血压”和“高血压病”、“糖尿病”和“DM”这类同义关系。我的方案是维护一份同义词表并在实体表里设置alias属性from pyahocorasick import Automaton class EntityLinker: def __init__(self, entity_dict: dict): # entity_dict: {高血压: Disease, 高血压病: Disease, 西瓜: Food} self.automaton Automaton() for word, etype in entity_dict.items(): self.automaton.add_word(word, (word, etype)) self.automaton.make_automaton() def link_all(self, text: str) - list: matches [] for end_index, (word, etype) in self.automaton.iter(text): start_index end_index - len(word) 1 matches.append({entity: word, type: etype, start: start_index, end: end_index}) # 按开始位置排序并合并重叠匹配 matches.sort(keylambda x: (x[start], -len(x[entity]))) merged [] for m in matches: if merged and m[start] merged[-1][end]: continue # 跳过被前面更长实体覆盖的匹配 merged.append(m) return merged为什么用Aho-Corasick而不是jieba因为实体名词典是专有名词集合jieba会把“糖尿病人”切分成“糖尿病”和“人”而AC自动机能一次扫描出所有字典命中词速度快且可控。合并重叠匹配时我采用“最长优先保留”策略比如问句里同时有“高血压”和“高血压病”如果两个词都命中词典则保留更长的“高血压病”。这个策略在医疗术语里很重要因为很多病名是另一个病名的前缀。实体名归一化放在查询阶段处理。我从Neo4j里加载所有节点的name和alias属性到内存统一转成小写链接时先用别名映射到标准名ALIAS_MAP { 高血压病: 高血压, dm: 糖尿病, 高血糖: 糖尿病, # 注意高血糖是症状这里仅示例 } def normalize_entity(entity_text: str) - str: return ALIAS_MAP.get(entity_text, entity_text)同义词表是医疗问答的命脉。第一次做完系统后我用100条真实问句测试发现实体链接准确率只有72%原因就是市面上有一个病叫“高血压危象”和“高血压”是不同实体却被前缀匹配切错了。解决方式是引入边界词校验匹配到实体后检查该词前后字符是否属于实体名的合法边界避免把“高血压危象”切出“高血压”。4.3 Cypher查询模板与答案生成意图和实体都确定后组装Cypher查询。每种意图对应一个模板函数模板接受实体名参数GRAPH_QUERIES { disease_symptom: ( MATCH (d:Disease {name: $entity})-[:表现为]-(s:Symptom) RETURN s.name AS answer LIMIT 20 ), disease_drug: ( MATCH (d:Disease {name: $entity})-[:治疗用药]-(drug:Drug) RETURN drug.name AS answer LIMIT 20 ), disease_drug_forbid: ( MATCH (d:Disease {name: $entity})-[:禁忌药物]-(drug:Drug) RETURN drug.name AS answer LIMIT 20 ), disease_department: ( MATCH (d:Disease {name: $entity})-[:所属科室]-(dep:Department) RETURN dep.name AS answer ), disease_exam: ( MATCH (d:Disease {name: $entity})-[:推荐检查]-(e:Exam) RETURN e.name AS answer LIMIT 20 ), } def query_graph(entity: str, intent: str): cypher GRAPH_QUERIES.get(intent) if not cypher: return None with GraphDatabase.driver(URI, auth(USER, PASSWORD)) as driver: with driver.session() as session: result session.run(cypher, entityentity) return [record[answer] for record in result]答案生成不是直接把节点名拼起来而是结合意图生成自然语言。比如查询结果是[多饮, 多尿, 体重减轻]生成“糖尿病常见的症状包括多饮、多尿、体重减轻。”。如果查询结果为空不要硬编答案走拒答流程。INTENT_ANSWER_TEMPLATE { disease_symptom: lambda disease, symptoms: f{disease}常见的症状包括{, .join(symptoms)}。, disease_drug: lambda disease, drugs: f{disease}的治疗药物有{, .join(drugs)}。, disease_drug_forbid: lambda disease, drugs: f{disease}患者禁忌使用的药物包括{, .join(drugs)}。, disease_department: lambda disease, dept: f{disease}建议挂{dept}。, disease_exam: lambda disease, exams: f{disease}通常需要做的检查包括{, .join(exams)}。, } def generate_answer(entity: str, intent: str, query_result: list) - str: if not query_result: return f抱歉我暂时没有找到{entity}相关的{intent}信息。 template_func INTENT_ANSWER_TEMPLATE[intent] if intent disease_department: return template_func(entity, query_result[0]) return template_func(entity, query_result)这里有个细节LIMIT 20是硬限制。一个疾病的症状可能有三十多条全部列出来用户也记不住限定20条后加一句话“以上为常见症状具体请以医生诊断为准”。医疗问答必须带免责声明这也是规范要求。4.4 最小可运行Demo整合成QASystem类以上组件整合成一个类提供answer(question)接口class MedicalQASystem: def __init__(self, builder, linker): self.builder builder # 复用数据构建模块中的Neo4j连接 self.linker linker def answer(self, question: str): intent, entity_phrase parse_intent(question) if intent unknown: return 抱歉我无法理解这个问题。请换一种方式提问比如“高血压有哪些症状”。 linked self.linker.link_all(entity_phrase) if not linked: return 未能识别出有效的疾病或症状实体。 entity_text normalize_entity(linked[0][entity]) result query_graph(entity_text, intent) return generate_answer(entity_text, intent, result) if __name__ __main__: # 初始化实体词典和同义词表 entity_dict load_entity_dict_from_neo4j() linker EntityLinker(entity_dict) qa MedicalQASystem(builderNone, linkerlinker) test_questions [ 高血压都有什么症状, 糖尿病用什么药, 头痛应该挂哪个科, 高血压不能吃什么药, ] for q in test_questions: print(f问{q}) print(f答{qa.answer(q)})运行这段代码前确保Neo4j已启动并且load_entity_dict_from_neo4j从库里读取了所有节点名。这个Demo没有用到机器学习和深度学习适合python入门阶段的开发者复现也适合快速验证图谱数据质量。如果换成真实项目你还需要给answer加日志记录、耗时统计、兜底回复这些后面避坑章细说。5. 避坑与排查医疗问答项目里常见的5个翻车现场5.1 现象药品别名查不到原因同义词表缺失用户问“阿莫西林和青霉素过敏的人能吃吗”系统识别出“阿莫西林”但图谱里只有“阿莫西林胶囊”这个节点链接失败返回“未识别出有效实体”。原因是建立图谱时只收了标准药品名没有收别名和剂型词。解决在药品实体上增加alias属性把“阿莫西林”“阿莫仙”“阿莫西林胶囊”都挂到同一标准节点查询时对name和alias同时做匹配。Cypher可以写成MATCH (d:Drug) WHERE d.name $entity OR $entity IN d.alias。这个工作看起来琐碎却是医疗问答体验的分水岭。5.2 现象问句里同时出现两个疾病实体链接抽错问“高血压和糖尿病有什么症状”意图模板捕获到“高血压和糖尿病”AC自动机匹配出两个实体。我的原始实现只取第一个返回了高血压的症状用户立刻觉得系统“瞎”。解决修改意图流程当检测到多个实体时根据连接词拆分问句分别查询再合并答案。另外模板里的正则表达式要用非贪婪匹配避免把“和”字前后两个病名连在一起。更稳妥的方案是让实体链接返回所有匹配项交给下游逻辑判断是并列还是修饰关系而不是直接取第一个。5.3 现象Neo4j查询超时索引没建图谱写入10万节点后查询“糖尿病的治疗药物”竟然卡了十几秒。查Cypher执行计划发现每次都是全库扫描Disease标签。原因只建了唯一约束但Cypher里用了WHERE d.name $entity这个查询在未命中索引时走全表扫描。解决对name属性建立索引或者用MATCH (d:Disease {name: $entity})语法让优化器自动用索引。注意唯一约束和索引是两回事约束保证不重复索引加速查询。建议先建约束再导入数据导入完成后对所有常用查询字段补建索引。5.4 现象部署后中文乱码/编码问题本地跑得好好的部署到Linux服务器后问答结果显示乱码。排查过程先看Neo4j导入文件编码再检查Python脚本头部是否声明UTF-8最后发现是CSV文件在Windows下保存为GBK编码LOAD CSV读取时按UTF-8解码导致乱码。解决所有数据文件统一转换为UTF-8无BOM格式Python代码里读写文件显式指定encodingutf-8。另外Neo4j的数据库配置里如果修改过dbms.memory.heap.max_size重启后可能需要清除旧索引重建。这个坑和python本身无关但很容易发生在“本机正常、服务器翻车”的部署场景。5.5 现象答案模板生硬长尾问句无法回答“我爸爸有糖尿病平时需要注意什么”这类问句意图模板匹配失败走了拒答。拒答是安全的但体验差。解决把这类问句归入“general_advice”意图图谱里建立“注意事项”属性或者依赖“并发症、禁忌食物、推荐检查”多条关系联合生成答案。更进一步的方案是引入一个轻量分类模型比如FastText对问句向量化意图模板负责确定性高的短句分类模型兜底长尾问句两层结合。这个方案的工作量不大但可以显著提升覆盖率。我后来在项目里加了这个模块未知意图率从22%降到了9%。5.6 现象角色权限和数据更新伦理问题被忽略医疗图谱内容如果来自公开网页需要注意数据授权和更新频率。药品说明书、指南类数据经常修订图谱不能“导入一次用三年”。常见做法是给每条三元组加source和updated_at属性定期重跑导入任务。另外问答系统输出的医疗信息必须附带“请咨询医生”之类的提示这是医疗AI的基本伦理要求也是项目能否通过审核的底线。6. 把系统做得更可靠评测指标、拒答机制与迭代习惯医疗问答系统的开发最后拼的不是模型花哨而是评测和迭代闭环。我的做法是准备一份200条真实问句的测试集分成四类模板内问句、同义改写问句、跨实体复杂问句、完全无关的闲聊。每轮迭代后跑一遍统计准确率、覆盖率、拒答率三个数。准确率是回答正确且图谱路径可解释的比例覆盖率是回答了问题且内容非空的比例拒答率是主动说不知道的比例。这三个数互相制约调阈值时要有取舍。我一般把拒答率控制在10%以下准确率目标大于90%宁可多拒答也不乱答。拒答机制除了意图匹配失败触发还要在图谱查询结果为空时触发。另外加一个置信度判断实体链接匹配到多个候选且分数接近时返回一个追问“您说的是高血压还是高血压危象”而不是直接取第一个。这个交互方式比强硬给一个可能出错的答案要好得多。验证自动化也要做。我写了一个简单的python脚本把测试集和期望答案存成JSON每次改完代码执行一遍import json, sys with open(test_cases.json, r, encodingutf-8) as f: cases json.load(f) def run_evaluation(qa_system): correct 0 total len(cases) for case in cases: answer qa_system.answer(case[question]) if case[expected] in answer: correct 1 else: print(fFAIL: {case[question]} {answer}) print(f准确率: {correct / total:.2%}) if __name__ __main__: qa MedicalQASystem(...) run_evaluation(qa)这个脚本简单到没有用pytest但足够在调整同义词表或模板后快速回归。注意测试集里的期望答案要留余量不要把“高血压常见的症状包括头晕、头痛、心悸”这种全量字符串作为期望而是用“头晕”这种关键实体做包含验证。最后分享一个习惯每次发布前我会把用户问句和系统回答导出来人工检查20条重点看那些“回答了但路径可疑”的case。比如图谱里如果有一条错误关系“高血压-治疗用药-西瓜霜”系统就会一本正经地推荐西瓜霜这种错误比拒答更危险。所以我会定期抽查随机20条回答的图谱路径确保质量不滑坡。医疗知识图谱问答系统做到能上线靠的就是这种“慢工出细活”的评测和迭代希望帮到你。本文还有配套的精品资源点击获取