简介面向医疗知识图谱与问答系统入门者的Python项目资料完整覆盖知识图谱构建、医疗实体/关系数据整理以及基于规则匹配的对话系统实现。资源共58个文件包含设计报告Word文档、Python源码与医疗数据、项目运行截图等其中py脚本负责图谱构建与问答流程txt为词典数据png/jpg为界面与图谱可视化展示压缩包约20MB。已有2570人学习下载。项目分为简单知识图谱、医疗知识图谱构建、以及对话系统三个模块可直接运行体验从数据到图谱再到问答的完整链路遇到自由输入分析不足、输出受限等局限时还可结合设计报告与源码理解其规则匹配原理为后续引入深度学习模型做过渡。适合作为课程设计、毕业设计或知识图谱入门的学习参考。1. 医疗知识图谱问答系统把“高血压患者能吃布洛芬吗”变成一条图查询你输入“高血压患者能吃布洛芬吗”传统搜索会拆出“高血压”和“布洛芬”两个词然后各查各的资料却回答不了“能不能同服”。基于Python实现的医疗知识图谱问答系统解决的正是这类多实体、多跳关系的问题把疾病、症状、药品、科室、检查组织成一张图再把自然语言问句翻译成对这张图的查询最后拼成人话返回。做这个东西不是给大医院做临床决策而是给健康类App、导诊机器人、医学检索工具一个能定制的问答底座。想入门的Python开发者能借此把分词、实体识别、图数据库、意图分类串成一条完整的工程流水线已经有后端经验的人则能直接在这套结构上扩充数据和模型把答案从“能查到”做到“答得准”。2. 系统架构与组件选型五个环节、一个图库、一个FastAPI服务2.1 问答流水线拆解从问句到Cypher查询的五个环节拿到这个问题时我会先画一条管道而不是先写代码。整条问答流水线分五个环节问句预处理、实体识别与链接、意图识别、槽位填充与查询生成、答案生成。问句进来先分词、去停用词然后从分词结果里挑出疾病、症状、药品、科室这类实体接着判断用户到底想问什么是问症状还是问禁忌还是问科室再把这些信息填进预设的查询模板最后把图数据库返回的记录组转成一句自然语言。五个环节的耦合方式决定了后面好不好维护。我的做法是每个环节输出固定的结构实体识别输出一个实体列表意图分类输出一个意图标签查询生成只接收这两个结果。这样你换分词器、换意图模型都不会牵连到查询层。先跑通一个FastAPI服务让外部请求能进到流水线里来。from fastapi import FastAPI from pydantic import BaseModel from qa import QAPipeline app FastAPI() qa QAPipeline() # 内部串起分词 - 实体 - 意图 - 查询 - 答案 class Question(BaseModel): text: str app.post(/qa) def answer_question(q: Question): result qa.run(q.text) return { answer: result.answer, intent: result.intent, entities: result.entities, }这段代码把HTTP接口和问答逻辑分开QAPipeline是后面章节要实现的组合类result里除了答案还带意图和实体方便调试时定位是哪一环出了问题。接口层不做业务判断只负责接收问句和返回结构化结果这样后面无论接命令行、Web页面还是微信小程序都只需要再套一层适配器。2.2 存储选型为什么不用MySQL全文索引而是用图数据库存储层我一般直接选Neo4j不是因为图数据库听起来高级而是因为医疗知识天然是图结构。一个“高血压”节点连着“头晕”症状、“硝苯地平”药品、“心内科”科室这种关系的查询如果用MySQL需要设计四五张关联表然后做多表JOIN查询深度一旦超过两层脚本就变得很难读。而Neo4j的Cypher查询里多跳关系的表达几乎是白话。数据模型按节点和关系拆我的设计如下节点类型典型例子主要属性疾病高血压、2型糖尿病name, icd10, department症状头晕、干咳、心悸name, describe药品布洛芬、阿莫西林name, dosage, usage科室心内科、呼吸内科name检查血常规、心电图name, purpose关系类型要表达语义而不是模糊的“相关”。比如疾病-[:HAS_SYMPTOM]-症状、药品-[:TREATS]-疾病、药品-[:CONTRA_INDICATION]-症状、疾病-[:BELONGS_TO]-科室。关系本身就是答案里最难解释的部分把关系名定清楚了后面写查询模板才能直接复用。也有场景不适合上Neo4j如果你的问答只做单实体、单跳查询比如只查“阿莫西林的说明书”那MySQL加全文索引反而更省事。我的判断标准是——问句里有没有出现两个及以上实体以及是否需要关系属性。出现“同时含两个实体、要走关系”这类需求再上Neo4j不迟。2.3 Python环境与服务端骨架先把连接层写对开始写代码前先检查本机的Python环境我用的是Python 3.8以上版本。新手最容易在环境上卡住建议先用python --version确认版本再用python -m venv venv建独立虚拟环境。日常开发我习惯用VS Code配好Python插件后直接F5启动服务断点调试问句处理流程比反复打印日志快得多。服务端依赖控制在最小集合一个requirements.txt文件就能装完fastapi uvicorn neo4j jieba pandas连接Neo4j时要注意Python驱动有两个常用选择官方neo4j驱动和py2neo。我推荐直接用官方驱动它维护稳定也支持事务和批量提交。连接串用Bolt协议不是HTTPfrom neo4j import GraphDatabase class Neo4jClient: def __init__(self, uri, user, password): self.driver GraphDatabase.driver(uri, auth(user, password)) def query(self, cypher, paramsNone): with self.driver.session() as session: records session.run(cypher, params or {}) return [record.data() for record in records] def close(self): self.driver.close()连接参数写在配置里不要硬编码在代码里。uri填bolt://localhost:7687如果填成7474会一直卡在连接超时。这个封装类后面所有查询都要用你可以在query方法里打印Cypher语句和参数排查问题时会轻松很多。3. 医疗知识图谱构建实操从数据清洗到实体关系抽取入库3.1 数据源选择与结构化清洗出能入库的DataFrame医疗图谱的数据源决定了系统答案的上限。常见公开数据源包括ICD-10中文疾病分类、药品说明书的结构化字段、医院科室对照表、部分开源的中文医学知识图谱子集。这些数据格式不统一可能是Excel、CSV、JSON甚至PDF我的做法是先统一转成CSV再用pandas做清洗把数据变成节点表和关系表两类。这里要特别提醒不要用未脱敏的真实患者病历医疗数据涉及隐私和合规开发和演示阶段只用公开的结构化数据就够跑通全流程。清洗的关键步骤是去重、补空值和规范化。比如“高血压”和“高血压病”在不同数据源里可能同时存在如果不合并后面的实体链接就会断裂。import pandas as pd # 读入原始疾病表 df pd.read_csv(raw_disease.csv, encodingutf-8) print(df.head(), df.shape) # 先看字段和规模 # 去重同一标准名只保留一个 df df.drop_duplicates(subset[name_std], keepfirst) # 标准化统一科室名去掉前后空格和全角空格 df[name_std] df[name_std].str.strip().str.replace( , ) # 空值处理科室缺失时填“未分类”症状缺失时保留空列表 df[department] df[department].fillna(未分类) # 输出清洗后的节点表 df[[name_std, icd10, department]].to_csv(disease_node.csv, indexFalse)drop_duplicates按标准名字段去重避免同一个疾病在建图时生成两个节点。fillna补空值是为了后续导入Neo4j时不需要额外处理None值。清洗完的disease_node.csv每条记录就是一个疾病节点。同样的逻辑也适用于药品表和症状表只是字段不同药品表要留dosage和usage字段。3.2 实体与关系抽取自定义词典加规则抽取的落地写法知识图谱里最有价值也最花时间的是关系数据。从公开数据里拿到的往往是“疾病有哪些症状”“药能治什么病”这样的半结构化文本需要用抽取脚本把它变成关系列表。我没有一开始就上复杂的BIO标注模型先用自定义词典加规则把80%的典型关系抽出来。jieba默认词典几乎没有医学词需要准备一个custom_dict.txt每行写一个词格式是“词 词频 词性”。比如“硝苯地平 10 nz”“2型糖尿病 10 nz”。加载后分词结果会明显贴合医学实体。import jieba import re jieba.load_userdict(custom_dict.txt) def extract_entities(text): 从问句或描述文本中提取候选实体 words jieba.lcut(text) entities [] for word in words: if word in medicine_set or word in disease_set or word in symptom_set: entities.append(word) return entities # 规则抽取药品用法例如“一次2片一日3次” dosage_pattern re.compile(r(一次|每次)(\d)(片|粒|袋|毫升)?.{0,4}(一日|每天)(\d)次) m dosage_pattern.search(阿莫西林用法一次2片一日3次) if m: print(提取剂量:, m.group(2), 粒/片, 频率:, m.group(4), 次/日)实体抽取这步只负责圈定“候选词”不做语义判断。用三个集合medicine_set、disease_set、symptom_set来过滤分词结果是因为在问答场景里问句很短不需要全部词都参与查询。药品剂量用正则抽是因为说明书里的剂量写法高度模板化正则比模型更可靠。这些规则后续要维护到一个独立配置文件里不要散落在各处代码中。3.3 批量入库Neo4j用MERGE加UNWIND替换逐条CREATE建图时最容易犯的错误是拿到数据后立刻用CREATE逐条插入。少量测试数据这样写没问题几千条数据时速度就会慢到怀疑人生。正确做法是先用约束保证节点唯一性再用UNWIND批量写入。先启动Neo4j执行一次约束创建语句CREATE CONSTRAINT disease_unique IF NOT EXISTS FOR (n:疾病) REQUIRE n.name IS UNIQUE;Python侧的导入脚本用官方驱动按批提交from neo4j import GraphDatabase import pandas as pd driver GraphDatabase.driver(bolt://localhost:7687, auth(neo4j, password)) diseases pd.read_csv(disease_node.csv).to_dict(records) insert_cypher UNWIND $rows AS row MERGE (d:疾病 {name: row.name_std}) ON CREATE SET d.icd10 row.icd10, d.department row.department ON MATCH SET d.icd10 row.icd10, d.department row.department RETURN count(d) with driver.session() as session: for i in range(0, len(diseases), 100): # 每100条一批 batch diseases[i:i100] result session.run(insert_cypher, rowsbatch) print(batch, i, inserted, result.single().value())MERGE是建图的关键它先按name查重节点存在就更新属性不存在就创建配合上面的唯一约束就能避免重复节点。分批提交每批100条既不会撑爆内存出错了也只需要重跑失败的那一批。ON CREATE和ON MATCH两个分支把属性更新的逻辑写全这样脚本重复执行是幂等的。关系数据的入库逻辑类似只是Cypher变成两段匹配再加合并UNWIND $rows AS row MATCH (d:疾病 {name: row.disease}) MATCH (s:症状 {name: row.symptom}) MERGE (d)-[:HAS_SYMPTOM]-(s)这里有个隐含要求症状节点必须已经存在否则第二条MATCH匹配为空整个批次都不会写入。所以入库顺序是先节点后关系而且要确保两边的名称都标准化过。4. 问答模块拆开做实体识别、意图分类与Cypher模板生成4.1 实体识别与归一化把“高血压”和“原发性高血压”指到同一个节点实体识别在问答里的任务不是“找出医学名词”而是“找出图谱里存在的节点”。很多系统跑完分词后直接把词拿去匹配结果图谱里存的是“高血压病”问句里是“高血压”匹配失败返回空答案。做这一步我靠一张同义词表来兜底。把所有可能指向同一节点的写法归到一个standard_name上。# synonyms.py SYNONYM_MAP { 高血压: 高血压病, 原发性高血压: 高血压病, 高血压病: 高血压病, 糖尿病: 2型糖尿病, 血糖高: 2型糖尿病, 布洛芬: 布洛芬缓释胶囊, } def normalize_entity(entity): 把同义词归一到标准名 return SYNONYM_MAP.get(entity, entity)实际执行时normalize_entity要放在图谱查询之前调用。比如问句“高血压患者能吃布洛芬吗”中抽出的“高血压”先归一成“高血压病”再作为查询参数传入匹配成功的概率会高很多。同义词表本质上是成本最低的知识更新手段。每发现一个“问不到答案”的case查一下是不是同义词问题是的话往表里加一行。这张表积累到500行以上时系统的有效回答率会有肉眼可见的提升。如果你想用向量相似度或者模型来做语义归一化那是进阶方向但它没办法完全替代同义词表因为医疗名词的别名差异远大于通用文本的语义差异。4.2 意图识别先定规则后用模型意图识别的目标是判断用户问的是哪一类问题。我按医疗问答的高频场景把意图分成五类意图标签问法示例需要的槽位disease_symptom高血压有什么症状疾病symptom_disease头晕是什么病症状contraindication高血压能吃布洛芬吗疾病、药品disease_drug高血压吃什么药疾病department_query高血压挂什么科疾病在数据量不足的初期规则就能覆盖主要问法。每个意图配几个关键词和正则在问句里找。规则识别不到的再考虑上分类模型比如TextCNN或BERT。不要一上来就训练模型规则已经能解决七八成问题。def detect_intent(text, entities): 基于规则识别意图返回意图标签和槽位 dis any(e in entity_types.get(疾病, []) for e in entities) med any(e in entity_types.get(药品, []) for e in entities) sym any(e in entity_types.get(症状, []) for e in entities) if dis and med: return contraindication, {disease: dis, drug: med} if dis and (症状 in text or 表现 in text or 有什么反应 in text): return disease_symptom, {disease: dis} if sym and (什么病 in text or 怎么办 in text): return symptom_disease, {symptom: sym} if dis and (吃什么药 in text or 用药 in text): return disease_drug, {disease: dis} if dis and (挂什么科 in text or 哪个科室 in text): return department_query, {disease: dis} return unknown, {}这里的判定顺序就是优先级顺序先判断多实体组合比如“疾病药品”组合优先走到禁忌查询否则“高血压能吃什么药”这类问句会因为同时出现“症状”两个字被误判成症状查询。规则里的关键词要覆盖口语变体“有什么反应”“需要注意什么”这类说法都要加进去否则会落到unknown。4.3 查询模板与答案生成写一组能覆盖主流问法的Cypher模板有了意图和槽位下一步就是把槽位填进Cypher模板。模板用字典维护每个意图对应一条查询语句QUERY_TEMPLATES { disease_symptom: MATCH (d:疾病 {name: $disease})-[:HAS_SYMPTOM]-(s:症状) RETURN s.name AS symptom , contraindication: MATCH (drug:药品 {name: $drug})-[:CONTRA_INDICATION]-(sym:症状) WHERE sym.name IN MATCH (d:疾病 {name: $disease})-[:HAS_SYMPTOM]-(s:症状) RETURN s.name RETURN drug.name AS drug, sym.name AS symptom , disease_drug: MATCH (d:疾病 {name: $disease})-[:TREATS]-(drug:药品) RETURN drug.name AS drug, drug.usage AS usage , }禁忌查询的模板值得一提它先用疾病-症状找高血压这个病的症状集再看布洛芬的禁忌症状集合里有没有和它重叠的这正好利用了图的多跳查询能力。查询结果组答案时要给空结果留退路def compose_answer(intent, records): if not records: return 图谱中暂时没有找到相关信息建议更换问法或补充数据。 if intent disease_symptom: symptoms [r[symptom] for r in records] return 常见症状有 、.join(symptoms[:8]) if intent disease_drug: drugs [r[drug] for r in records] return 常用药物包括 、.join(drugs[:5]) return 查询到了相关结果请参考 str(records[:3])答案生成不追求改写先把实体和关系原样呈现保证信息准确。等图谱数据和模板稳定后再做同义改写不迟。records[:8]和records[:5]这类的截断很重要用join拼接过多名词会让答案变得很长用户根本读不完不如只给最主要的信息。5. 常见避坑与排查问不到答案时先检查这四个环节5.1 图谱里明明有数据问答却返回空现象用Cypher在Neo4j Browser里查“高血压”的记录能查到问答系统返回“找不到相关信息”。原因实体归一化没做全。用户输入“高血压”图谱节点存的是“高血压病”两边对不上。同义词表里没有这组映射实体识别再准也会在查询这步静默失败。解决第一步加日志把查询前的槽位值和查询Cypher打到控制台。看到disease“高血压”而图谱里是“高血压病”就能确定是归一化问题。往SYNONYM_MAP里补一行映射即可。排查动作要形成习惯每次“返回空”先确认是实体没对齐、模板写错还是数据缺失而不是盲目改模板。5.2 neo4j驱动连接失败认证和协议不匹配现象neo4j.exceptions.ServiceUnavailable提示无法连接到服务器。原因连接串写错是最高频原因。我把URI写成http://localhost:7474但官方驱动只认Bolt协议。还有一种是Neo4j默认只允许neo4j用户在本地认证如果密码改了没同步也会报认证错误。解决URI改成bolt://localhost:7687。用户名密码再三确认密码可以在Neo4j Desktop的数据库设置里重置。我一般把连接信息放进.env文件代码里用os.getenv读取既方便换环境也避免把密码提交到代码仓库。5.3 批量导入几万条数据太慢跑了半小时还没完现象节点插入10000条脚本执行时间超过30分钟Neo4j控制台CPU飙高。原因每循环一次就session.run(create...)每次提交一个事务。事务提交的固定开销远比数据操作本身大数据量上来后自然慢。解决改成UNWIND批量提交每批100到200条能在一两分钟跑完。配合索引约束的MERGE同一批数据重跑也不会产生重复节点。如果还想更快可以临时关掉约束全部导入后再CREATE CONSTRAINT重建索引但要注意这会让重跑测试变得不安全我这边的做法是保留约束、调大批次优先保证可重复执行。5.4 实体识别把“干咳”拆成“干”和“咳”实体错位导致答案全错现象问“长期干咳是什么病”识别出的症状实体是“咳”而不是“干咳”查询结果指向了一堆不相关的咳嗽类型。原因jieba默认词库里“干咳”不是词它把单字拆开当成两个独立词处理。自定义词典里虽然加了“干咳”但没有控制词频字典优先级没覆盖到。解决在自定义词典里把“干咳”词频调高比如写成“干咳 100 nz”。jieba.load_userdict对相同前缀词的切分优先级说到底还是看词频。另一个方法是在实体识别后加一次“最长词优先”的合并遍历候选实体集合看某个词是否作为另一个更长词的后缀是的话就丢弃短词。这两招配合“干咳”和“头痛”“胃痛”这类复合症状基本都能保住。6. 评估与调优用离线指标和Bad Case把系统调到可以交付6.1 用Precision1把问答效果变成可比较的数字问答系统最怕“感觉还行”四个字。我一般会攒一份100条左右的评测集每一条包含问句、期望意图、期望实体集合。跑完一轮后统计实体命中率和意图命中率作为每次迭代的基准线。def evaluate(test_set, qa_pipeline): total len(test_set) intent_hit 0 entity_hit 0 for case in test_set: result qa_pipeline.run(case[question]) if result.intent case[expected_intent]: intent_hit 1 if set(result.entities) set(case[expected_entities]): entity_hit 1 print(fintent_acc{intent_hit/total:.2f}, entity_acc{entity_hit/total:.2f})评测集每加一条Bad Case就跑一遍这个脚本看准确率有没有变化。实体命中率是更敏感的指标因为实体错了后面所有环节都会跟着错。意图准确率到0.85以上实体准确率到0.9以上这个问答系统才有交付给业务方试用的底气。6.2 上线前必须补的功课同义词表扩充、知识重建、兜底回答跑通只是第一步真正见得光的系统至少还要做三件事。第一把评测集里所有“实体识别对但归一化错”的样本全部补齐到同义词表不要今天加两条明天加三条。第二建一个定时任务每月或每季度重跑一遍数据清洗和入库脚本让新药品、新说明书能进图否则知识会随时间过期。第三给unknown意图和空答案设计兜底话术让用户感觉系统“知道自己的边界”而不是怒怼一个“查无此人”。这套系统的价值不在于模型多深而在于工程结构清晰。把实体、意图、模板三个模块解耦之后换模型、换数据都只影响局部血泪经验告诉我这比追求单一环节的“更聪明”更重要。希望帮到你。本文还有配套的精品资源点击获取