简介：面向医疗知识图谱与问答系统入门者的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查询里，多跳关系的表达几乎是白话。

数据模型按节点和关系拆，我的设计如下：

关系类型要表达语义，而不是模糊的“相关”。比如疾病-[: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协议，不是HTTP：

from neo4j import GraphDatabase class Neo4jClient: def __init__(self, uri, user, password): self.driver = GraphDatabase.driver(uri, auth=(user, password)) def query(self, cypher, params=None): 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", encoding="utf-8") print(df.head(), df.shape) # 先看字段和规模 # 去重：同一标准名只保留一个 df = df.drop_duplicates(subset=["name_std"], keep="first") # 标准化：统一科室名，去掉前后空格和全角空格 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", index=False)

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:i+100] result = session.run(insert_cypher, rows=batch) 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 意图识别：先定规则，后用模型

意图识别的目标是判断用户问的是哪一类问题。我按医疗问答的高频场景把意图分成五类：

在数据量不足的初期，规则就能覆盖主要问法。每个意图配几个关键词和正则在问句里找。规则识别不到的，再考虑上分类模型，比如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 用Precision@1把问答效果变成可比较的数字

问答系统最怕“感觉还行”四个字。我一般会攒一份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(f"intent_acc={intent_hit/total:.2f}, entity_acc={entity_hit/total:.2f}")

评测集每加一条Bad Case，就跑一遍这个脚本，看准确率有没有变化。实体命中率是更敏感的指标，因为实体错了，后面所有环节都会跟着错。意图准确率到0.85以上，实体准确率到0.9以上，这个问答系统才有交付给业务方试用的底气。

6.2 上线前必须补的功课：同义词表扩充、知识重建、兜底回答

跑通只是第一步，真正见得光的系统至少还要做三件事。第一，把评测集里所有“实体识别对但归一化错”的样本全部补齐到同义词表，不要今天加两条明天加三条。第二，建一个定时任务，每月或每季度重跑一遍数据清洗和入库脚本，让新药品、新说明书能进图，否则知识会随时间过期。第三，给unknown意图和空答案设计兜底话术，让用户感觉系统“知道自己的边界”，而不是怒怼一个“查无此人”。

这套系统的价值不在于模型多深，而在于工程结构清晰。把实体、意图、模板三个模块解耦之后，换模型、换数据都只影响局部，血泪经验告诉我这比追求单一环节的“更聪明”更重要。希望帮到你。

本文还有配套的精品资源，点击获取