简介这是一套基于Python实现的医疗知识图谱知识问答系统课程设计包面向医疗信息化、知识图谱及NLP问答方向的初学者与课程设计者。资源围绕“构建医疗知识图谱—实现对话系统”主线包含设计报告、完整项目源码与医疗数据并配套项目截图源码按知识图谱构建与规则问答两条逻辑展开覆盖实体关系构建、问句分类、答案查询等关键环节。该对话系统无需训练、响应迅速但灵活度有限适合入门理解规则式问答作者也点明了后续结合深度学习模型的扩展方向。包内共58个文件以py源码、txt字典、png截图、json数据、docx设计报告为主整体约20.13MB结构清晰便于按模块逐步学习。已有2569人学习使用适合需要快速上手医疗知识图谱问答项目的读者。1. 医疗知识图谱问答系统在解决什么先看懂它值不值得做基于Python实现的医疗知识图谱知识问答系统这个名字技术栈和场景都说得很直白Python 做整个流水线的胶水知识图谱负责存放医疗实体之间的关系问答系统负责把用户问句转换成一类可以查询的 Cypher最后返回人能看懂的答案。这类系统最常见的落地场景是导诊咨询、药品说明书问答、辅助医生做诊疗关联检索。它适合三类人要做毕业设计的学生想给院内系统加一个对话入口的工程师以及想验证图谱比关键词搜索强在哪的产品经理。我按自己实践过的路线讲关系建模、数据导入、问句解析、Flask 接口最后落在评测和避坑上跟着走能跑出一个可用版本。2. 把医疗数据建成图谱实体关系建模与 Neo4j 导入2.1 实体与关系怎么设计才不会在问答阶段返工医疗知识图谱最怕的不是没数据而是关系设计得含糊。常见的做法是先定五类实体疾病、症状、药物、检查项目、科室关系用动词明确语义比如治疗表现禁忌确诊检查就诊科室。这比把所有内容都塞进一个相关关系要可靠得多问答阶段生成 Cypher 时才能用上明确的关系方向。实体类型示例常见别名疾病 Disease高血压、2型糖尿病高血压病、高血压症症状 Symptom头晕、心悸眩晕、心慌药物 Drug硝苯地平、阿司匹林拜新同、阿司匹林肠溶片检查 Check血压测量、糖化血红蛋白测血压、糖化科室 Department心内科、内分泌科心血管内科、内分泌门诊关系设计有个经验每条关系都要求从疾病出发有明确语义。比如(高血压)-[:治疗]-(硝苯地平)和(硝苯地平)-[:用于治疗]-(高血压)在物理上都是同一件事但方向一旦不统一问答模板就要写两套维护成本翻倍。我一般统一约定成主语是疾病、谓语是动作、宾语是结果的主动方向并且写进接口文档后续所有查询模板只认这个方向。另一点是别急着建大而全的 Schema。先拿 20 条真实问句反推需要哪些关系比如问高血压挂什么科你就知道必须有就诊科室问糖尿病要做什么检查就得有确诊检查。反推比拍脑袋建模可靠后面改 Schema 意味着重新导数据代价很高。2.2 用 Python 抽取实体关系从半结构化文本到三元组医疗数据来源很多药品说明书、临床指南、公开的医学百科都能用。爬虫拿到手之后能直接落库的其实是半结构化文本比如疾病高血压/疾病可用药物硝苯地平/药物治疗。这种带标签的行文是最容易解析的格式我把抽取逻辑做成一个函数import re import jieba.posseg as pseg # 自定义词典每行格式: 词 词频 词性比如 硝苯地平 100 drug jieba.load_userdict(medical_dict.txt) DISEASE_PAT re.compile(r疾病(.*?)/疾病) DRUG_PAT re.compile(r药物(.*?)/药物) SYMPTOM_PAT re.compile(r症状(.*?)/症状) def parse_half_structured(line): 从一行带标签文本里抽出 (实体1, 关系, 实体2) 三元组。 disease DISEASE_PAT.search(line) drug DRUG_PAT.search(line) symptom SYMPTOM_PAT.search(line) triples [] if disease and drug: triples.append((disease.group(1), 治疗, drug.group(1))) if disease and symptom: triples.append((disease.group(1), 表现, symptom.group(1))) return triples这段逻辑并不复杂用三个正则把标签里的实体抠出来再根据标签组合关系。关键在load_userdict那行医疗领域里硝苯地平拜新同这类词不在通用分词词典里不加载自定义词典的话后续槽位填充会把这些词切得七零八落。参数100是词频建议统一给一个较大的数让 jieba 优先按整词切分词性部分用drug、dis这类自定义标签和通用词性区分开。2.3 批量导入 Neo4jpy2neo 批量事务还是 LOAD CSV数据量在十万三元组以内用 py2neo 写 Python 脚本导入最顺手如果到了百万级LOAD CSV 加USING PERIODIC COMMIT才是更稳的方案。先看 py2neo 的批量写法from py2neo import Graph, Node, Relationship graph Graph(bolt://localhost:7687, auth(neo4j, password)) def import_triples(triples, batch_size500): 批量写入三元组先按 name 查重再建关系避免重复节点。 tx graph.begin() count 0 for subj, rel, obj in triples: s graph.nodes.match(Entity, namesubj).first() if s is None: s Node(Entity, namesubj) tx.create(s) o graph.nodes.match(Entity, nameobj).first() if o is None: o Node(Entity, nameobj) tx.create(o) tx.create(Relationship(s, rel, o)) count 1 if count % batch_size 0: tx.commit() tx graph.begin() tx.commit()batch_size500是控制事务大小的关键参数。一次事务塞太多操作Neo4j 的堆内存会飙高本地开发时常能看到报错500 条一提交既能保证速度又不至于把内存打爆。每次match再创建是为了防重否则同一条数据跑两遍图谱里就会出现两个叫高血压的节点后面的问答查询会返回重复结果而且很难排查。十万以上规模时我更推荐先导成 CSV再用 Cypher 的 LOAD CSV 导入。把三元组写进 CSV 后执行USING PERIODIC COMMIT 1000 LOAD CSV WITH HEADERS FROM file:///medical_triples.csv AS row MATCH (s:Entity {name: row.subject}) MERGE (o:Entity {name: row.object}) MERGE (s)-[:REL {type: row.relation}]-(o)注意一点大规模导入前一定先给Entity(name)建唯一约束否则 MERGE 的查重会退化成全表扫描速度慢到无法接受。建约束的命令是CREATE CONSTRAINT FOR (e:Entity) REQUIRE e.name IS UNIQUE;。3. 问答链路问句分类、槽位填充与 Cypher 生成3.1 先给问句分对类规则优先还是模型优先问答的质量一半取决于意图分类。医疗问答有个特殊性问法相对固定翻来覆去就是有什么症状吃什么药挂什么科不能吃什么这几种模式。用正则规则先顶住高频问句性价比极高而且可解释、可调试BERT 分类模型放到规则覆盖不了的模糊问句上兜底才划算。INTENT_RULES [ (r.*(能|可以|是否).*(吃|服用|使用).*(治疗).*, drug_usage), (r.*什么.*(症状|表现).*, symptom), (r.*(禁忌|不能吃|不宜|忌口).*, contraindication), (r.*(挂|去).*(科|门诊).*, department), ] def classify_intent(question): 返回问句意图兜底为 fallback。 for pattern, intent in INTENT_RULES: if re.search(pattern, question): return intent return fallback这组规则里顺序有讲究先匹配能不能吃这类组合问法再匹配单纯的症状问法。因为高血压能吃什么药里也有什么两个字如果symptom规则排在前面就会被误判成症状查询。规则的直觉很简单但维护者必须清楚每条正则覆盖的问法集合我习惯在每个意图后面直接写两三条真实问句做注释避免后来的人改错顺序。3.2 槽位填充用 jieba 和自定义词典识别医疗实体意图确定之后再抽槽位抽的是问句里的医疗实体比如高血压“硝苯地平”头晕。这一步我用 jieba 的词性标注和自定义词典配合做不用引入重型模型import jieba.posseg as pseg MEDICAL_DICT { 高血压: dis, 糖尿病: dis, 硝苯地平: drug, 阿司匹林: drug, 头晕: sym, 心悸: sym, } def extract_slots(question): 抽医疗实体返回 {disease: [], drug: [], symptom: []}。 slots {disease: [], drug: [], symptom: []} for word, flag in pseg.cut(question): if word in MEDICAL_DICT: slots[flag].append(word) elif flag nt: # 机构名弱化成疾病候选 slots[disease].append(word) return slots这里有个细节MEDICAL_DICT的映射关系是词 → 实体类型而pseg.cut返回的 flag 恰好就是字典里的值所以slots[flag]能直接命中。如果你用的词典里词性命名不一致比如用了disease而字典键是dis槽位就会丢这也是一个特别隐蔽的坑。实体别名的问题放到第 5 章详细讲这里先剧透一句槽位填充拿到的是用户原词比如高血压病但图谱里存的是高血压这一步必须先做归一化再拿去拼 Cypher否则查不到。3.3 从槽位到 Cypher五种意图的模板映射意图和槽位都有了下一步就是拼 Cypher。常见做法是维护一张意图到查询模板的表模板里留出槽位占位符用format填进去意图用户问法生成的 Cypher 骨架symptom高血压有什么症状MATCH (d:Entity {name:高血压})-[:表现]-(s:Entity) RETURN s.nametreat高血压怎么治疗MATCH (d:Entity {name:高血压})-[:治疗]-(m:Entity) RETURN m.namecontraindication高血压不能吃什么MATCH (d:Entity {name:高血压})-[:禁忌]-(m:Entity) RETURN m.namedepartment高血压挂什么科MATCH (d:Entity {name:高血压})-[:就诊科室]-(dep:Entity) RETURN dep.namePython 里拼模板有个绕不开的坑Cypher 的节点{name:高血压}本身带大括号和str.format的语法冲突。我的处理方式是模板里用双大括号转义槽位用单大括号占位INTENT_TEMPLATES { symptom: MATCH (d:Entity {{name:{disease}}})-[:表现]-(s:Entity) RETURN s.name AS name, treat: MATCH (d:Entity {{name:{disease}}})-[:治疗]-(m:Entity) RETURN m.name AS name, contraindication: MATCH (d:Entity {{name:{disease}}})-[:禁忌]-(m:Entity) RETURN m.name AS name, department: MATCH (d:Entity {{name:{disease}}})-[:就诊科室]-(dep:Entity) RETURN dep.name AS name, } def build_cypher(intent, slots): disease slots.get(disease, [])[0] if intent not in INTENT_TEMPLATES or not disease: return None return INTENT_TEMPLATES[intent].format(diseasedisease)看着绕但规则很简单{{代表输出一个{}代表输出一个}单{disease}是留给format的变量。如果不做双大括号转义format会把 Cypher 里的花括号当成格式控制符直接报 KeyError。这个报错信息特别容易把人带偏我会在注释里单独标一句Cypher 大括号必须双写。4. 用 Flask 把问答系统包成服务接口设计与对接规范4.1 接口设计POST /api/qa 的请求与响应结构问答核心逻辑写完后暴露给前端的最小接口就是一个POST /api/qa接收 JSON返回 JSON。用 Flask 实现非常直接from flask import Flask, request, jsonify from neo4j import GraphDatabase app Flask(__name__) driver GraphDatabase.driver(bolt://localhost:7687, auth(neo4j, password)) app.route(/api/qa, methods[POST]) def qa(): payload request.get_json(forceTrue) question payload.get(question, ).strip() if not question: return jsonify({code: 400, msg: question 不能为空}), 400 answer pipeline(question) return jsonify({code: 0, data: answer}) def pipeline(question): 完整的问答链路意图 - 槽位 - Cypher - 图查询 - 答案。 intent classify_intent(question) slots extract_slots(question) cypher build_cypher(intent, slots) if cypher is None: return {type: text, content: 我还没学会这个问题换个问法试试。} with driver.session() as session: records session.run(cypher).data() names [r[name] for r in records] return {type: list, content: names}driver是全局唯一的这是 Neo4j Python 驱动的推荐用法一个 driver 实例内部维护连接池千万别在函数里反复创建。pipeline这条链路上任何一步失败都会抛异常生产环境建议在最外层包一层 try/except返回兜底话术而不是把堆栈直接暴露给前端。响应结构我用code data的包裹式code0表示成功非 0 表示业务异常。data里再分出type目前只有两种text是纯文本答案list是列表答案。后续要加知识卡片、药品说明书详情就在data里再扩展字段前端不用改调用逻辑。4.2 连接池与超时别只在本地跑通本地单机跑没问题一上测试环境就卡十有八九是连接池和超时没配。Neo4j Python 驱动支持在初始化时指定连接池大小和获取连接的超时时间driver GraphDatabase.driver( bolt://localhost:7687, auth(neo4j, password), max_connection_pool_size20, connection_acquisition_timeout60.0, )max_connection_pool_size决定驱动最多建多少个底层连接默认值 100 在个人项目里偏大多开几个 Flask worker 后 Neo4j 的连接数会被打满connection_acquisition_timeout设置的是排队等连接的上限如果连接池耗尽这个超时比 TCP 层卡死更早暴露问题日志里能看到Failed to obtain a connection from connection pool这时候就该考虑加连接数或者上缓存了。另外要区分两把锁单条 Cypher 执行慢优先看图谱有没有走索引整体并发上不去才去看连接池。前者是查询计划问题后者是资源问题混在一起排查会浪费很多时间。4.3 跟前端的对接约定答案的三种形态接口最容易翻车的不是后端逻辑而是返回结构没说清。我见过前端按data.content是字符串来渲染结果后端返回了数组页面上直接显示object Object。我在接口文档里约定的三种形态纯文本、列表、实体卡片。文本型直接渲染列表型前端做 tag 流式展示卡片型用于查询某种药品说明书这类需要展示多字段的场景。列表型响应里我加了一个empty标志方便前端判断空结果{ code: 0, data: { type: list, content: [头晕, 头痛, 心悸], empty: false } }约定好了之后前后端并行开发才不会互相等。我一般建议后端把接口的 mock 响应先固定下来前端拿 mock 先做页面后端再慢慢补真正的逻辑。5. 避坑医疗知识图谱问答系统最常见的 5 个隐藏坑5.1 同义词没归一化图谱节点查不到现象用户问高血压饮食注意什么系统答非所问甚至直接兜底但图谱里明明有高血压节点。原因图谱构建时用的是标准名高血压问句里是高血压病或原发性高血压槽位填充拿到的是用户原词拼进 Cypher 后{name:高血压病}匹配不到节点。解决在extract_slots之后加一层归一化映射把别名统一映射到标准名。数据导入时也要做同一件事保证图谱里只有标准名别名单独存到一个字典或 Neo4j 的ALIAS关系中。提示别名表是这类系统的后悔药每发现一个查不到的词就补一条成本极低收益极高。5.2 实体识别准了但 Cypher 查回来是空的现象识别出胃疼没问题但库里的关系是(胃炎)-[:表现]-(胃疼)用户问的是胃炎吃什么药查[:治疗]关系为空。原因用户问法和图谱里的谓词不匹配。吃什么药对应治疗但有的数据源里写的是使用或用于关系命名不统一。解决定义关系别名表构建图谱时把治疗使用用于统一成标准关系治疗。查询模板只认标准关系不认别名。5.3 关系方向搞反答案牛头不对马嘴现象用户问高血压有什么症状返回了一堆药名。原因构建时有人写成了(高血压)-[:表现]-(硝苯地平)或者导入脚本里主语宾语写反了。数据量一大方向错误很难靠肉眼排查。解决导入前就强制约定主语是疾病、宾语是目标在import_triples里加断言拦截明显异常的方向。查询模板里统一写(d)-[:关系]-(o)不留第二条路。这个约定应该写进团队的接口文档防止后来者按自己的习惯再加一套反向关系。5.4 兜底回答太生硬用户直接放弃现象系统回复我还没学会这个问题用户转头就去用了别的工具。原因兜底逻辑只有一句固定话术没有给用户任何线索。解决兜底时先检查槽位是否抽到了实体。如果抽到了实体但意图是 fallback就回复我掌握的症状/药物信息有限你试试问我‘某个病有什么症状’。如果实体也没抽到再走完全兜底。这个两级兜底让用户觉得系统在努力理解而不是直接摆烂。5.5 压测一上来就超时问题多半不在 Python现象单条请求 100ms并发 50 后延迟涨到 5 秒。原因三个叠加问题——图谱没建索引、每条 Cypher 全表扫、Flask 开发服务器单线程。解决先给Entity(name)建唯一约束再把 Flask 换成 gunicorn 多 worker 运行最后看 Neo4j 的查询计划确认是否走了索引。排查顺序是从数据库到服务端而不是反过来改代码因为压测超时 90% 的根因都在数据层。6. 验证与进阶从能聊到能用6.1 用 100 条问句评估准确率问答系统做完了得给一个数字证明它值得用。我习惯维护一个 100 条左右的小评测集每条问句标注期望实体集合再跑脚本算准确率TEST_CASES [ (高血压有什么症状, [头晕, 头痛, 心悸]), (高血压能吃什么药, [硝苯地平, 卡托普利]), (糖尿病挂什么科, [内分泌科]), ] def evaluate(test_cases): 计算系统在测试集上的准确率期望答案与返回答案有交集即算命中。 hits 0 for question, expected in TEST_CASES: answer pipeline(question) got set(answer.get(content, [])) if got and set(expected) got: hits 1 return round(hits / len(test_cases), 4)判定标准我特意放宽成有交集因为一个高血压有什么症状本来就有多个正确答案模型返回其中一部分也算对。严格的 MRR、Hit3 这些指标等上线后再补也不迟。评测集要跟着需求迭代每上线一批新问法就追加用例相当于给系统做回归测试。6.2 知识更新的半自动流程图谱不是一次建完就结束的。药品说明书改版、新指南发布都需要持续更新。我的做法是每类数据源走一条固定的更新通道数据源更新频率导入方式药品说明书每月CSV 增量导入临床指南每季度人工审核后全量重建同义词表按需运行时字典热加载增量导入的关键是只跑变更部分先导出新增的三元组再执行一次查重合并避免重复节点。同义词表热加载则让我不用重启服务就能加别名规则。我做过的最蠢的一次事故是没做评测集就上线结果同义词表改坏了一大批查询用户反馈了两天才发现。现在我的习惯是任何改动先跑一遍evaluate()分数没掉才敢部署。这个习惯帮我省掉了大量回归排查的时间希望帮到你。本文还有配套的精品资源点击获取
