简介基于Python的医学知识图谱问答系统设计源码是一套面向人工智能学习者、医学研究人员及开发者的完整医学信息处理项目覆盖医学知识图谱构建、问句分类、意图解析与答案检索全流程可用于课程设计、毕业设计或科研探索。核心模块涵盖build_medicalgraph.py、question_classifier.py与answer_search.py等配合prepare_data与data目录下的疾病、症状、药品、检查等医学文本数据可系统理解医学术语处理、图谱存储与问答路由实现。包体共31个文件包含8个Python源文件、9个文本数据、9张PNG架构示意图、3个pyc编译文件、1个JSON配置及1个PPTX演示文稿压缩包约49.19MB目录结构清晰。图片与演示文稿便于快速掌握系统整体流程文本数据可直接作为知识库扩展基础源码则可作为二次开发与功能优化的起点。目前已有176人浏览学习适合希望从代码层面掌握医学知识图谱问答系统设计思路并在此基础上进行项目复现或深入扩展的中高级Python开发者。1. 医学知识图谱问答系统为什么直接给答案比搜十页网页难十倍在门诊病历系统里敲一个药名你想要的往往不是十页搜索结果而是“这个药与头孢类抗生素同服会增加出血风险”这样直接给到眼前的答案。基于Python的医学知识图谱问答系统做的就是这件事把医学知识拆成疾病、症状、药物、检查等实体以及它们之间的关系统一存进图数据库再把用户的自然语言问句翻译成图查询最终拼出一句人话。这个方向对从业者最现实的价值在于它不像深度问答那样吃GPU和标注数据用Python加Neo4j就能在两三周内搭出一个可演示、可迭代的版本。适合正在做毕业论文、医院信息化预研以及想从爬虫和CRUD里往上游走一步的Python工程师。2. 先把架构立住数据层、解析层与查询合成层怎么分工2.1 三层架构的边界划分哪一层该管什么一个能长期维护的医学问答系统源码不该是单文件堆出来的。常见做法是拆成三层每层只解决一个问题数据层只负责和Neo4j打交道。实体写入、关系写入、索引创建、查询执行全部收口在图数据库封装类里外部代码不直接拼Cypher。解析层负责把自然语言变成结构化指令包括医学词典加载、分词、实体抽取、问句意图识别。这一层不碰数据库。查询合成层拿到意图和实体之后决定用哪条Cypher模板参数怎么传结果怎么拼成中文回答。这一层是系统的翻译官也是最容易写出烂代码的地方。我见过不少半途翻车的项目问题都出在解析层和查询合成层互相纠缠分词逻辑里写SQL查询函数里又做字符串正则。一旦纠缠后续加一个意图就要动三处代码改完必出回归bug。所以源码组织上哪怕项目再小也建议在包结构上把边界划清楚。2.2 医学实体与关系建模决定问答上限的设计表医学知识图谱的问答能力上限在设计表时就定死了。实体类型少了问句里的词抽出来无处安放关系方向反了查询模板怎么写都别扭。第一版不要贪多先把最能出效果的六类实体和五类关系搭起来。实体类型实体类型中文含义常见属性Disease疾病name, alias, icd_codeSymptom症状name, alias, descriptionDrug药物name, generic_name, contraindicationsDepartment科室nameExamination检查项name, normal_rangeFood食物name关系类型关系方向典型问句HAS_SYMPTOMDisease - Symptom高血压有什么症状TREATSDrug - Disease高血压吃什么药BELONGS_TODisease - Department高血压挂什么科REQUIRESDisease - Examination糖尿病要做什么检查FORBIDDEN_WITHDrug - Drug阿司匹林不能和什么一起吃这个模型很朴素但它覆盖了医学问答里最高频的几类问题症状查询、治疗推荐、就诊科室、检查建议、药物禁忌。如果你要加“食物相克”这类问题只需要加一个Food节点和一条INTERACTS_WITH关系解析层加一个意图模板查询层加一条Cypher。后面第4章我会写这套新增流程具体怎么落地。2.3 把Neo4j接进Python连接配置与索引初始化数据层建议直接用Neo4j官方Python驱动而不是py2neo。py2neo在批量导入和事务控制上便利但维护节奏慢官方驱动API更稳。连接封装类要至少做到统一入口、统一返回结构、关闭时释放连接。# kg/neo4j_client.py from neo4j import GraphDatabase class MedicalGraphDB: def __init__(self, uri, user, password): self.driver GraphDatabase.driver(uri, auth(user, password)) def close(self): self.driver.close() def execute(self, cypher, **params): with self.driver.session() as session: result session.run(cypher, **params) return [record.data() for record in result]这段代码的关键在execute方法所有会话管理收在一个函数里返回值统一成列表每条记录是字典。这样上层代码不需要关心会话关闭问题也不会出现忘记关连接导致连接池耗尽的现象。uri填bolt://localhost:7687user和password按本地Neo4j实例配置driver对象全局只创建一次不要每次请求都new一个。建索引和唯一约束要放在数据导入之前。否则你后面跑MERGE时会发现同一个“高血压”被插成了十个节点干净的数据被搅成一团。CREATE CONSTRAINT disease_name IF NOT EXISTS FOR (d:Disease) REQUIRE d.name IS UNIQUE; CREATE CONSTRAINT symptom_name IF NOT EXISTS FOR (s:Symptom) REQUIRE s.name IS UNIQUE; CREATE CONSTRAINT drug_name IF NOT EXISTS FOR (d:Drug) REQUIRE d.name IS UNIQUE;如果你用的是Neo4j 4.x约束语法是CREATE CONSTRAINT ON (d:Disease) ASSERT d.name IS UNIQUE5.x才支持REQUIRE写法。先确认版本再执行不然报错会绕一大圈。2.4 目录结构与依赖一个能run的最小骨架源码设计的落地形态是目录结构。我习惯按“入口、知识图谱、自然语言处理、数据、测试”五个方向组织第一版能跑通的最小骨架长这样medical_qa/ ├── app.py # Web入口Flask/FastAPI都可以 ├── kg/ │ ├── __init__.py │ ├── neo4j_client.py # 图数据库连接封装 │ ├── schema.py # 实体类型和关系类型常量 │ ├── importer.py # 数据入库脚本 │ └── queries.py # 答案查询函数 ├── nlp/ │ ├── __init__.py │ ├── segment.py # 分词与实体识别 │ ├── intent.py # 意图模板匹配 │ └── dict.txt # 医学自定义词典 ├── data/ │ ├── diseases.json │ ├── drugs.json │ └── relations.json ├── tests/ │ └── test_qa.py └── requirements.txt依赖项不需要多越少越不容易在别人机器上跑不起来。requirements.txt里至少是neo4j、jieba、Flask或者fastapi。如果走Web接口FastAPI对请求参数校验更省心如果只是本地演示Flask足够。不要去装bert-serving这类重型依赖第一版规则引擎就够了后面要升级再单独开分支加。3. 从零构建医学知识图谱数据清洗到批量入库的落地脚本3.1 数据准备从公开词典与结构化表格里抽实体关系图谱的数据来源不复杂常见的是公开医学词典、药品说明书结构化的字段、以及已有的症状-疾病对照表。对这些资料做抽取时最怕的是把脏数据直接灌进图库。比如同一份数据里“高血压”和“高血压病”混着出现入库后查询阶段就会因为实体名对不上而大面积查空。所以入库前先做一次归一化把全角半角统一、去掉括号备注、把疾病别称映射到标准名。# data/preprocess.py import re def normalize_entity(name): # 统一全角转半角、去掉多余空格与括号备注 result name.replace(, ,).replace(。, .).strip() result re.sub(r.*?, , result) result re.sub(r\(.*?\), , result) return result这个函数是数据清洗的第一道闸门。我一般会在生成data/relations.json之前对每一对实体关系都跑一次。注意括号备注比如“高血压原发性”和“原发性高血压”其实是同一实体但在你的归一化逻辑里可能还是两条。要做彻底建议单独维护一张别名映射表在归一化之后再替换一次。3.2 用MERGE批量导入幂等入库避免重复节点导入脚本用MERGE关键字而不是CREATE这是防止重复节点的关键。MERGE会先按实体名匹配存在则跳过不存在才创建。关系也用MERGE这样脚本重复执行不会把图谱撑出脏数据。# kg/importer.py import json from kg.neo4j_client import MedicalGraphDB def import_disease_symptom(db, rows): cypher MERGE (d:Disease {name: $disease_name}) MERGE (s:Symptom {name: $symptom_name}) MERGE (d)-[:HAS_SYMPTOM]-(s) count 0 for row in rows: db.execute(cypher, disease_namerow[disease], symptom_namerow[symptom]) count 1 return count if __name__ __main__: db MedicalGraphDB(bolt://localhost:7687, neo4j, your_password) with open(../data/relations.json, encodingutf-8) as f: relations json.load(f) # 只导入症状关系药物关系单独走另一个函数 n import_disease_symptom(db, relations[disease_symptom]) print(finserted/merged {n} disease-symptom relations) db.close()这里的cypher是参数化查询$disease_name和$symptom_name会由驱动安全替换不用做字符串拼接既避免了中文引号转义问题也防住了查询注入。如果你按字符替换的方式用f-string拼Cypher遇到实体名里含英文单引号会直接跑崩这是新手最容易踩的坑。rows是从JSON解析出来的列表每个元素是一个字典字段要和JSON文件里的key对得上。3.3 导入后的验证节点数、关系数、样本查询数据导完不是结束要立刻在Neo4j里验证。验证分三步总量、样本、边界。-- 统计节点数按实体类型 MATCH (d:Disease) RETURN count(d) AS disease_count; -- 统计关系数 MATCH ()-[r:HAS_SYMPTOM]-() RETURN count(r) AS relation_count; -- 抽查一种疾病看看关系是否完整 MATCH (d:Disease {name: 高血压})-[:HAS_SYMPTOM]-(s:Symptom) RETURN s.name LIMIT 10;我每次导完都会跑一遍这三条确认数字在合理区间。比如disease_symptom有500条关系但抽查高血压时只返回2个症状那大概率是数据源里高血压的症状列被清洗丢了。还有一条高频检查查一种完全没有关系的疾病是否存在比如某个生僻病名只有节点没有关系。这种情况在问答时表现就是实体识别成功、但Cypher查不到任何结果最后系统只能回一句“答不上来”。4. 问答链路实现问句分词、意图识别到Cypher查询生成4.1 加载自定义词典让“高血压”不被切碎到了问答链路第一个坑来自分词。jieba默认词表里“高血压”经常被切成“高”和“血压”实体识别直接失效。解决方式是自定义词典把这些医学专有名词作为一个词强行保留。# nlp/dict.txt 高血压 100 n 高血压病 100 n 阿司匹林 100 n 头孢类药物 100 n 糖尿病 100 n 冠心病 100 n加载方式如下# nlp/segment.py import jieba jieba.load_userdict(nlp/dict.txt) def segment(question): words jieba.lcut(question) # 过滤纯停用词与单字 useful [w for w in words if len(w.strip()) 1] return usefuldict.txt里的每一行格式是“词 词频 词性”词频100起步就够不用去抄别人给的数值。加载后一定要打印一次分词结果做回归“高血压有什么症状”应该切成[“高血压”, “什么”, “症状”]如果还是碎的检查dict.txt文件路径是不是相对路径导致没读到。4.2 意图识别与槽位抽取用规则模板扛住第一版医学问句的句式相对封闭规则模板在第一版完全够用。意图识别用关键词匹配槽位抽取用正则抓取两个模块分开写。# nlp/intent.py INTENT_KEYWORDS { symptom: [什么症状, 有哪些症状, 临床表现, 症状是什么], treatment: [怎么治, 如何治疗, 吃什么药, 用什么药, 治疗方案], department: [挂什么科, 去哪个科室, 哪个科室], contraindication: [禁忌, 不能吃, 不能和, 禁用], } def detect_intent(question): for intent, keywords in INTENT_KEYWORDS.items(): for kw in keywords: if kw in question: return intent return fallback这个函数输出的是意图名后面查询合成层用意图名选择Cypher模板。注意关键词有顺序问题“高血压不能吃什么药”里同时有“吃什么药”和“不能吃”先匹配到treatment就答错了。这里要用优先级把禁忌类的关键词放在最前面遍历或者检测到“不能”时把treatment降级为contraindication。实体抽取我用最长匹配而不是直接把分词结果当实体。# nlp/segment.py def extract_entity(question, entity_dict): # entity_dict 来自图数据库中的实体名和类型映射 matched [] for entity, label in entity_dict.items(): if entity in question and entity not in matched: matched.append((entity, label)) if not matched: return None, None # 按实体名长度从长到短排序取最长那个 matched.sort(keylambda x: len(x[0]), reverseTrue) return matched[0]这里用最长匹配的意义在于问句里可能同时出现“高血压”和“血压”两个候选如果不排序就随机命中可能抓到“血压”这个短实体而丢掉疾病类型。实体名从哪来从Neo4j里一次性加载到内存构建 entity_dict。量级在几千条时没问题如果后面膨胀到百万实体就要改成词典树或者引入ES了。4.3 查询转换与答案组装把查询结果变成人话意图和实体拿到后查询合成层要做三件事选Cypher模板、填参数、格式化答案。模板映射建议做成字典意图名对应一个Cypher模板函数。# kg/queries.py def build_cypher(intent, entity_label): templates { symptom: ( MATCH (: entity_label {name: $entity}) -[:HAS_SYMPTOM]-(s:Symptom) RETURN s.name AS name LIMIT 8 ), treatment: ( MATCH (d:Drug)-[:TREATS]-(:Disease {name: $entity}) RETURN d.name AS name LIMIT 8 ), department: ( MATCH (: entity_label {name: $entity}) -[:BELONGS_TO]-(dep:Department) RETURN dep.name AS name LIMIT 3 ), } return templates.get(intent)这里的entity_label是实体类型比如Disease、Drug、Symptom。为什么模板里要动态拼接实体类型因为同样问“有什么症状”主体可以是疾病也可以是药物副作用。如果不区分Cypher匹配不到结果。拼接entity_label是程序内部生成的不来自用户输入所以不存在注入风险真正来自用户的$entity必须走参数。答案组装阶段要把查询结果转成自然语言。比如治疗问句查出来的是一串药名拼成“高血压可考虑使用的药物有硝苯地平、卡托普利”比直接返回JSON列表直观得多。def format_answer(intent, records): if not records: return 抱歉暂时没有找到对应信息。 names [r[name] for r in records] if intent treatment: return 可考虑使用的药物有 、.join(names[:5]) if intent symptom: return 常见症状包括 、.join(names[:5]) return .join(names[:5])4.4 兜底策略查不到答案时至少别崩问答系统第一版最容易出的问题是用户问了一句模板之外的话系统直接抛异常或者返回空白。兜底策略要覆盖两种场景意图识别到fallback、实体抽取为None。这两个场景下不要硬查数据库直接返回预设话术并且把问句记录到日志文件里。def answer_question(question, entity_dict, db): intent detect_intent(question) entity, label extract_entity(question, entity_dict) if intent fallback or entity is None: log_unhandled(question, intent) return 我还不能完全理解这个问题试试问“高血压有什么症状”这类具体问题。 cypher build_cypher(intent, label) if cypher is None: return 这个问题暂时没有可用的查询逻辑。 records db.execute(cypher, entityentity) return format_answer(intent, records)log_unhandled是把未命中的问题写到本地CSV文件字段至少包括问句、时间、识别出的意图。这个日志就是你后面迭代词典和模板的数据来源比你自己拍脑袋想用户会问什么靠谱得多。兜底话术不要用“系统错误”这种冷冰冰的词给个示例问句反而能让用户重新描述。5. 医学知识图谱问答系统避坑指南五个踩过才知道的问题5.1 “高血压”被切成“高/血压”自定义词典没生效现象实体识别时“高血压”被拆成“高”和“血压”匹配entity_dict时只能命中“血压”实体类型错成Symptom查询结果全空。原因jieba词典加载失败或路径不对。最常见的是在IDE里运行时工作目录和代码不在同一层级load_userdict用的相对路径找不到文件但分词照常执行只是没加载你的词表。解决在load_userdict前打印当前工作目录改成绝对路径或用os.path.join定位到项目根目录。加载后立刻对“高血压有什么症状”做一次分词断言把这个断言写进测试用例tests/test_qa.py里后续任何人动了词表都能发现。5.2 MERGE出重复节点唯一约束建晚了现象导入脚本跑了两遍Neo4j里“高血压”出现10个节点每个节点只有一半关系。问答时Cypher用name匹配随机命中其中一个结果不稳定。原因MERGE的语义是“若图中有匹配节点则复用否则创建”但匹配的前提是对应属性上有唯一约束。没有约束时MERGE会退化成按节点是否存在判断同一实体名还是可能被重复创建。解决在导入前先执行CREATE CONSTRAINT并且用Cypher验证约束确实建立成功CALL db.constraints。如果已经产生了重复节点先手工去重再建约束不要指望MERGE能自动合并历史脏数据。5.3 问“血压高”查不到“高血压”别名与归一化没做现象图谱里存的标准名是“高血压”用户问“血压高怎么办”实体抽取返回None系统答不上来。原因实体词典里只有标准名没有覆盖口语化表达。医疗场景这个词表几乎是必须维护的患者口语和病历书面语差异极大。解决给实体节点加alias属性把“血压高”“高血压病”等别名一并入库。查询时把实体匹配从name扩展到aliasMATCH (n:Disease) WHERE n.name $entity OR n.alias CONTAINS $entity。成本最低的维护路径是把用户历史的“答不上来”日志定期过一遍把高频新词补充进词典和alias。5.4 Cypher查询慢得离谱索引没建现象图谱数据量到两万节点后一次问答查询耗时3秒以上Neo4j日志出现全库扫描告警。原因Cypher里按name属性过滤时没有可用的索引。Neo4j不建索引也能跑但它是逐个节点扫描属性匹配数据量一大就血崩。解决对高频查询的实体name和alias建立索引。唯一约束本身也是一种索引所以前面建过约束的属性不需要再建普通索引。验证查询是否走索引用EXPLAIN前缀看执行计划看到NodeByLabelScan基本就是没走索引。5.5 关系方向存反了查出来永远是空现象“高血压吃什么药”查不到结果但直接在Neo4j Browser里手工查能看到药物节点和疾病节点都存在。原因导入时把TREATS关系建成了(Disease)-[:TREATS]-(Drug)方向反了模板MATCH (d:Drug)-[:TREATS]-(:Disease)自然匹配不到。解决在schema.py里把每个关系的方向写成注释常量用代码约束自己。更稳妥的做法是导入时统一按“主动词流向”定义方向药物作用于疾病、疾病表现为症状、疾病属于科室、疾病需要检查。查询时遇到查不到先检查关系方向而不是改代码。养成习惯先在Browser里跑一条不带方向的MATCH ()-[r:TREATS]-() RETURN r LIMIT 5看看关系的起点和终点标签到底是什么。6. 让问答系统有据可依回归测试与迭代策略6.1 用一组典型问句做回归冒烟没有测试的问答系统改一句模板就崩一个分支。我习惯在tests/test_qa.py里维护一个冒烟用例表每个用例包含问句、预期意图、预期实体、预期结果非空四个断言字段。# tests/test_qa.py CASES [ {question: 高血压有什么症状, intent: symptom, entity: 高血压}, {question: 高血压吃什么药, intent: treatment, entity: 高血压}, {question: 阿司匹林不能和什么一起吃, intent: contraindication, entity: 阿司匹林}, ] def run_smoke_test(entity_dict, db): passed 0 for case in CASES: intent detect_intent(case[question]) entity, _ extract_entity(case[question], entity_dict) if intent case[intent] and entity case[entity]: records db.execute(build_cypher(intent, Disease), entityentity) if records: passed 1 else: print(fFAILED: {case[question]}) return passed这段测试的价值在改代码时体现你加了新意图或者改了关键词优先级跑一遍就知道有没有影响到旧问句。我吃过的亏是改了一次“不能”的优先级把“高血压不能吃什么”和“吃什么药”的区分逻辑弄反测试随手一跑就暴露了。6.2 把答错的问题变成数据资产问答系统不是写完就完。上线第一天收集到的“答不上来”问句第二天就该进词典和模板。我习惯每周做一次日志复盘把log_unhandled输出的CSV里频次最高的30条问句逐个分类——缺实体、缺意图、缺别名、缺关系。缺实体就加词典词条缺意图就加关键词缺关系就补数据导入脚本。这个闭环比调整模型参数带来的提升更直接尤其是面向垂直领域时问法就那几十种规则引擎加完善的词典效果能打到可用线。另外如果要向语义方向升级就沿着“意图模板→同义句扩展→向量召回”的路径走。先收集用户的真实问句标注意图再训练一个轻量分类模型不要一开始就上BERT。趁着数据量小规则模板加人工维护词典是性价比最高的阶段。我现在的习惯是每次改完问答链路先跑冒烟测试再看一轮未命中日志确认没有新词汇变多才会合代码。这套流程看起来土但让这个项目从“能跑”走到了“敢用”。希望帮到你。本文还有配套的精品资源点击获取