简介这是一份基于Python实现的医学知识图谱问答系统源码项目面向医学信息处理学习者、医疗AI初学者以及需要快速搭建问答原型的开发者。项目覆盖数据爬取、医疗词典构建、实体对齐、问题分类、答案搜索等关键环节能够帮助读者理解医学知识图谱从构建到问答落地的完整流程也可作为毕业设计或课题改造基础。压缩包内共32个文件以8个Python源码文件为核心配合医疗实体词典、问答规则、界面图片和演示文稿等辅助材料整体包体约49.2MB目录结构较为清晰便于按模块钻研。目前已有433人学习下载。通过研读源码可掌握医疗数据清洗、图谱存储、问句解析与答案路由等实现思路复用其中的分词与检索设计节省自研时间提升医学问答系统开发效率。1. 医疗问答为什么需要知识图谱一次拆解 QASystemOnMedicalKG当用户问“感冒咳嗽一周还胸闷到底该做哪些检查”时常见的关键词搜索引擎会返回一堆百科链接而基于FAQ的客服机器人只能命中完全匹配的搭话。要真正让机器理解疾病、症状、检查、药品之间的约束关系应该把答案建模成图疾病节点连接症状节点和检查节点再通过“推荐”关系指向药品。本资源就是一个把该思路落地的Python医学知识图谱问答系统代码只包含8个Python文件和若干词表却完整覆盖了数据采集、分词、图谱构建、问句分类、Cypher检索和对话回显。对正在做医疗助手、病历结构化或知识图谱应用的工程师来说它是一套能直接运行的最小闭环实现比单纯看某篇论文里的算法更容易调通。2. 数据到图谱爬虫、最大分词与 build_medicalgraph.py打开源码压缩包首先看到的是 data_spider.py、prepare_data、build_data.py 以及 data.json。这几个文件决定了知识图谱里的数据从哪里来、以什么形式进入图库。2.1 原始数据与 data_spider.py从 HTML 到 medical.jsondata_spider.py 是一个Python爬虫脚本负责从医学百科类页面抓取疾病、药品、食物、检查、科室等实体并把实体之间的关系整理成 JSON 数组保存下来。这样做的目的很直接先把半结构化的网页转成统一的 data.json后续建图时只需要读这一个文件不会被页面改版干扰。# 简化自 data_spider.py抓取单个疾病页面的核心字段 def parse_disease(url): resp requests.get(url, headersheaders, timeout10) soup BeautifulSoup(resp.text, html.parser) disease { name: soup.select_one(h1).text.strip(), symptom: [a.text for a in soup.select(.symptom a)], drug: [a.text for a in soup.select(.drug a)], check: [a.text for a in soup.select(.check a)], department: soup.select_one(.department).text.strip() } return disease这里需要注意几个参数timeout防止单页卡死整个爬虫.select_one在页面缺失该标签时返回None后续保存前要补if disease[department] else 其他之类的默认值。源码里往往把这些容错逻辑压缩在一两行内真正放到生产环境时会建议增加重试和日志。抓取结果统一写入medical.json结构是“疾病作为根节点内部挂数组”的形式。数据清洗脚本prepare_data和build_data.py会接着将 JSON 拆成节点列表和关系列表很多同学一开始只看得到 build_medicalgraph.py 里的建图操作容易忽略前面的清洗步骤导致图里出现大量空名字节点。2.2 dict 词表与 max_cut.py以最大匹配为医学术语分界源码里的 dict 目录包含 8 个纯文本词表这是整个医学知识图谱问答系统能跑起来的关键资产。它们不是给人看的注释而是喂给分词器和分类器用的领域词典。文件名实体类型典型词条disease.txt疾病感冒、高血压、冠状动脉粥样硬化symptom.txt症状咳嗽、胸闷、发热drug.txt药品阿莫西林、布洛芬check.txt检查项目血常规、胸片、CTdepartment.txt科室呼吸内科、心内科food.txt食物苹果、生姜、绿豆producer.txt生产厂家华北制药、同仁堂deny.txt否定触发词没有、不是、无词表会被加载为 Python 的set然后在max_cut.py里做正向最大匹配分词。为什么不直接依赖 jieba因为医学复合词太长比如“弥漫性泛细支气管炎”通用词典容易把它切成“弥漫性/泛/细/支气管炎”导致后端的实体识别对不上图谱节点。用自定义词表约束切分路径既能让分词结果稳定也让“感冒”这类短词不被错误合并。# max_cut.py 中常见的函数实现正向最大匹配 class MaxCut: def __init__(self, word_dict, max_window4): self.word_dict word_dict self.max_window max_window def cut(self, sentence): result, start [], 0 while start len(sentence): matched sentence[start] for end in range(start self.max_window, start, -1): word sentence[start:end] if word in self.word_dict: matched word break result.append(matched) start len(matched) return resultmax_window4表示一次最多尝试匹配 4 个汉字超过这个长度就按单字切。参数不是越大越好太大的窗口会显著增加每轮扫描次数在 8 万词表规模下窗口 4 到 6 是常见折中。分词器得到的实体词会传递到build_data.py用于统计实体共现和生成问答所需的候选词表。2.3 build_medicalgraph.py把三元组写进 Neo4j 的批量方式图谱构建脚本build_medicalgraph.py是整个源码中和后端最接近的一部分。它读取medical.json用 py2neo 向本地 Neo4j 写入疾病、症状、检查、药品等节点以及疾病与它们之间的“HasSymptom”“RecommendDrug”等关系。为了不让导入过程慢到无法忍受建图脚本通常采用“先建节点再批量建关系”的两段式事务。# 简化自 build_medicalgraph.py批量创建疾病和症状节点 from py2neo import Graph, Node, Relationship graph Graph(bolt://localhost:7687, auth(neo4j, neo4j)) def create_nodes(tx, entities, label): nodes [] for name in entities: node Node(label, namename) tx.merge(node, label, name) nodes.append(node) return nodes tx graph.begin() disease_nodes create_nodes(tx, disease_names, Disease) symptom_nodes create_nodes(tx, symptom_names, Symptom) graph.commit(tx)这里tx.merge比tx.create更适合重复导入。merge(node, Disease, name)的语义是如果存在 name 属性相同的节点就复用否则创建。这样即使脚本被多次执行也不会堆积重复实体。注意 merge 需要为目标节点提供唯一键约束在 Neo4j 中建议提前执行CREATE CONSTRAINT FOR (d:Disease) REQUIRE d.name IS UNIQUE;。关系写入同样建议放在独立事务里。不要在一个事务中先合并节点再立刻合并关系因为关系 merge 要求两个端点都已持久化如果在同一事务里py2neo 会承担额外的节点解析开销数据量到几万条时速度会明显下降。作者提供的kg_route.png和graph_summary.png展示了最终图谱在 Neo4j Browser 里的可视化和统计面板那是确认建图结果最直观的方式。2.4 如何验证图谱建立成功图谱构建完成后不能只看控制台没有报错就收工。我一般会在 Neo4j Browser 里执行下面两类查询MATCH (n:Disease) RETURN n.name LIMIT 20;MATCH (d:Disease)-[:HAS_SYMPTOM]-(s:Symptom) WHERE d.name 感冒 RETURN s.name;第一句验证节点是否写入第二句验证关系是否指向正确的症状集合。如果第二句返回空常见原因不是数据没导入而是关系类型名不匹配例如脚本里写的是HAS_SYMPTOM但查询时写成了HAS_SYMTPOM。检查方法是在浏览器中执行CALL db.relationshipTypes();它会列出所有已存在的关系类型直接和源码里的字符串比对即可。3. 问句理解层question_classifier.py 与 question_parser.py 的实现知识图谱建完之后真正决定用户体验的是问句理解模块。question_classifier.py负责判断用户提问里的实体和意图question_parser.py负责把判断结果翻译成图谱查询模板。这个组合类似于自然语言理解中的“槽位填充 意图识别”。3.1 意图分类为什么要用“词典规则”而不是“模型”很多第一次做问答系统的同学会直接上一个BERT二分类模型但在这个项目场景里训练数据只有百余条人工构造的问句模型很难覆盖医学实体组合的庞大多样性。源码采用的方案是纯词典和规则先读取 dict 下的词表再对问句做包含匹配。# 简化自 question_classifier.py检查问句中是否出现词表实体 def check_words(self, vocabulary, question): hit None for word in vocabulary: if word in question: hit word break return hit这种方式的优点是可解释性和执行效率都很好。缺点也明显如果用户说“我没咳嗽”词表会照常匹配到“咳嗽”并把症状实体抽出后续答案就会把疾病与咳嗽关联起来与用户本意相反。后面的deny.txt词表就是用来补救这个问题的分类器在拿到实体后还要扫描问句里是否出现否定词若有则把对应关系反转或直接置空。3.2 分类器内部结构实体识别与意图打分一个典型的问句分类结果包含两部分args是抽取到的实体intent是归类后的查询类型。源码中意图的分类规则通常是“实体类型 关键词触发词”的组合。# 示意代码根据触发词赋予意图 def classify(self, question): entities { disease: self.check_words(self.disease_dict, question), drug: self.check_words(self.drug_dict, question), symptom: self.check_words(self.symptom_dict, question) } if entities[disease] and (吃什么药 in question or 用药 in question): return {intent: DISEASE_DRUG, args: entities} if entities[disease] and 症状 in question: return {intent: DISEASE_SYMPTOM, args: entities} return {}实际工程里意图类型远不止两种。源码会覆盖以下常见问答场景intent用户问法示例图谱查询目标disease_symptom感冒有什么症状疾病 - 症状disease_cause高血压的病因是什么疾病 - 病因disease_check肺炎需要做什么检查疾病 - 检查disease_drug治疗胃炎吃什么药疾病 - 药品disease_food糖尿病患者能吃什么疾病 - 食物disease_department看癫痫挂哪个科疾病 - 科室这里的“意图打分”不是用概率模型打分而是通过触发词多少做简单投票比如问句同时包含“怎么治”和“药”就优先落到disease_drug。把触发词列在源码顶部后续调整业务口径时只改列表不改逻辑是这套设计的优点。3.3 解析器把 intent 转换成一个可执行的模板question_parser.py的作用是把分类器输出转换成cypher查询语句的模板。它和直接写查询的区别在于解析器只关心“要查什么”不关心“怎么连数据库”。# 示意代码question_parser.py 的模板转换 class QuestionParser: def parser(self, classify_result): intent classify_result[intent] disease classify_result[args][disease] if intent DISEASE_SYMPTOM: sql MATCH (d:Disease)-[:HAS_SYMPTOM]-(s:Symptom) WHERE d.name{0} RETURN s.name.format(disease) return sql if intent DISEASE_DRUG: sql MATCH (d:Disease)-[:HAS_DRUG]-(drug:Drug) WHERE d.name{0} RETURN drug.name.format(disease) return sql这里用字符串format拼接名字在这个源码场景里是可行的因为disease来自本地词典白名单而不是直接输入的用户文本。但如果你打算把这个系统开放成公网 API我建议把拼接改成 py2neo 的参数化查询避免用户用}) RETURN 1 //之类的内容污染 Cypher 语句。第 4 章会给出参数化的写法。4. 答案检索与对话循环answer_search.py 和 chatbot_graph.py问句解析成 Cypher 模板后真正执行图查询的是answer_search.py。这一段是系统里最接近“数据库驱动”的代码也是回答正确率的核心保障。4.1 Cypher 模板生成将问句模板映射为图查询answer_search.py中会接收上一步生成的查询语句连接 Neo4j 并执行随后把结果整理为自然语言格式。它的接口设计往往是这样# answer_search.py 核心结构 from py2neo import Graph class AnswerSearcher: def __init__(self): self.graph Graph(bolt://localhost:7687, auth(neo4j, neo4j)) def search_disease_symptom(self, disease): sql MATCH (d:Disease)-[:HAS_SYMPTOM]-(s:Symptom) \ WHERE d.name $name RETURN s.name result self.graph.run(sql, namedisease).data() return [item[s.name] for item in result]与第 3 章拼接字符串不同这里采用$name参数。py2neo 会把参数作为独立数据传递给 Neo4j不会与CYPHER语句结构混在一起安全性和可读性都更好。self.graph.run()返回的是一个游标对象.data()会一次性加载为列表字典适合问答系统这种小结果集场景。如果查找结果为空answer_search.py会返回一个默认话术比如“暂时没有找到关于该疾病的相关信息”。千万不要让这个空结果一路传到前端变成空白用户会以为是系统崩溃。4.2 查询边界与性能考量Cypher 查询虽然写起来简单但在医学图谱里容易踩两个坑关系方向写反以及没有限制返回条数。源码图谱中关系方向的约定通常是从疾病指向其他实体所以在回答“发烧会提示哪些疾病”这类反方向问题时需要显式写成MATCH (s:Symptom)-[:HAS_SYMPTOM]-(d:Disease)也就是把起始点换到症状节点。查询语义Cypher 起点 - 终点注意点疾病症状(d:Disease)-[:HAS_SYMPTOM]-(s:Symptom)疾病在左症状可能疾病(s:Symptom)-[:HAS_SYMPTOM]-(d:Disease)方向相反疾病科室(d:Disease)-[:DEPARTMENT]-(dep)若关系类型不同需替换在执行耗时上一般只在疾病、症状等高频实体上建立name的唯一约束查询路径很短毫秒级返回没有问题。如果后续图谱规模超过几十万节点可以在关系属性上增加索引并在 Cypher 末尾追加LIMIT 20避免一次性拉回上千条无关答案。4.3 机器人的主循环一次问答是如何串起来的chatbot_graph.py是让整个系统以一个对话机器人的形式跑起来的入口。它把前面三个 Python 文件串成流水线接收用户问题调用分类器解析器生成查询检索器去图库拿答案最后格式化输出。源码里通常还会带一个基于input()的循环方便在命令行直接验证。# chatbot_graph.py 主流程 from question_classifier import QuestionClassifier from question_parser import QuestionParser from answer_search import AnswerSearcher class ChatBotGraph: def __init__(self): self.classifier QuestionClassifier() self.parser QuestionParser() self.searcher AnswerSearcher() def chat(self, question): classify_result self.classifier.classify(question) if not classify_result: return 我暂时无法回答该问题你可以换一种问法 cypher self.parser.parser(classify_result) answers self.searcher.search(cypher) return \n.join(answers)这段代码后面的while True循环本身不复杂值得留意的是它没有任何上下文状态。也就是说用户问“感冒呢”系统是不知道上一轮在聊高血压的。如果生产环境要接微信客服必须在外面套一层会话记忆把上一轮实体传入下一次分类器而不是改chatbot_graph.py的内部逻辑。5. 部署调优与一个常用扩展把问答系统包装成 HTTP 接口这段分享一些把源码跑通并拓展成服务的经验。源码压缩包里有__pycache__下的 pyc 文件标注是cpython-36说明原运行环境是 Python 3.6。现代环境建议直接用 Python 3.8 以上版本pyc 文件跨版本无法复用解释器会在执行时重新生成。5.1 环境配置与依赖版本坑安装依赖时不要一把梭conda create -n medical_kg python3.8 -y conda activate medical_kg pip install py2neo2021.2.3 jieba requests beautifulsoup4 flask这里把 py2neo 版本锁到 2021.2.3是为了兼容 Neo4j 4.x。如果你本机装的是 Neo4j 5.x推荐用更高的 py2neo 4.x 版本如果使用过老的 py2neo 连接新 Neo4j最常见报错是The client is unauthorized due to authentication failure这不一定是密码错而是版本间的握手机制不匹配。启动 Neo4j 后用CYPHER 5语法在浏览器里执行RETURN 1验证版本。5.2 图谱构建的提速与安全重启开发过程中反复跑build_medicalgraph.py会留下大量测试产生的脏关系。清库注意用DETACH DELETEMATCH (n) DETACH DELETE n;DETACH DELETE会先删除所有关系再删节点比逐条删关系快很多。清完库再跑导入脚本能避免节点属性冲突导致 merge 失效。如果图库已经不小可以在build_medicalgraph.py里把graph.begin()和graph.commit(tx)包裹的数据条数设为 2000 条左右事务太大会导致 Neo4j 堆内存溢出事务太慢又拖慢导入速度。5.3 用 Flask 将 chatbot_graph 包装成 API要让这个基于命令行的问答系统对接外部前端直接改__main__里的循环并不优雅。我一般会新建一个app.py把ChatBotGraph实例化一次然后暴露 POST 接口。from flask import Flask, request, jsonify from chatbot_graph import ChatBotGraph app Flask(__name__) kg ChatBotGraph() app.route(/qa, methods[POST]) def qa(): payload request.get_json(forceTrue) question payload.get(question, ).strip() if not question: return jsonify({error: empty question}), 400 answer kg.chat(question) return jsonify({question: question, answer: answer}) if __name__ __main__: app.run(host0.0.0.0, port5000, debugFalse)这里把ChatBotGraph放在全局变量里避免每次请求都重新加载词典和连接 Neo4j。debugFalse防止调试器暴露到公网。启动后用curl验证curl -X POST http://127.0.0.1:5000/qa \ -H Content-Type: application/json \ -d {question:高血压患者适合吃什么食物}返回的 JSON 里answer字段就是图谱查询后的自然语言结果。如果中文出现乱码在 Flask 启动前执行import sys; sys.stdout.reconfigure(encodingutf-8)并确保终端使用 UTF-8 编码这是 Windows 环境下最容易踩的坑。本文还有配套的精品资源点击获取 基于知识图谱与Neo4j的医学问答系统实现解析
简介这是一份基于Python实现的医学知识图谱问答系统源码项目面向医学信息处理学习者、医疗AI初学者以及需要快速搭建问答原型的开发者。项目覆盖数据爬取、医疗词典构建、实体对齐、问题分类、答案搜索等关键环节能够帮助读者理解医学知识图谱从构建到问答落地的完整流程也可作为毕业设计或课题改造基础。压缩包内共32个文件以8个Python源码文件为核心配合医疗实体词典、问答规则、界面图片和演示文稿等辅助材料整体包体约49.2MB目录结构较为清晰便于按模块钻研。目前已有433人学习下载。通过研读源码可掌握医疗数据清洗、图谱存储、问句解析与答案路由等实现思路复用其中的分词与检索设计节省自研时间提升医学问答系统开发效率。1. 医疗问答为什么需要知识图谱一次拆解 QASystemOnMedicalKG当用户问“感冒咳嗽一周还胸闷到底该做哪些检查”时常见的关键词搜索引擎会返回一堆百科链接而基于FAQ的客服机器人只能命中完全匹配的搭话。要真正让机器理解疾病、症状、检查、药品之间的约束关系应该把答案建模成图疾病节点连接症状节点和检查节点再通过“推荐”关系指向药品。本资源就是一个把该思路落地的Python医学知识图谱问答系统代码只包含8个Python文件和若干词表却完整覆盖了数据采集、分词、图谱构建、问句分类、Cypher检索和对话回显。对正在做医疗助手、病历结构化或知识图谱应用的工程师来说它是一套能直接运行的最小闭环实现比单纯看某篇论文里的算法更容易调通。2. 数据到图谱爬虫、最大分词与 build_medicalgraph.py打开源码压缩包首先看到的是 data_spider.py、prepare_data、build_data.py 以及 data.json。这几个文件决定了知识图谱里的数据从哪里来、以什么形式进入图库。2.1 原始数据与 data_spider.py从 HTML 到 medical.jsondata_spider.py 是一个Python爬虫脚本负责从医学百科类页面抓取疾病、药品、食物、检查、科室等实体并把实体之间的关系整理成 JSON 数组保存下来。这样做的目的很直接先把半结构化的网页转成统一的 data.json后续建图时只需要读这一个文件不会被页面改版干扰。# 简化自 data_spider.py抓取单个疾病页面的核心字段 def parse_disease(url): resp requests.get(url, headersheaders, timeout10) soup BeautifulSoup(resp.text, html.parser) disease { name: soup.select_one(h1).text.strip(), symptom: [a.text for a in soup.select(.symptom a)], drug: [a.text for a in soup.select(.drug a)], check: [a.text for a in soup.select(.check a)], department: soup.select_one(.department).text.strip() } return disease这里需要注意几个参数timeout防止单页卡死整个爬虫.select_one在页面缺失该标签时返回None后续保存前要补if disease[department] else 其他之类的默认值。源码里往往把这些容错逻辑压缩在一两行内真正放到生产环境时会建议增加重试和日志。抓取结果统一写入medical.json结构是“疾病作为根节点内部挂数组”的形式。数据清洗脚本prepare_data和build_data.py会接着将 JSON 拆成节点列表和关系列表很多同学一开始只看得到 build_medicalgraph.py 里的建图操作容易忽略前面的清洗步骤导致图里出现大量空名字节点。2.2 dict 词表与 max_cut.py以最大匹配为医学术语分界源码里的 dict 目录包含 8 个纯文本词表这是整个医学知识图谱问答系统能跑起来的关键资产。它们不是给人看的注释而是喂给分词器和分类器用的领域词典。文件名实体类型典型词条disease.txt疾病感冒、高血压、冠状动脉粥样硬化symptom.txt症状咳嗽、胸闷、发热drug.txt药品阿莫西林、布洛芬check.txt检查项目血常规、胸片、CTdepartment.txt科室呼吸内科、心内科food.txt食物苹果、生姜、绿豆producer.txt生产厂家华北制药、同仁堂deny.txt否定触发词没有、不是、无词表会被加载为 Python 的set然后在max_cut.py里做正向最大匹配分词。为什么不直接依赖 jieba因为医学复合词太长比如“弥漫性泛细支气管炎”通用词典容易把它切成“弥漫性/泛/细/支气管炎”导致后端的实体识别对不上图谱节点。用自定义词表约束切分路径既能让分词结果稳定也让“感冒”这类短词不被错误合并。# max_cut.py 中常见的函数实现正向最大匹配 class MaxCut: def __init__(self, word_dict, max_window4): self.word_dict word_dict self.max_window max_window def cut(self, sentence): result, start [], 0 while start len(sentence): matched sentence[start] for end in range(start self.max_window, start, -1): word sentence[start:end] if word in self.word_dict: matched word break result.append(matched) start len(matched) return resultmax_window4表示一次最多尝试匹配 4 个汉字超过这个长度就按单字切。参数不是越大越好太大的窗口会显著增加每轮扫描次数在 8 万词表规模下窗口 4 到 6 是常见折中。分词器得到的实体词会传递到build_data.py用于统计实体共现和生成问答所需的候选词表。2.3 build_medicalgraph.py把三元组写进 Neo4j 的批量方式图谱构建脚本build_medicalgraph.py是整个源码中和后端最接近的一部分。它读取medical.json用 py2neo 向本地 Neo4j 写入疾病、症状、检查、药品等节点以及疾病与它们之间的“HasSymptom”“RecommendDrug”等关系。为了不让导入过程慢到无法忍受建图脚本通常采用“先建节点再批量建关系”的两段式事务。# 简化自 build_medicalgraph.py批量创建疾病和症状节点 from py2neo import Graph, Node, Relationship graph Graph(bolt://localhost:7687, auth(neo4j, neo4j)) def create_nodes(tx, entities, label): nodes [] for name in entities: node Node(label, namename) tx.merge(node, label, name) nodes.append(node) return nodes tx graph.begin() disease_nodes create_nodes(tx, disease_names, Disease) symptom_nodes create_nodes(tx, symptom_names, Symptom) graph.commit(tx)这里tx.merge比tx.create更适合重复导入。merge(node, Disease, name)的语义是如果存在 name 属性相同的节点就复用否则创建。这样即使脚本被多次执行也不会堆积重复实体。注意 merge 需要为目标节点提供唯一键约束在 Neo4j 中建议提前执行CREATE CONSTRAINT FOR (d:Disease) REQUIRE d.name IS UNIQUE;。关系写入同样建议放在独立事务里。不要在一个事务中先合并节点再立刻合并关系因为关系 merge 要求两个端点都已持久化如果在同一事务里py2neo 会承担额外的节点解析开销数据量到几万条时速度会明显下降。作者提供的kg_route.png和graph_summary.png展示了最终图谱在 Neo4j Browser 里的可视化和统计面板那是确认建图结果最直观的方式。2.4 如何验证图谱建立成功图谱构建完成后不能只看控制台没有报错就收工。我一般会在 Neo4j Browser 里执行下面两类查询MATCH (n:Disease) RETURN n.name LIMIT 20;MATCH (d:Disease)-[:HAS_SYMPTOM]-(s:Symptom) WHERE d.name 感冒 RETURN s.name;第一句验证节点是否写入第二句验证关系是否指向正确的症状集合。如果第二句返回空常见原因不是数据没导入而是关系类型名不匹配例如脚本里写的是HAS_SYMPTOM但查询时写成了HAS_SYMTPOM。检查方法是在浏览器中执行CALL db.relationshipTypes();它会列出所有已存在的关系类型直接和源码里的字符串比对即可。3. 问句理解层question_classifier.py 与 question_parser.py 的实现知识图谱建完之后真正决定用户体验的是问句理解模块。question_classifier.py负责判断用户提问里的实体和意图question_parser.py负责把判断结果翻译成图谱查询模板。这个组合类似于自然语言理解中的“槽位填充 意图识别”。3.1 意图分类为什么要用“词典规则”而不是“模型”很多第一次做问答系统的同学会直接上一个BERT二分类模型但在这个项目场景里训练数据只有百余条人工构造的问句模型很难覆盖医学实体组合的庞大多样性。源码采用的方案是纯词典和规则先读取 dict 下的词表再对问句做包含匹配。# 简化自 question_classifier.py检查问句中是否出现词表实体 def check_words(self, vocabulary, question): hit None for word in vocabulary: if word in question: hit word break return hit这种方式的优点是可解释性和执行效率都很好。缺点也明显如果用户说“我没咳嗽”词表会照常匹配到“咳嗽”并把症状实体抽出后续答案就会把疾病与咳嗽关联起来与用户本意相反。后面的deny.txt词表就是用来补救这个问题的分类器在拿到实体后还要扫描问句里是否出现否定词若有则把对应关系反转或直接置空。3.2 分类器内部结构实体识别与意图打分一个典型的问句分类结果包含两部分args是抽取到的实体intent是归类后的查询类型。源码中意图的分类规则通常是“实体类型 关键词触发词”的组合。# 示意代码根据触发词赋予意图 def classify(self, question): entities { disease: self.check_words(self.disease_dict, question), drug: self.check_words(self.drug_dict, question), symptom: self.check_words(self.symptom_dict, question) } if entities[disease] and (吃什么药 in question or 用药 in question): return {intent: DISEASE_DRUG, args: entities} if entities[disease] and 症状 in question: return {intent: DISEASE_SYMPTOM, args: entities} return {}实际工程里意图类型远不止两种。源码会覆盖以下常见问答场景intent用户问法示例图谱查询目标disease_symptom感冒有什么症状疾病 - 症状disease_cause高血压的病因是什么疾病 - 病因disease_check肺炎需要做什么检查疾病 - 检查disease_drug治疗胃炎吃什么药疾病 - 药品disease_food糖尿病患者能吃什么疾病 - 食物disease_department看癫痫挂哪个科疾病 - 科室这里的“意图打分”不是用概率模型打分而是通过触发词多少做简单投票比如问句同时包含“怎么治”和“药”就优先落到disease_drug。把触发词列在源码顶部后续调整业务口径时只改列表不改逻辑是这套设计的优点。3.3 解析器把 intent 转换成一个可执行的模板question_parser.py的作用是把分类器输出转换成cypher查询语句的模板。它和直接写查询的区别在于解析器只关心“要查什么”不关心“怎么连数据库”。# 示意代码question_parser.py 的模板转换 class QuestionParser: def parser(self, classify_result): intent classify_result[intent] disease classify_result[args][disease] if intent DISEASE_SYMPTOM: sql MATCH (d:Disease)-[:HAS_SYMPTOM]-(s:Symptom) WHERE d.name{0} RETURN s.name.format(disease) return sql if intent DISEASE_DRUG: sql MATCH (d:Disease)-[:HAS_DRUG]-(drug:Drug) WHERE d.name{0} RETURN drug.name.format(disease) return sql这里用字符串format拼接名字在这个源码场景里是可行的因为disease来自本地词典白名单而不是直接输入的用户文本。但如果你打算把这个系统开放成公网 API我建议把拼接改成 py2neo 的参数化查询避免用户用}) RETURN 1 //之类的内容污染 Cypher 语句。第 4 章会给出参数化的写法。4. 答案检索与对话循环answer_search.py 和 chatbot_graph.py问句解析成 Cypher 模板后真正执行图查询的是answer_search.py。这一段是系统里最接近“数据库驱动”的代码也是回答正确率的核心保障。4.1 Cypher 模板生成将问句模板映射为图查询answer_search.py中会接收上一步生成的查询语句连接 Neo4j 并执行随后把结果整理为自然语言格式。它的接口设计往往是这样# answer_search.py 核心结构 from py2neo import Graph class AnswerSearcher: def __init__(self): self.graph Graph(bolt://localhost:7687, auth(neo4j, neo4j)) def search_disease_symptom(self, disease): sql MATCH (d:Disease)-[:HAS_SYMPTOM]-(s:Symptom) \ WHERE d.name $name RETURN s.name result self.graph.run(sql, namedisease).data() return [item[s.name] for item in result]与第 3 章拼接字符串不同这里采用$name参数。py2neo 会把参数作为独立数据传递给 Neo4j不会与CYPHER语句结构混在一起安全性和可读性都更好。self.graph.run()返回的是一个游标对象.data()会一次性加载为列表字典适合问答系统这种小结果集场景。如果查找结果为空answer_search.py会返回一个默认话术比如“暂时没有找到关于该疾病的相关信息”。千万不要让这个空结果一路传到前端变成空白用户会以为是系统崩溃。4.2 查询边界与性能考量Cypher 查询虽然写起来简单但在医学图谱里容易踩两个坑关系方向写反以及没有限制返回条数。源码图谱中关系方向的约定通常是从疾病指向其他实体所以在回答“发烧会提示哪些疾病”这类反方向问题时需要显式写成MATCH (s:Symptom)-[:HAS_SYMPTOM]-(d:Disease)也就是把起始点换到症状节点。查询语义Cypher 起点 - 终点注意点疾病症状(d:Disease)-[:HAS_SYMPTOM]-(s:Symptom)疾病在左症状可能疾病(s:Symptom)-[:HAS_SYMPTOM]-(d:Disease)方向相反疾病科室(d:Disease)-[:DEPARTMENT]-(dep)若关系类型不同需替换在执行耗时上一般只在疾病、症状等高频实体上建立name的唯一约束查询路径很短毫秒级返回没有问题。如果后续图谱规模超过几十万节点可以在关系属性上增加索引并在 Cypher 末尾追加LIMIT 20避免一次性拉回上千条无关答案。4.3 机器人的主循环一次问答是如何串起来的chatbot_graph.py是让整个系统以一个对话机器人的形式跑起来的入口。它把前面三个 Python 文件串成流水线接收用户问题调用分类器解析器生成查询检索器去图库拿答案最后格式化输出。源码里通常还会带一个基于input()的循环方便在命令行直接验证。# chatbot_graph.py 主流程 from question_classifier import QuestionClassifier from question_parser import QuestionParser from answer_search import AnswerSearcher class ChatBotGraph: def __init__(self): self.classifier QuestionClassifier() self.parser QuestionParser() self.searcher AnswerSearcher() def chat(self, question): classify_result self.classifier.classify(question) if not classify_result: return 我暂时无法回答该问题你可以换一种问法 cypher self.parser.parser(classify_result) answers self.searcher.search(cypher) return \n.join(answers)这段代码后面的while True循环本身不复杂值得留意的是它没有任何上下文状态。也就是说用户问“感冒呢”系统是不知道上一轮在聊高血压的。如果生产环境要接微信客服必须在外面套一层会话记忆把上一轮实体传入下一次分类器而不是改chatbot_graph.py的内部逻辑。5. 部署调优与一个常用扩展把问答系统包装成 HTTP 接口这段分享一些把源码跑通并拓展成服务的经验。源码压缩包里有__pycache__下的 pyc 文件标注是cpython-36说明原运行环境是 Python 3.6。现代环境建议直接用 Python 3.8 以上版本pyc 文件跨版本无法复用解释器会在执行时重新生成。5.1 环境配置与依赖版本坑安装依赖时不要一把梭conda create -n medical_kg python3.8 -y conda activate medical_kg pip install py2neo2021.2.3 jieba requests beautifulsoup4 flask这里把 py2neo 版本锁到 2021.2.3是为了兼容 Neo4j 4.x。如果你本机装的是 Neo4j 5.x推荐用更高的 py2neo 4.x 版本如果使用过老的 py2neo 连接新 Neo4j最常见报错是The client is unauthorized due to authentication failure这不一定是密码错而是版本间的握手机制不匹配。启动 Neo4j 后用CYPHER 5语法在浏览器里执行RETURN 1验证版本。5.2 图谱构建的提速与安全重启开发过程中反复跑build_medicalgraph.py会留下大量测试产生的脏关系。清库注意用DETACH DELETEMATCH (n) DETACH DELETE n;DETACH DELETE会先删除所有关系再删节点比逐条删关系快很多。清完库再跑导入脚本能避免节点属性冲突导致 merge 失效。如果图库已经不小可以在build_medicalgraph.py里把graph.begin()和graph.commit(tx)包裹的数据条数设为 2000 条左右事务太大会导致 Neo4j 堆内存溢出事务太慢又拖慢导入速度。5.3 用 Flask 将 chatbot_graph 包装成 API要让这个基于命令行的问答系统对接外部前端直接改__main__里的循环并不优雅。我一般会新建一个app.py把ChatBotGraph实例化一次然后暴露 POST 接口。from flask import Flask, request, jsonify from chatbot_graph import ChatBotGraph app Flask(__name__) kg ChatBotGraph() app.route(/qa, methods[POST]) def qa(): payload request.get_json(forceTrue) question payload.get(question, ).strip() if not question: return jsonify({error: empty question}), 400 answer kg.chat(question) return jsonify({question: question, answer: answer}) if __name__ __main__: app.run(host0.0.0.0, port5000, debugFalse)这里把ChatBotGraph放在全局变量里避免每次请求都重新加载词典和连接 Neo4j。debugFalse防止调试器暴露到公网。启动后用curl验证curl -X POST http://127.0.0.1:5000/qa \ -H Content-Type: application/json \ -d {question:高血压患者适合吃什么食物}返回的 JSON 里answer字段就是图谱查询后的自然语言结果。如果中文出现乱码在 Flask 启动前执行import sys; sys.stdout.reconfigure(encodingutf-8)并确保终端使用 UTF-8 编码这是 Windows 环境下最容易踩的坑。本文还有配套的精品资源点击获取