简介：面向Python开发者与知识图谱初学者的课程设计项目，完整覆盖医疗知识图谱的构建与基于图谱的问答实现。项目从构建简单知识图谱入手，进而搭建覆盖疾病、症状、药品、科室等实体的医疗知识图谱，并基于该图谱实现一个无需训练的规则式对话系统，推理速度快，便于快速掌握知识图谱落地流程；也明确指出其难以分析任意输入、输出灵活度有限的局限，为结合深度学习模型增强交互能力留下方向。压缩包共58个文件，以Python源码、txt数据字典、JSON图谱数据、PNG/JPG截图和Word设计报告为主，整体约20.13MB，并按知识图谱基础、医疗图谱构建、问答机器人三个模块清晰组织，便于对照学习与二次开发。已有2568人学习，适合作为课程设计、毕业设计预研或知识图谱入门练手的完整参考，可从中获取从0到1构建医疗知识图谱、封装问答接口的完整思路，以及可直接运行的项目源码和设计文档。
1. 从病历文本到可问答的医疗知识图谱，差的不是算法而是流程
医疗知识图谱问答系统，本质上是一条流水线：先把分散在结构化表格、病历文本、医学指南里的知识抽成“实体-关系-实体”三元组，再把这些三元组存入图数据库，最后通过意图识别和查询改写把自然语言变成图查询。基于Python做这套系统，难点不在某个单点技术上，而在实体对齐的精度、查询模板的覆盖率和答案生成的可靠性上。
适合读这篇文章的人，是已经会Python基础语法、想用知识图谱做垂直领域问答的工程师，或者需要交付医疗NLP相关Demo的团队。你不需要懂医疗，但需要有耐心处理脏数据。真正的医疗问答系统不是用大模型代替全部模块，而是让大模型只做它擅长的事——实体归一和答案润色。有一个反直觉的结论先放在这里：在医疗问答场景里，用Neo4j存图并用Cypher做查询，比直接用向量数据库召回的效果更稳定，因为它能严格约束关系的合法性，避免模型胡乱联想，这恰是医疗场景最看重的可解释性。
2. 医疗知识图谱的本体设计与Python建模
2.1 医疗实体和关系的分层定义
知识图谱的起点不是代码，而是本体。医疗领域建议采用六类实体：疾病（Disease）、症状（Symptom）、药物（Drug）、检查（Test）、科室（Department）、食物（Food）。关系则按临床逻辑设定：疾病-症状属于“表现为”，疾病-药物属于“宜用”，药物-食物属于“忌口”，疾病-检查属于“需做”。三元组的设计要保证每一条都能被病人或医生读懂，也就是（头实体, 关系, 尾实体）的三元组语义足够直白。
写本体时不要用太复杂的层级，医疗知识图谱只显示25个标签都能把前端页面拖垮，更不要说上百种关系类型。实操中我一般把关系控制在8种以内，把属性挂在节点上而不是做成关系。例如“阿莫西林”节点的属性包括剂量、用法、禁忌人群，而不是把“剂量”建成长尾关系，这样查询时可以用WHERE条件过滤，性能会好很多。
2.2 用Python将结构化数据转换成三元组
医疗数据常用格式是CSV或Excel，包含疾病名称、症状描述、推荐用药、注意事项等字段。先写一个数据清洗函数，把空值和重复项处理掉，再用pandas逐行生成三元组。以下是可复用的数据预处理与三元组构建代码：
import pandas as pd def load_and_clean(path): df = pd.read_csv(path) df = df.dropna(subset=['disease']) df['symptom'] = df['symptom'].str.replace(' ', '').str.split('、') df['drug'] = df['drug'].str.replace(' ', '').str.split('、') records = [] for _, row in df.iterrows(): disease = row['disease'] if not disease: continue for sym in row['symptom']: records.append((disease, '表现为', sym)) for drug in row['drug']: records.append((disease, '宜用', drug)) return records triples = load_and_clean('medical_data.csv') print(f'生成三元组数量: {len(triples)}') print(triples[:10])这段代码利用str.split('、')把症状和药物字段统一拆成列表，因为中文医学数据里顿号是使用频率最高的分隔符。dropna(subset=['disease'])把疾病名称为空的行直接丢弃，避免后续建图时出现悬空节点。三元组的生成在内存中完成，数据量超过十万条时可改用生成器逐个产出，避免内存溢出。需要考虑的是，不同来源的同一种疾病名称可能写法不同，“高血压”和“高血压病”应通过别名表归一，否则图谱会出现大量冗余节点。在生成三元组的同时维护一个别名映射字典，是性价比最高的实体对齐方案。
2.3 从Python到Neo4j的批量导入方案
目前知识图谱构建领域比较成熟的方式是用Neo4j作为存储引擎，Cypher作为查询语言。对Python开发者来说，py2neo是最快的上手方式，但对大批量数据推荐使用neo4j官方驱动配合UNWIND批量写入。不要一条一条执行CREATE语句，那效率极低。以下是用官方驱动批量写入的代码：
from neo4j import GraphDatabase class MedicalGraph: def __init__(self, uri, user, password): self.driver = GraphDatabase.driver(uri, auth=(user, password)) def create_graph(self, triples): with self.driver.session() as session: session.execute_write(self._create, triples) @staticmethod def _create(tx, triples): query = """ UNWIND $triples AS t MERGE (h:Entity {name: t.h}) MERGE (r:Entity {name: t.r}) MERGE (h)-[rel:RELATION {type: t.rel}]->(r) SET rel.type = t.rel """ tx.run(query, triples=[ {'h': h, 'rel': rel, 'r': r} for h, rel, r in triples ]) graph = MedicalGraph('bolt://localhost:7687', 'neo4j', 'password') graph.create_graph(triples)批量写入的核心在于UNWIND $triples AS t，把Python列表整体传给Cypher，只做一次网络往返，性能远高于循环插入。MERGE而不是CREATE的关键作用是，重复的三元组不会被重复创建，实现了幂等写入。关系上使用{type: t.rel}属性记录关系类型，而不是把关系类型直接写死在查询里，原因是一旦你后续需要新增关系类型，可以只改Python数据源而不用改Cypher模板。对百万级三元组的导入，合理控制批大小（每批5000条）通常能在两分钟内完成。
3. 知识问答系统的核心Pipeline：从问句到Cypher
3.1 医疗问答的意图识别与实体抽取
问答系统的第一步是把用户问题拆成“意图”和“实体”。意图决定了查询模板，实体决定了查询参数。医疗场景中意图一般是以下几种：症状查疾病（“头疼应该看什么科”）、疾病查用药（“高血压吃什么药”）、疾病查症状（“糖尿病有哪些表现”）、药物查禁忌（“阿莫西林不能和什么一起吃”）。意图识别可以采用规则和轻量模型结合的方式。
实体抽取推荐三种工具配合使用：pkuseg做分词、自定义词典做术语匹配、正则做症状量化词识别。医疗词典的质量直接决定抽取效果，需要收集科室名、常用药名、疾病名和症状名。注意一个问题：jieba对医学专业词汇的切分效果很差，“阿莫西林胶囊”会被切成“阿莫西林/胶囊”，所以需要在分词前强制加载自定义词典。这里给出一个结合正则和词典的抽取器：
import re from pkuseg import pkuseg seg = pkuseg(model_name='medicine', postag=False) class MedicalExtractor: def __init__(self, symptom_dict, drug_dict, disease_dict): self.symptom_dict = symptom_dict self.drug_dict = drug_dict self.disease_dict = disease_dict self.symptom_re = re.compile('|'.join(sorted(symptom_dict, key=len, reverse=True))) def extract(self, question): symptom_hits = self.symptom_re.findall(question) words = seg.cut(question) drug_hits = [w for w in words if w in self.drug_dict] disease_hits = [w for w in words if w in self.disease_dict] return { 'symptom': symptom_hits, 'drug': drug_hits, 'disease': disease_hits }正则匹配先按长度倒序排列，保证“上呼吸道感染”这种长词优先于“感染”被匹配，避免长实体被短实体截胡。pkuseg的model_name='medicine'如果环境中没有该模型，会默认下载并缓存，所以在第一次运行时需要保持网络畅通。层级实体存在嵌套关系时，比如“肺部感染”既是疾病也包含“感染”这个词，基于词典的方式会同时命中，处理方式是优先选择最长匹配并合并别名。
3.2 基于规则模板的Cypher查询生成
实体抽取完成后的下一步，是根据意图把实体映射成Cypher查询。规则模板是生产环境中用的最多的方案，因为医疗问答对准确率要求高，规则可以做到每个查询都有据可查。每一类意图对应一个模板，模板里的槽位由实体填充。若多个实体同时命中，比如用户既提到症状又提到疾病，优先级排序应该是疾病 > 药物 > 症状，因为疾病通常是查询的主体。
一个典型的问题是“高血压患者咳嗽应该吃什么药”，抽取结果里同时有疾病“高血压”和症状“咳嗽”，此时正确的做法是先用疾病去约束，再用症状去过滤。实体抽取和模板填充可能产生歧义，所以整体流程必须支持方便的调试查看。给出一个模板匹配的核心类：
class CypherGenerator: def __init__(self): self.templates = { 'symptom_to_disease': "MATCH (d:Disease)-[:表现为]->(s:Symptom) WHERE s.name = $symptom RETURN d.name", 'disease_to_drug': "MATCH (d:Disease)-[:宜用]->(m:Drug) WHERE d.name = $disease RETURN m.name", 'disease_to_symptom': "MATCH (d:Disease)-[:表现为]->(s:Symptom) WHERE d.name = $disease RETURN s.name", 'drug_to_food': "MATCH (d:Drug)-[:忌口]->(f:Food) WHERE d.name = $drug RETURN f.name" } def generate(self, intent, entities): if intent == 1 and entities.get('symptom'): return self.templates['symptom_to_disease'], {'symptom': entities['symptom'][0]} if intent == 2 and entities.get('disease'): return self.templates['disease_to_drug'], {'disease': entities['disease'][0]} # 更多意图分支... return None, None参数使用$symptom而不是字符串拼接，是安全底线，可以防止Cypher注入。这些模板看起来简单，但在生产环境里至少要覆盖40个以上的查询变体，因为用户不会按照模板说话，会表达为“咳嗽该挂什么科”而不是“咳嗽的症状”，所以模板要同时覆盖陈述语气和疑问语气。
3.3 有界子图查询与答案格式化
当用户问题涉及多个实体时，简单的模板已经不够，需要做有界子图查询。比如“高血压病人吃阿莫西林有什么风险”，要查的是疾病、药物、以及它们之间经由症状或检查建立的关联路径。Cypher中的MATCH p=(d:Disease)-[*1..3]-(m:Drug)允许指定路径长度为1到3跳，但需要注意这种查询成本很高，必须限定模式。
从Neo4j返回的数据是嵌套结构，不能直接展示给用户。答案的格式化规则是：只返回实体的name属性和关系type，丢弃内部id和标签。更进一步，需要做结果的融合，把同一实体的不同表达合并成用户可读的短句。例如查询结果里有“头痛”和“头疼”，如果图谱中两者被归一为同一个实体，则直接去重；如果未归一，就要在查询层做字符串相似度合并。
def format_answer(records): answer = [] for record in records: rel = record['rel'] target = record['target'] answer.append(f"{target}（{rel}）") unique_answer = list(dict.fromkeys(answer)) return "，".join(unique_answer[:5]) if unique_answer else "暂未找到匹配信息"这里限定最多返回5个结果，原因有两个：一是医疗答案超过5条用户就读不过来了，二是减少前端渲染压力。答案展示格式用的是“实体（关系）”的模板，让用户知道为什么返回这个结果。在真实系统中，答案格式化模块和前端是解耦的，后端返回JSON数组，前端负责渲染卡片，这样同一套后端可同时服务Web端和推荐系统接口。
4. 医疗问答系统的再检索与交互优化适配
4.1 问答正确率的离线验证方法
医疗问答系统上线前不做测验就是拿用户当小白鼠。一个可复现的验证方法是构建一套“问题-答案-支撑路径”的三元组测试集，每一条都对应图谱中真实存在的路径。验证时跑三个指标：实体命中率、Cypher生成正确率、答案文本匹配率。答案是A，Cypher生成错了但碰巧答案对，也算错，因为Cypher错了意味着可解释性断了。Python脚本里用subprocess直接调用cypher-shell比对查询结果，是离线回归最简单的方式。
def evaluate(test_cases, generator, graph): correct = 0 for case in test_cases: query, params = generator.generate(case['intent'], case['entities']) if query is None: continue result = graph.run_query(query, params) if set(result) == set(case['expected']): correct += 1 return correct / len(test_cases)测试集要覆盖三种典型问题：常见症状问疾病、疾病问用药、药物问食物禁忌。每种至少20条，并且要故意加入口语化表达，比如“感冒了喝板蓝根行不行”这类不在标准模板内的问题。这种回归测试在每次修改实体词典或查询模板后必须全量跑一遍，防止改一个bug引出三个新bug。
4.2 图谱更新时的索引重建策略
医疗知识图谱不是建完就完事的，药品说明说改了就改，医院科室调整了也要改。图谱增量更新的常见做法是：新增节点时直接MERGE，删除过期关系时使用MATCH ... DETACH DELETE。但有一个坑是：旧版关系可能仍被历史查询缓存命中，导致前端展示过期数据。需要在更新完后重建全文索引。
CREATE INDEX disease_name IF NOT EXISTS FOR (n:Disease) ON (n.name) CREATE INDEX drug_name IF NOT EXISTS FOR (n:Drug) ON (n.name)Neo4j的索引不是自动全量创建的，而是启动时就存在，但对于后续大批量插入的数据，需要手动触发重建。有时前端只显示25个标签的原因不是数据量不够，而是默认的label显示配置限制。解决方案有两个：后端做聚合前端分页，或者在生成结果时预先分组再推送。注意对于Neo4j迁移或升级的场景，导出的dump文件需要重新执行一次CREATE INDEX，确保新实例上的查询性能。
4.3 查询性能的瓶颈定位与参数调整
医疗知识图谱数据量到几十万节点后，查询响应时间会明显上升。可以先检查Cypher执行计划，用EXPLAIN看是否走了索引：
cypher-shell "EXPLAIN MATCH (d:Disease)-[:宜用]->(m:Drug) WHERE d.name = '高血压' RETURN m.name"命中的输出会显示NodeByLabelScan带索引提示，而不命中的会显示NodeByLabelScan全表扫描，这是最常见的慢查询来源。如果WHERE条件不是等值而是包含模糊匹配，比如CONTAINS，索引会失效，得引入STARTS WITH甚至全文索引。图数据库中Python驱动连接池的默认配置也需要调整，max_connection_pool_size保持默认50对高频问答够用，但connection_timeout建议调短，否则在Neo4j负载高时Python侧会堆积大量等待线程。
5. 医疗问答中医疗数据语义歧义的处理技巧
医疗问答和通用问答最不一样的地方，是同一个词在不同上下文中的含义可能完全不同。“麻疹”既可以是一种疾病，也可以是症状“皮疹”在特定语境下的指代。处理这种歧义，首选方法是基于实体类型和关系约束的语义消歧。有一个比较稳的技巧：一个实体在问题中同时命中两个类型时，优先保留与查询意图同类型的那一个，如果意图是疾病查询，那么“麻疹”就取疾病类型。给出一个简单而有效的消歧函数：
def disambiguate(entity, intent, types): if len(types) < 2: return types[0] if types else None primary_type = {'disease': ['疾病'], 'drug': ['药物']}.get(intent_name(intent), None) if primary_type and primary_type in types: return primary_type return types[0]医疗实体抽取中，对否定词和程度副词的处理也需要单独处理。“没有头痛”“不怎么咳嗽”这类包含否定含义的句子，如果直接把“头痛”当作正症状入查询，答案会完全错位。要在抽取阶段同时输出否定标记，查询模板层把否定标记拼成NOT EXISTS子句，这才算一个完整的交互。这一系列技巧配上离线回归测试，就能让系统的错误类型从“答案离谱”收敛为“查询无结果”，这种收敛在医疗场景里意义非常具体——宁可不答，不能错答。
本文还有配套的精品资源，点击获取
