简介一套面向医学领域的Python知识图谱问答系统适合作为毕业设计、课程设计或期末大作业也适合希望了解知识图谱与问答系统结合的开发者。代码附有详细注释整体经过严格调试下载部署后即可运行新手也能较快上手。资源包共70个文件约51.62MB以py源码为主并包含pyc缓存、json配置文件、docx说明文档、pkl模型数据、bat一键启动脚本和txt使用说明覆盖模型训练、知识图谱构建、意图识别与接口调用等环节。内容涉及基于BERT的意图识别、bilstm序列标注、知识抽取与图谱构建等关键模块目录结构清晰便于按模块阅读和二次开发。系统功能完善、界面美观还配有使用说明和启动脚本已有149人学习浏览下载后按文档提示即可快速跑通可用于答辩演示或进一步扩展。1. 医疗知识图谱问答系统先搞清楚它到底解决什么问题挂号台每天被问得最多的不是“怎么挂号”而是“我头疼三天了该吃什么药”“这个检查报告里肌酐偏高说明什么”。这些问题有一个共同特征患者想要的是基于医学知识库的直接回答而不是一堆搜索结果。python 技术栈做医疗知识图谱问答系统就是把零散的医学知识整理成图结构再让用户用自然语言问、系统用 Cypher 查询去答。它比纯关键词搜索强在能回答“高血压患者长期吃阿司匹林要注意什么”这种需要多跳关系的复杂问题。本文按数据建模、实体识别、查询生成、答案排序到测评避坑的顺序给你一条能落地的路径。想从 0 跑通一个能演示、能评估、能继续迭代的医疗智能问答系统这套方案可以直接照做。2. 搭建医疗知识图谱从结构化数据到 Neo4j 的三层设计2.1 医疗数据选型与清洗为什么不能直接用公开的原始数据医疗知识图谱的地基是数据。很多初学者第一步就翻车——拿到的公开医学数据里诊断名称、症状描述、药品名称的格式五花八门“高血压”被写成“高血压病”“原发性高血压”“头痛”和“头疼痛”混在一起。如果这些字段直接进图数据库查询结果必然残缺。常见的做法是选一类结构相对规整的数据源作为起点比如医学百科的结构化词条、药品说明书字段、临床指南里的诊断标准。我一般建议先做「三层清洗」字段归一化统一大小写、全半角、去掉多余空格把“35天”和“3-5天”统一成“3-5天”。同义词映射建一张同义词字典把“头疼”“头痛”“headache”全部映射到标准实体“头痛”。关系去重同一条“疾病-症状”关系如果从多个来源抽取到只保留一条合并证据来源。清洗后的数据保存成三张表实体表、关系表、属性表。属性表用来存“用药剂量”“发病季节”这类节点属性不单独建节点。import pandas as pd # 假设原始数据是从某个结构化医学数据源导出的 CSV df pd.read_csv(raw_medical_data.csv) # 归一化函数统一全半角、去掉两端空字符、压缩内部空格 def normalize(s): if not isinstance(s, str): return s s.strip().replace(, ().replace(, )).replace(, :) return .join(s.split()) df[disease_name] df[disease_name].apply(normalize) df[symptom_name] df[symptom_name].apply(normalize) # 同义词归一把口语化/英文别名映射到标准名 synonym_map { 头疼: 头痛, 头痛病: 头痛, headache: 头痛, 高血压病: 高血压, 原发性高血压: 高血压, } df[symptom_name] df[symptom_name].replace(synonym_map) df[disease_name] df[disease_name].replace(synonym_map) # 按疾病, 症状, 关系类型去重保留第一条 df_dedup df.drop_duplicates(subset[disease_name, symptom_name, relation_type]) df_dedup.to_csv(cleaned_triples.csv, indexFalse)这段代码的核心逻辑不是做复杂 NLP而是把脏字符串变成可以匹配的键。normalize()里的全半角转换特别重要中文医疗文本里括号和冒号全角半角混用非常常见不做这一步后面实体匹配命中率会直线下降。同义词映射表是医疗问答系统的“后悔药”——宁可在这里多花时间。每映射一个词问答里就少一种“明明库里有就是查不到”的玄学问题。清理完之后把三元组存成 CSV供下一步写入图数据库。2.2 实体关系建模主诉、疾病、药品、检查四类节点的设计知识图谱没有标准答案设计取决于问答系统要回答什么问题。对于医疗问答最少需要四类实体节点和六类关系实体类型含义典型属性Disease疾病疾病/诊断名名称、别名、科室、概述Symptom症状患者主诉名称、部位、特征Drug药品用药建议名称、剂型、禁忌Check检查检查项目名称、目的关系设计上最常用的是六类Disease-HAS_SYMPTOM-Symptom疾病表现、Disease-TREAT_WITH-Drug疾病用药、Disease-NEED_CHECK-Check疾病需检查、Check-CHECK_FOR-Disease检查确诊疾病、Symptom-SUGGEST_CHECK-Check症状提示检查、Drug-TREAT_FOR-Disease药品治病。问答系统 80% 的查询集中在两类链路从症状找疾病症状→疾病→用药从疾病找检查和用药疾病→检查/药品。建模时把这两个链路做成正向关系代码写起来顺手Cypher 也不会绕弯。节点属性里“别名”字段要单独存一个列表属性不要拼成字符串。Neo4j 支持数组属性用[高血压, hypertension, 原发性高血压]这种形式存储查询时用ANY(alias IN node.alias WHERE alias $name)匹配比用CONTAINS做模糊匹配快得多也不会误匹配“高血糖”和“高血压”。from py2neo import Graph, Node, Relationship, Subgraph # 连接 Neo4j注意默认密码在首次启动后要改 graph Graph(bolt://localhost:7687, auth(neo4j, your_password)) # 读取清洗后的三元组 triples pd.read_csv(cleaned_triples.csv) # 批量创建节点和关系使用事务分批写入 def create_medical_graph(graph, triples, batch_size500): batch [] entity_cache {} def get_entity(entity_type, name, attrsNone): # 实体缓存避免重复创建同一节点 key (entity_type, name) if key in entity_cache: return entity_cache[key] node Node(entity_type, namename, **(attrs or {})) entity_cache[key] node return node for i, row in triples.iterrows(): disease get_entity(Disease, row[disease_name]) symptom get_entity(Symptom, row[symptom_name]) rel Relationship(disease, HAS_SYMPTOM, symptom) batch.extend([disease, symptom, rel]) if len(batch) batch_size * 3: graph.create(Subgraph(batch)) batch [] if batch: graph.create(Subgraph(batch)) create_medical_graph(graph, triples)batch_size500的控制是为了避免在 Neo4j 里创建超大事务。py2neo 的create(Subgraph(batch))会把节点和关系统一提交如果数据量在十万级三元组以内这种方式比graph.run一条条执行快一个数量级。还有一个细节实体节点用name做属性名不要用id。图数据库里id是保留概念用户很容易把业务主键和数据库内部 id 搞混。所有节点统一带source属性标记数据来源后面排查错误数据时能直接定位到是药品说明书还是百科词条的问题。2.3 用 py2neo 批量写入 Neo4j导入脚本与索引策略数据量和写入速度在实际操作里是第一个坎。用 py2neo 的graph.create()逐条插入一万条三元组可能要跑十几分钟。批量写入是必须的上面代码里的事务分批已经是常见做法。导入完成后索引设置比数据本身更影响问答速度。Neo4j 默认对name字段没有索引如果问答查询直接WHERE n.name 头痛在十万节点上是全库扫描延迟能从 10ms 飙到 700ms。# 在 Neo4j Browser 里执行或者用 Python 的 graph.run() CREATE INDEX FOR (d:Disease) ON (d.name); CREATE INDEX FOR (s:Symptom) ON (s.name); CREATE INDEX FOR (dr:Drug) ON (dr.name); CREATE INDEX FOR (c:Check) ON (c.name);这四条索引建完之后按名称查实体的查询走索引响应时间基本都能控制在 50ms 以内。实际项目中还会给Disease.alias建索引因为用户问“高血压”时系统要先在别名列表里找到Disease(name高血压)如果别名用列表属性存储需要用EXISTS子查询或展开成字符串这一步是后续查询优化的重点。导入完成后要验证一下图结构是否合理。常见做法是在 Neo4j Browser 里执行MATCH (d:Disease)-[r:HAS_SYMPTOM]-(s:Symptom) RETURN d.name, s.name LIMIT 20查看抽样关系。我一般会额外统计每个实体类型的节点数和关系数防止清洗阶段把数据丢了MATCH (d:Disease) RETURN count(d)和清洗前源表的行数对比差异超过 1% 就要检查清洗逻辑是不是把关系过滤掉了。3. 把自然语言问句解析成 Cypher 查询实体识别和意图识别怎么做3.1 基于词典与规则的最小落地方案先跑通再谈深度学习医疗问答的输入是五花八门的口语“大夫说我爸高血压平时能吃布洛芬吗”“嗓子疼了两天该吃什么药”。这些句子往往没有规范的结构但有一个好处实体词相对固定。“嗓子疼”“高血压”“布洛芬”都是图谱里的实体词只要能把它们从问句里切出来再识别出意图就能查库。第一步是用 jieba 分词并且加载自定义词典。医疗术语是典型的“分词即坑”领域“胸骨后疼痛”如果被切成“胸骨/后/疼痛”实体匹配必然失败。自定义词典里要把所有实体名加进去。import jieba # 把实体名称写入词典文件每行一个词附词频和词性可选项 jieba.load_userdict(medical_entities.txt) # 强制优先匹配长词 def extract_entities(question): # 直接分词 tokens list(jieba.cut(question)) # 在切好的词里找实体 entity_types set(tokens) set(all_entity_names) return list(entity_types)这种做法的限制很明显词典没有覆盖的症状词会被切开丢掉。缓解办法不是追求完美分词而是加一层“最长匹配滑动窗口”def extract_entities_with_window(question, entity_list): matches [] # 按长度降序遍历实体词保证“胸骨后疼痛”优先于“疼痛”被匹配 for entity in sorted(entity_list, keylen, reverseTrue): if entity in question: matches.append(entity) # 注意简单替换会重复命中需记录终止位置 return matches窗口匹配的关键是排序最长实体优先匹配避免“高血压”命中后“原发性高血压”被忽略。两个方法各有问题jieba 分词容易切碎未登录词滑动窗口容易重复匹配。实际项目我会结合起来——先用 jieba 切得到候选词再用实体表对问句做一次最长匹配最后把两者交集作为候选。宁可多召回不提前过滤因为下一步意图识别和候选排序会处理噪音。3.2 意图识别与槽位填充让“我头疼三天该怎么办”变成查询条件实体匹配解决“问句里有什么”意图识别解决“用户想干什么”。医疗问答系统至少要区分四类意图查疾病什么病、查症状有什么表现、查用药吃什么药、查检查做什么检查。意图分类用规则比用模型稳妥。医疗问句的句式相对固定“该吃什么药”包含“药”“该怎么办”泛化成“查疾病”。常见做法是维护一组触发词表# 意图触发词表 INTENT_RULES { disease: [什么病, 怎么了, 怎么回事, 怎么办, 啥问题], drug: [吃什么药, 用什么药, 用药, 吃什么, 能不能吃], check: [做什么检查, 查什么, 怎么查, 检查], symptom: [什么症状, 表现, 有啥反应], } def detect_intent(question, entities): for intent, patterns in INTENT_RULES.items(): for pattern in patterns: if pattern in question: # 用实体类型来校验意图是否合理 if intent drug and not any(e.type in (Drug, Disease) for e in entities): continue return intent return disease # 默认意图意图识别要和实体类型联动。“布洛芬能治什么病”里实体是 Drug意图应该从“drug”反转为“disease”——用户不是问吃什么药而是问药治什么病。这种联动规则写在validate_intent函数里根据主体实体的类型修正意图。槽位填充针对“头疼三天”这种带时限的表达。常见做法是做正则抽取把“三天”“一周”“两个月”提取成数值单位作为时间限定槽位。医疗问答里时限不参与图谱查询只参与答案模板渲染——严谨的回答会提示“症状持续 3 天未缓解请及时就医”。import re def extract_time_slot(question): # 匹配数字 时间单位 pattern r(\d{1,2})[天周个月年] match re.search(pattern, question) if not match: return None, None duration match.group(1) unit re.search(r([天周个月年]), question[match.start():])[1] return duration, unit注意re.search可能匹配到“血糖 6.8 算高吗”中的“6.8”吗不会因为正则要求数字后必须跟时间单位而“6.8 算”不满足。但实际中文问句里“6个月”这种描述非常常见需要把“半年”也映射成“6个月”这一步用同义词映射表扩展。3.3 生成 Cypher 并查询 Neo4j属性匹配、关系方向与返回字段实体和意图都有了接下来把三元组装成 Cypher 查询。这里是最容易翻车的地方——Cypher 语法在关系方向上非常敏感。问“高血压有什么症状”时意图是“symptom”主体实体是 Disease查询写成def build_cypher(entities, intent): main_entity entities[0] # 按权重排权重高的作为查询主体 query_templates { # 从疾病找症状 symptom: MATCH (d:Disease {name: $name})-[:HAS_SYMPTOM]-(s:Symptom) RETURN s.name LIMIT 10, # 从疾病找药 drug: MATCH (d:Disease {name: $name})-[:TREAT_WITH]-(dr:Drug) RETURN dr.name LIMIT 10, # 从症状找疾病 disease: MATCH (s:Symptom {name: $name})-[:HAS_SYMPTOM]-(d:Disease) RETURN d.name LIMIT 10, } cypher query_templates[intent] return cypher, {name: main_entity.name}关系方向别记反HAS_SYMPTOM是从 Disease 指向 Symptom所以“从症状找疾病”要用反向匹配-[:HAS_SYMPTOM]-(d:Disease)。很多新手在这里查出来空结果不是数据没有是方向错了。查询结果返回后要做去重和排序。同一个症状可能关联多个疾病同一个疾病可能关联多种药。我的策略是先LIMIT 20拉回候选在 Python 里排序而不是在 Cypher 里ORDER BY。原因后面讲排序逻辑依赖问答系统的上下文权重在 Python 里改起来更快。def query_neo4j(graph, cypher, params): result graph.run(cypher, params).data() field list(result[0].keys())[0] if result else None return [row[field] for row in result] if field else []这里的graph.run(...).data()返回字典列表比如[{d.name: 高血压}]。取字段时用list(result[0].keys())[0]是通用写法避免写死字段名。4. 问答系统的答案生成模板匹配与候选排序4.1 答案模板设计把查询结果转成人话Cypher 查询返回的是实体名列表直接丢给用户会显得很“机器腔”。答案生成环节要把结构化结果渲染成通顺的句子。模板要区分单实体和多实体两类场景。def render_answer(intent, entities, result_list): if not result_list: return 目前知识图谱里没有找到直接相关的信息建议描述更具体一些或去医院做检查。 if intent drug: main_disease entities[0].name drugs 、.join(result_list[:5]) return f针对{main_disease}知识库中提及的药物有{drugs}。具体用药需遵医嘱不能自行调整剂量。 if intent disease: main_symptom entities[0].name diseases 、.join(result_list[:5]) return f{main_symptom}可能关联的疾病有{diseases}。这个结果只做参考请勿自行诊断。 return f查询到信息{、.join(result_list[:5])}模板里的免责声明很重要。医疗问答不像天气问答答错了有实际风险答案尾部出现“建议就医/遵医嘱”不是废话是整个系统可信度的底线。模板不是越多越好四到五种固定句式足够覆盖大多数问法泛化模板反而容易生成语义不通的句子。4.2 召回与排序问句有噪音时怎么选正确答案真实问句里实体识别大概率多抽或少抽。“嗓子疼该吃什么药”可能同时匹配到“嗓子疼”症状实体和“嗓子”部位词。如果部位词误当成实体查询就会拿到错误结果。解决方法是召回排序。排序权重我一般按如下规则设计实体长度优先匹配到的实体字符串越长越可能是真正的实体。“胸骨后疼痛”“疼痛”。实体位置优先出现在问句末尾的实体往往比开头的权重要因为“该吃什么药”里的“药”是意图触发词不是主体。意图匹配优先如果意图是“查疾病”那么 Disease 实体的权重提升Symptom 实体权重降低。def rank_entities(entities, question, intent): def score(entity): s 0 s len(entity.name) # 长实体优先 if question.rfind(entity.name) len(question) / 2: s 10 # 后半段出现优先 if intent disease and entity.entity_type Disease: s 20 elif intent symptom and entity.entity_type Symptom: s 20 return s return sorted(entities, keyscore, reverseTrue)sorted的 key 函数里叠加了三个维度的得分实测下来能解决 70% 的噪音实体问题。剩下的噪音比如“布洛芬”和“芬必得”的同义实体在同义词归一化阶段就要解决掉不能依赖排序兜底。4.3 全链路串起来一个最小可运行的问答循环把前面几段代码串成一个完整的问答函数输入问句输出答案def answer(question): # 1. 实体识别先词典后滑动窗口 entities extract_entities_with_window(question, entity_names) if not entities: return 我没有理解你说的症状或疾病请换个说法。 # 2. 构造 Entity 对象并标记类型 entity_objects [lookup_entity(name) for name in entities] # 3. 意图识别 intent detect_intent(question, entity_objects) # 4. 实体排序选主体实体 ranked rank_entities(entity_objects, question, intent) main_entity ranked[0] # 5. 生成 Cypher 并查询 cypher, params build_cypher([main_entity], intent) results query_neo4j(graph, cypher, params) # 6. 渲染答案 return render_answer(intent, [main_entity], results)这个循环已经能跑通最小系统。注意lookup_entity要去 Neo4j 查一次实体名拿到实体类型也可以提前把全部实体名加载到内存里的字典避免每次问答都打数据库。十万级实体的字典也就几十 MB 内存加载一次完全能接受。全链路跑通后下一步不是优化准确率而是构建测试集。没有测试集你根本不知道一次改动是变好还是变坏后面迭代就是玄学。5. 医疗问答常见避坑从实体链接错误到 Neo4j 内存翻车5.1 现象图里明明有“头痛”但问“头疼怎么办”查不出结果原因出在词典没做同义词归一。用户输入“头疼”时分词器切出“头疼”但图谱里只有“头痛”实体没有“头疼”节点。词典里两个词都存在就产生了两个独立实体。解决方法是建同义词映射表在实体识别后统一归一头疼 - 头痛。这个表要放在配置里不要硬编码在代码里。维护习惯是每遇到一次“查不到”的病例就把用户的原始说法记下来补充进映射表。运行一个月后这套系统能覆盖的口语表达会越来越全。# 兜底逻辑同义词表找不到时才建新实体 def normalize_entity(name): name synonym_map.get(name, name) return name5.2 现象Cypher 查询带中文参数查不出来Neo4j 日志报 index 错误原因中文字符串在 Neo4j 参数传递时出现编码问题。常见于使用graph.run(MATCH ... WHERE n.name {name}, name高血压)的旧版传参方式{name}语法在某些驱动版本下不支持中文。解决统一用$param语法传参# 正确写法 graph.run(MATCH (d:Disease {name: $name}) RETURN d, name高血压)关于 Python 侧编码Windows 终端下如果print中文乱码检查 Python 文件头部有没有# -*- coding: utf-8 -*-在 Python 3 下默认 UTF-8一般不需要。但 CSV 导入时encodingutf-8-sig很有必要否则首行\ufeff会污染第一列列名。5.3 现象大批量写入十万元组时 py2neo 卡死内存爆满原因是graph.create()每次都创建独立事务而且把节点数据囤积在 Python 列表里。我见过最夸张的代码把 5 万元素放在一个batch列表里一次性提交Neo4j 事务内存直接溢出整库无响应。解决事务分批 控制事务大小。前面第 2.3 节给出的batch_size500是经验值实际按节点关系的数量调整单事务不超过 3000 个实体对象。另外写入中途若报“Connection reset”不要直接重启 Neo4j先检查neo4j.conf里的dbms.memory.heap.max_size是不是默认的 512M改成 2G 以上再重启。# 修改 neo4j.conf dbms.memory.heap.initial_size2G dbms.memory.heap.max_size4G dbms.memory.pagecache.size2G内存配置是医疗知识图谱问答系统的“隐藏坑”数据量到 20 万三元组时默认配置必翻车。改完配置要重启 Neo4j 才生效。5.4 现象关系方向写反返回空结果且不报错原因Cypher 里(d)-[:HAS_SYMPTOM]-(s)和(s)-[:HAS_SYMPTOM]-(d)是两种不同语义。新手常把“从症状反查疾病”写成(s)-[:HAS_SYMPTOM]-(d)Cypher 不会报错但查询结果总是空。解决把所有关系方向整理成一张表注释清楚“谁指向谁”放在代码文件头部。每次写 Cypher 前先查这张表不要再凭感觉写箭头。5.5 现象同一个问题问两次得到的结果偶尔不一样甚至混乱原因Neo4j 查询结果默认排序不稳定尤其是LIMIT 10在没有ORDER BY时返回哪 10 条由存储引擎内部决定。加上图谱里有多条同类关系时结果集每次都不同。解决Cypher 里明确加ORDER BY字段名比如RETURN d.name ORDER BY d.name或者 Python 侧对结果排序。答案排序不稳定直接毁掉用户体验用户会怀疑系统“抽风”。排序列名需要查询结果里包含用RETURN s.name AS name ORDER BY name保持返回字段一致。6. 进阶把规则系统升级成可评估的医疗问答服务6.1 用自动化测试集给问答打分准确率、召回率与 F1 怎么算规则系统不经过评测就上线等于闭着眼睛开车。我建议至少准备 200 条测试问句每条标注“期望答案实体集合”。比如“高血压吃什么药”标注期望输出字符串包含“硝苯地平”“氨氯地平”“缬沙坦”等。test_cases [ { question: 高血压吃什么药, intent: drug, expected_entities: [硝苯地平, 氨氯地平], }, { question: 经常头痛该做什么检查, intent: check, expected_entities: [头颅CT], }, ] def evaluate(test_cases, answer_func): correct 0 total len(test_cases) for case in test_cases: output answer_func(case[question]) hit any(e in output for e in case[expected_entities]) correct hit if not hit: print(f失败{case[question]} - {output}) precision correct / total return precision为什么用“期望答案实体是否出现在输出里”而不是“字符串完全相等”因为模板渲染的句子可能多了免责声明和提示语完全相等太苛刻实体命中才是问答系统的核心能力。F1 的召回率在这个场景下不好算因为每条问句的期望答案集合大小不同我一般只报“命中率”和“首次查询平均耗时”这两个指标足够发现问题。测评要写进 git hooks 或 CI每次改代码都跑一遍。不跑测试就上线第二天用户就会拿你的系统跟竞品比一旦答错被截图发到网上整个项目信誉就崩了。6.2 日志与错误样本沉淀迭代问答系统的最省力方式规则系统有个好处错误是可解释的。每条失败问句沿着“分词 → 实体 → 意图 → 查询 → 渲染”链路都能找到断在哪一环。所以生产环境一定要记录完整链路日志。import logging def answer_with_log(question): logging.info(f问句: {question}) entities extract_entities_with_window(question, entity_names) logging.info(f实体: {[(e.name, e.entity_type) for e in entities]}) if not entities: logging.warning(无实体命中需要补充词典) return 未命中实体 intent detect_intent(question, entities) logging.info(f意图: {intent}) ...常见的迭代节奏是每周收集一批日志里的“未命中实体”问句人工检查是词典缺词还是同义词缺映射补完重新跑测试集确认不降低原有准确率再上线。这套流程跑三个月准确率能从 70% 提到 90% 以上。注意“不降低原有准确率”是硬指标规则系统的改动经常会修好一个旧问题又弄坏三个新问题只有自动化测试集能拦住这种回归。最后一个建议给答案加一个“可信度”字段。当候选少于 2 个时答案开头加“仅供参考信息较少请优先就医”候选多于 5 个时只展示前 3 个并提示“还有其他可能”。这种细节虽然不直接影响准确率但会显著提高用户对结果质量的接受度。我的习惯是医疗问答永远给参考、永远不确诊这个原则比任何技术优化都重要。希望帮到你。本文还有配套的精品资源点击获取