简介基于Python的医疗知识图谱自动问答系统源码包适合对自然语言处理、知识图谱和智慧医疗感兴趣的开发者、研究员及毕业设计选题学生。项目从数据预处理入手覆盖医学词汇、病症、药物、治疗方法等信息的清洗、去重与标准化进而借助Neo4j等图数据库将实体和关系建模构建结构化的医疗知识图谱同时使用NLTK、spaCy等工具完成问句的语义分析、实体识别与关系抽取配合Cypher查询语句实现问题与图谱节点的智能匹配最终通过Flask、Django搭建直观的问答交互界面形成从数据到回答的完整闭环。压缩包共26个文件以8个py源码脚本、8个txt词典与知识数据、5个xml工程配置、2个json规则文件为主大小15.54MB目录拆分清晰包含数据准备、图谱构建、问题分类、答案搜索等模块并附带演示图片方便对照运行结果。已有184人学习下载可作为从零搭建医疗问答系统的完整参考帮助掌握实体识别、意图分类、图谱查询、回答生成等关键技术也为扩展其他垂直领域问答提供了可复用的代码框架。1. 医疗知识图谱自动问答系统先搞清楚它解决什么问题前台护士每天要回答几十次“头疼挂什么科”一套基于 Python 的医疗知识图谱自动问答系统会先在知识图谱上查“头痛→神经内科”这条边再返回科室和下一步建议。它不是大模型不靠猜答案来自图中真实事实可解释、可控、可离线运行。这类系统由三块组成医疗知识图谱构建、问句解析与意图识别、查询与应答。标题里的“源码”就是这条链路的完整工程形态。适合正在入门 python、想用知识图谱做项目的开发者也适合医疗信息化团队拿来快速搭一套可演示的问答原型。2. 图谱构建从表格到Neo4j决定问答上限的一步2.1 一份最小可用的医疗数据与关系设计先解决“数据从哪来”。常见做法是收集三类结构化表格疾病-症状、疾病-科室、药品-适应症。很多公开医疗网站的“疾病百科”“就医指南”栏目就是这类知识抓下来整理成 CSV。这里有个误区觉得“数据量越大系统越聪明”。实际上图库问答的链路不是靠规模撑起来的100 个高频疾病、300 个症状、30 个科室、50 种常用药就能让系统有内容可答关键是覆盖高频问法和高频词。真正要花心思的是关系设计。我第一版会只保留四条关系边关系类型起点实体终点实体示例HAS_SYMPTOM疾病症状高血压 - 头晕DEPARTMENT疾病科室高血压 - 心血管内科TREATS药品疾病硝苯地平 - 高血压CHECKS疾病检查高血压 - 血压监测这张表就是知识图谱的骨架。实体类型也不多Disease、Symptom、Department、Drug、Check后面所有问答都围着这五种节点转。关系一旦定下来别急着加“并发症”“相关疾病”这类边先把主干跑通。2.2 数据清洗与三元组生成pandas先把脏数据挡在门外拿到 CSV 之后第一步不是入库是清洗。网页复制出来的表格经常带首尾空格、全角空格、换行符不清洗就会在 Neo4j 里形成“高血压”和“高血压 ”两个节点问答匹配时时好时坏非常折磨人。import pandas as pd def clean_entity_name(value): 统一实体名去掉首尾空格、全角空格、换行符避免匹配时翻车 if pd.isna(value): return None return str(value).strip().replace( , ).replace(\u3000, ).replace(\n, ) def build_triples(disease_symptom_file, disease_dept_file): ds pd.read_csv(disease_symptom_file, encodingutf-8-sig) dd pd.read_csv(disease_dept_file, encodingutf-8-sig) triples [] for _, row in ds.iterrows(): disease clean_entity_name(row[疾病]) symptom clean_entity_name(row[症状]) if disease and symptom: triples.append((Disease, disease, HAS_SYMPTOM, Symptom, symptom)) for _, row in dd.iterrows(): disease clean_entity_name(row[疾病]) dept clean_entity_name(row[科室]) if disease and dept: triples.append((Disease, disease, DEPARTMENT, Department, dept)) # 跨表去重保持首次出现顺序 seen set() unique_triples [] for t in triples: if t not in seen: seen.add(t) unique_triples.append(t) return unique_triples编码用 utf-8-sig 而不是 utf-8是因为很多从 Excel 另存出来的 CSV 带 BOM用 utf-8 读第一列会多一个不可见字符实体名匹配直接失败。clean_entity_name 是我做图谱项目必写的第一个函数它同时处理了末尾换行和全角空格。参数说明两个入参分别是疾病-症状表、疾病-科室表的路径返回三元组列表元素结构是起点类型起点名关系类型终点类型终点名。去重放在最后做因为同一个“疾病-症状”对可能出现在多张来源表里跨表去重必须统一处理。2.3 用py2neo批量写入Neo4j先建约束再按关系分组MERGE入库顺序比入库本身更重要。我一般先建唯一约束再用 UNWIND 批量写入。约束保证实体名唯一它同时也是查询索引问答阶段的匹配速度靠它兜底。from py2neo import Graph from collections import defaultdict graph Graph(bolt://localhost:7687, auth(neo4j, your_password)) # 先建唯一约束旧版Neo4j用ASSERT语法4.4用REQUIRE graph.run(CREATE CONSTRAINT disease_name IF NOT EXISTS FOR (n:Disease) REQUIRE n.name IS UNIQUE) graph.run(CREATE CONSTRAINT symptom_name IF NOT EXISTS FOR (n:Symptom) REQUIRE n.name IS UNIQUE) graph.run(CREATE CONSTRAINT dept_name IF NOT EXISTS FOR (n:Department) REQUIRE n.name IS UNIQUE) graph.run(CREATE CONSTRAINT drug_name IF NOT EXISTS FOR (n:Drug) REQUIRE n.name IS UNIQUE) graph.run(CREATE CONSTRAINT check_name IF NOT EXISTS FOR (n:Check) REQUIRE n.name IS UNIQUE) def import_triples(triples, batch_size500): # 按关系类型分组避免一个事务里混合多种关系模板 grouped defaultdict(list) for start_type, start_name, rel_type, end_type, end_name in triples: grouped[(start_type, rel_type, end_type)].append({ start: start_name, end: end_name }) for (start_type, rel_type, end_type), group in grouped.items(): for i in range(0, len(group), batch_size): rows group[i:i batch_size] cypher f UNWIND $rows AS row MERGE (s:{start_type} {{name: row.start}}) MERGE (e:{end_type} {{name: row.end}}) MERGE (s)-[:{rel_type}]-(e) graph.run(cypher, rowsrows) print(f关系 {rel_type} 已写入 {min(i batch_size, len(group))} / {len(group)} 条)三个 MERGE 不等于三个 CREATE。MERGE 在节点已存在时只做匹配不重复创建配合唯一约束能避免二次导入时报错。按关系类型分组是为了让每个事务里的 Cypher 模板一致UNWIND 一批 500 条既快又容易定位失败批次。类型名和关系名我直接拼进了 Cypher因为这些值来自代码里的固定常量不是用户输入。如果哪天数据表里混入了外部传入的关系名必须先做白名单校验再拼不要直接格式化进查询。batch_size 也是可调参数本地 Neo4j 可以调到 1000远程实例建议 200防止单事务过大把内存打满。2.4 本体建模的边界先少后多别让关系成为负担这一步其实就是知识图谱构建里最常说的本体建模定义实体类型和关系类型。医疗场景有个反直觉的经验——关系宁少勿多。常见做法是先覆盖“症状-疾病-科室-药品”四条主干边等问答跑通、确实回答不了一些问题时再按需补“并发症”“检查”边。我在早期项目里一口气加了七种关系结果每种问法都要重新设计 Cypher 模板维护成本成倍上涨。还有一个容易被忽略的查询边界尽量别用可变长度路径比如“-[:HAS_SYMPTOM*1..2]-”。在概念演示里它很漂亮但真实问答时它会把多条路径的答案混在一起系统很难向用户解释“这个答案是怎么推出来的”。医疗场景要的是可解释路径越短越清晰。3. 问答引擎意图识别、实体对齐与Cypher生成让机器听懂人话3.1 五类问题的意图识别先用规则模板跑通再谈模型自动问答的“自动”第一步是判断用户在问什么。常见问题类型有五类问症状、问科室、问药物、问病因、问预防。很多人上来就想用 BERT 分类模型但手里训练样本可能只有几百条效果反而不如规则模板稳定。import re INTENT_MAP { query_symptom: [什么症状, 有哪些症状, 症状表现, 会有什么表现], query_department: [挂什么科, 什么科室, 哪个科室, 应该看什么科], query_drug: [吃什么药, 用什么药, 药物, 怎么用药], query_cause: [病因, 什么原因, 怎么会得, 诱发因素], query_prevent: [怎么预防, 如何预防, 预防措施, 注意事项], } def classify_intent(question: str) - str: for intent, keywords in INTENT_MAP.items(): if any(kw in question for kw in keywords): return intent return query_general没有用正则表达式只做子串匹配原因是医疗问句的变体实在太多正则写起来又长又脆。每类意图维护一个关键词数组新增问法就加一行字符串代价最低。query_general 是兜底意图表示系统没听明白后面会走回退逻辑。这个函数是纯字符串匹配中文分词不参与跑得极快。3.2 实体抽取与别名对齐jieba加自定义词典命中“头疼头痛”问句进来之后要找出里面的疾病名、症状名。最常用的做法是 jieba 分词加自定义词典。医疗词有两个特点词长不稳定“神经内科”在默认词典里会被切成“神经/内科”经常会切错别名多“头疼”和“头痛”在知识图谱里可能是两个写法。import jieba # 自定义词典提高医疗实体分词稳定度 for word in [神经内科, 心血管内科, 高血压, 糖尿病, 脑梗死, 冠心病]: jieba.add_word(word) # 别名表口语/别名 - 图谱标准名 ALIAS_MAP { 头疼: 头痛, 血压高: 高血压, 血糖高: 糖尿病, 脑袋疼: 头痛, 胸口闷: 胸闷, } def extract_entities(question: str) - list: tokens jieba.lcut(question) hits [] for token in tokens: token ALIAS_MAP.get(token, token) # 先做别名归一 for entity_type, entity_names in ENTITY_NAMES.items(): if token in entity_names: hits.append((entity_type, token)) break return hitsENTITY_NAMES 是“实体类型 - 标准名集合”可以直接从图谱里读出来保持和库内完全一致。这里的关键是把别名映射放在分词之后、匹配标准名之前。顺序错了比如先匹配再归一结果就是“头疼”在标准名集合里查不到直接被丢弃。参数说明jieba.add_word 会提升该词在后续切分中的权重但它不是百分百强制极端情况下还是可能切错。更保险的做法是把自定义词典写进 jieba 的 user_dict 文件并把这个文件放进项目仓库里换环境时不用重新写一遍。3.3 相似度兜底分词没命中系统还能不能救回来分词毕竟有边界。用户说“血压偏高”图谱里只有“高血压”jieb 切出来是“血压”“偏高”实体抽取直接扑空。这种时候要有一个模糊匹配兜底常见做法是 difflib 的 SequenceMatcher。from difflib import SequenceMatcher def fuzzy_entity_lookup(name: str, entity_list: list, min_ratio: float 0.6): 在实体名列表里找最相似的候选低于min_ratio则不返回 best_match, best_ratio None, 0.0 for entity in entity_list: ratio SequenceMatcher(None, name, entity).ratio() if ratio best_ratio: best_match, best_ratio entity, ratio return best_match if best_ratio min_ratio else Nonemin_ratio 默认 0.6这个值是血泪经验调出来的。设太低比如 0.4“小儿肺炎”会匹配到“肺炎”给出的答案不准设太高比如 0.85别名基本救不回来。0.6 是在实体长度普遍为 2 到 6 个汉字时的折中值。要注意“血压偏高”这句由两个 token 组成fuzzy_entity_lookup 只能处理单个 token所以我会先把“血压偏高”整体拼回去做一次查询再考虑 token 级匹配。顺序是完整句子先试一次分词结果再试一次最后才是单 token 模糊。这一层兜底的意义是让系统回答“不确定”时有个合理的近似答案而不是直接报错。医疗场景宁可给近似条目并提示“为您找到相近的是”也不该空手而归。3.4 生成Cypher查询模板拼接与参数化一个都不能少意图和实体都齐了下一步是把它们翻译成图查询。这里的模板要按意图和实体类型同时匹配比如“高血压有什么症状”和“头痛挂什么科”两个问题的起点实体类型、关系边都不一样。def build_cypher(intent: str, entity_type: str, entity_name: str): if intent query_symptom: cypher ( fMATCH (d:{entity_type} {{name: $name}}) f-[:HAS_SYMPTOM]-(s:Symptom) RETURN s.name AS answer ) elif intent query_department: cypher ( fMATCH (d:{entity_type} {{name: $name}}) f-[:DEPARTMENT]-(dept:Department) RETURN dept.name AS answer ) elif intent query_drug: cypher ( fMATCH (d:{entity_type} {{name: $name}}) f-[:TREATS]-(drug:Drug) RETURN drug.name AS answer ) else: return None, None return cypher, {name: entity_name} def query_answer(cypher: str, params: dict): if not cypher: return [抱歉这个问题我暂时回答不了] records graph.run(cypher, parametersparams).data() return [record[answer] for record in records] or [知识图谱里暂未查到相关记录]实体名通过 $name 参数传入而不是直接拼进 Cypher。参数化在这里有两个作用一是避免实体名里带单引号、双引号时把查询语法破坏掉二是防止注入虽然问答场景风险有限但工程上保留这个习惯没有坏处。entity_type 依然走白名单拼装来源是 2.2 节三元组里的固定类型不受用户输入影响。到这里一个完整问答请求已经能够闭环问句进来识别意图抽取实体拼 Cypher查库返回答案列表。完整系统通常会把 query_answer 包成一个 Flask 接口前端 POST 一句 json 就能拿到答案。接口本身是壳问答核心全在前面几个函数里。4. 五个高频踩坑记录现象、原因、解决4.1 “头疼”查得到“头痛”查不到现象用户输入“头疼”返回“知识图谱里暂未查到相关记录”输入“头痛”却有完整答案。原因图谱标准名只收了“头痛”问句里的“头疼”没进别名表实体抽取阶段就丢了。解决先建别名表把口语说法、方言、缩写统一映射到标准名映射发生在实体匹配之前。别只做两三个别名实际收集时用搜索词的联想词去补表比如“血压高”“血压偏高”“HBP”都要收敛到“高血压”。4.2 图谱越大越慢约束没建、索引没有现象几百条数据时秒回到几千条后查询明显变慢有的 Cypher 要跑几秒钟。原因导入时没建唯一约束也没建属性索引Neo4j 只能按 label 扫描所有节点做 name 匹配。解决导入前执行 2.3 节那几条 CREATE CONSTRAINT 语句它同时创建索引。索引要建在查询的匹配字段上也就是 name 属性。这个问题在数据量过万后会更明显项目初期就要建别等卡了再补。4.3 换台电脑跑不起来py2neo与Neo4j版本不匹配现象代码在自己机器上跑得好好的换台电脑装好 Neo4j启动 Python 脚本直接报认证失败或者提示 Graph.run 方法不存在。原因py2neo 4.x 和 5.x 的部分 API 不兼容Neo4j 5.x 对旧版驱动支持也有限两个版本对不上就会出这类玄学问题。解决项目根目录放 requirements.txt锁死 py2neo 与 neo4j 驱动版本同时把 Neo4j 版本写进 README。我一般也会顺手写一个连接测试脚本启动后先跑一句 RETURN 1确认环境通了再跑导入和问答。4.4 问“病因”返回“科室”意图模板太粗现象用户问“高血压是什么原因造成的”系统返回“心血管内科”。原因INTENT_MAP 里 query_cause 只有“病因”“什么原因”这句恰好没触发整体落进了 query_department 的关键词“什么”。解决不要只扩充关键词要建一个意图自检集每个意图至少准备 15 个变体问句跑一遍自动测试看哪些问句被分错了类。规则模板的优势就是修起来快但也要有测试集兜住回归否则改一个词可能带崩另一类问题。4.5 导入时报ConstraintValidationFailed重复实体没有先合并现象第二次运行导入脚本图库报“Node already exists with label Disease and property name”。原因唯一约束建立后CREATE 语句试图新建重名节点会直接抛错。解决所有节点写入统一用 MERGE它存在就匹配、不存在才创建。另外 2.2 节的三元组去重也不是可有可无同一对“疾病-症状”可能在多张来源表里重复出现导入前去重能减少一半这类报错。5. 进阶从能问答到答得可靠先过评估、归一化、链指三关5.1 建一份测试问句集算准召评价问答系统不能靠拍脑袋。常见做法是准备 100 条测试问句每条标注标准实体、标准答案然后跑一个简单评估脚本def evaluate(test_set, answer_func): total_recall total_precision 0.0 n 0 for case in test_set: answers set(answer_func(case[question])) gold set(case[gold_answers]) if not gold: continue total_recall len(answers gold) / len(gold) total_precision (len(answers gold) / len(answers)) if answers else 0 n 1 return total_recall / n, total_precision / n召回率衡量“该给的答案漏没漏”精确率衡量“给的答案准不准”。医疗问答里我一般更看重召回答不全比答错好补救但精确率低于 0.7 时就要检查实体对齐和意图模板了。5.2 术语归一化把“血压高”“HBP”都收敛到“高血压”别名表做到后期会成为一张标准术语表这是医疗问答从“能用”到“好用”的分水岭。做法是给每个标准实体维护一个别名数组在问句预处理阶段统一替换。这一步值得提前投入因为后续做实体链指、接口拓展都依赖一套稳定的标准名。我见过团队在这里偷懒结果换个数据源别名表推倒重来代价远大于一开始就建词表。5.3 实体链指处理“它”“这个病”的指代连续对话里用户会说“它”“这个病”实体抽取抓不到。简单做法是保存上一轮命中的实体当前问句出现代词时替换回去def resolve_coreference(question, last_entity): if not last_entity: return question return question.replace(它, last_entity).replace(这个病, last_entity)只适用于单轮指代。多轮对话的指代消解要上模型医疗场景里结论必须保守指代不确定时宁可多问一句“您说的是哪个病”也不要自作聪明。我最早搭这个系统时数据表里有一行疾病名后面带了个换行符结果实体匹配时好时坏查了一晚上才发现是清洗环节漏了换行符。后来我把所有实体名统一走 clean_entity_name顺序也固定为先洗数据、再建约束、最后写问答逻辑。这个顺序值得你记下来。希望帮到你。本文还有配套的精品资源点击获取