写到第四篇前面几篇从知识图谱的基础概念聊到医疗问答系统的数据怎么整理再到上一篇的代码分析第一弹把这套“基于医疗知识图谱的问答系统”的骨架搭起来了。这篇继续啃代码我默认你已经看过前三篇至少知道我们要建的是什么样的东西一批疾病、症状、药品、科室的医疗实体以及它们之间“疾病-症状”“疾病-药物”“疾病-科室”的关系构建成图谱然后通过自然语言问句从图里查答案。这次代码分析的重点是后面这半截关系怎么抽、图谱怎么入库、问句怎么转成查询、答案怎么拼回来。也就是真正让系统“跑起来”的部分。我把实际能跑的代码结构、关键逻辑、参数怎么选、踩过哪些坑都拆开写清楚适合正在复现同类项目、或者准备从零做一个医疗图谱问答的开发者参考。1. 项目整体架构与代码模块划分1.1 项目目录里藏着什么先交代一下整个项目的代码布局这样后面讲代码时你不会迷路。我用的目录结构大致是这样的medical_qa/ ├── data/ │ ├── disease_symptom.txt │ ├── disease_drug.txt │ ├── disease_department.txt │ └── dict/ │ ├── disease.txt │ ├── symptom.txt │ ├── drug.txt │ └── department.txt ├── kg/ │ ├── entity_recognizer.py │ ├── relation_extractor.py │ └── graph_builder.py ├── qa/ │ ├── intent_parser.py │ ├── query_generator.py │ └── answer_formatter.py └── main.pydata/里是清洗后的结构化数据kg/里的三个文件负责把数据变成图谱qa/里的三个文件负责把问题变成图谱查询main.py把整条链路串起来。这套结构的核心思路是把“知识处理”和“问答处理”彻底分开。知识处理只管把数据变成图谱问答处理只管在图谱上做检索。这样做的好处是以后想替换数据来源、扩展实体类型或者换一种问答策略改动范围都收敛在单个模块里不用把整艘船拆了重造。1.2 数据流向从原始数据到答案整个系统跑起来是这样的链路读取data/dict/下的实体词典构建实体识别器的匹配库。读取data/下的三元组文件通过关系抽取器解析成标准结构。图谱构建器连接图数据库先把实体节点建好再建立关系。用户输入问题意图解析器判断问的是“症状”“用药”还是“科室”。查询生成器根据意图和实体动态拼出一条查询语句。答案格式化器把图数据库返回的记录拼成一句人能读懂的话。这篇我重点分析第2、3、5、6步的代码第1、4步上一篇已经聊过了这里只做衔接性回顾。提示代码示例里我用的是适合本地开发和中小规模知识的方案如果你的数据量到了十万级实体以上做法要换成批量导入工具这部分我在第5章会讲。2. 知识抽取代码医疗实体与关系是怎么挖出来的2.1 实体识别器代码回顾与边界处理上一篇文章里我详细分析过实体识别器的实现核心就是“词典 正向最大匹配”。再贴一下核心片段方便衔接后面的关系抽取class EntityRecognizer: def __init__(self, dict_dir): self.dictionary {} for name in [disease, symptom, drug, department]: path f{dict_dir}/{name}.txt with open(path, encodingutf-8) as f: self.dictionary[name] {line.strip() for line in f if line.strip()} self.max_word_len max( len(word) for words in self.dictionary.values() for word in words ) def recognize(self, text): hits [] i 0 n len(text) while i n: matched False for l in range(min(self.max_word_len, n - i), 1, -1): w text[i : i l] for etype, words in self.dictionary.items(): if w in words: hits.append({word: w, type: etype, start: i, end: i l}) i l matched True break if matched: break if not matched: i 1 return hits这里的边界处理有几个细节值得新手上路前搞清楚。第一max_word_len决定了每次从多长的窗口开始尝试匹配是从最长词长往下递减这叫“正向最大匹配”。它比逐字匹配快得多但要注意如果词典里最长的词有十几个字每次匹配的开销都会偏大所以我这个实现里还对窗口加了保护避免越界。第二代码里for etype, words in self.dictionary.items()在窗口层内嵌套了一层实体类型的遍历这意味着同一个词如果同时出现在“疾病”和“症状”词典里先命中的类型会胜出。实际数据里这种情况不少比如“糖尿病”可能在某些资料里被当成症状描述但在我们的词典里它必须是疾病。我的处理办法是构建词典时做了一次优先级去重优先保留疾病类型这块逻辑写在了数据清洗脚本里不在这个类里。第三识别器没有做分词兼容。用户问题里如果夹杂了英文、数字比如“血压BP 160算高吗”“感冒发烧3天”目前的匹配逻辑是老老实实滑过这些字符实体照样能抓到但统计信息会乱一点。这个项目是医疗专用场景强行上通用分词器反而会被错误分词干扰所以保持“专词专用”是最稳的。2.2 关系抽取器的数据解析逻辑关系抽取器在项目里的作用不是从自由文本里用算法抽取关系而是把整理好的三元组文件解析成代码能直接消费的关系列表。先把原文件长什么样展示一下disease_symptom.txt里每一行是这样感冒 发热 常见 感冒 咳嗽 常见 感冒 流鼻涕 常见 偏头痛 恶心 偶见 糖尿病 多饮 常见字段之间用 Tab 分割第一列是头实体第二列是尾实体第三列是关系权重描述。为什么要加第三列因为医疗关系不能都是布尔式的“有/无”症状对疾病的提示价值差别很大——“发热”对“感冒”是常见症状“恶心”对“偏头痛”可能只是偶见。保留这个强度信息后期做图谱关系属性或问答排序都有用。再看关系抽取器的代码和你想的可能不一样它其实很薄class RelationExtractor: RELATION_MAP { disease_symptom.txt: HAS_SYMPTOM, disease_drug.txt: TREAT_WITH, disease_department.txt: VISIT_DEPARTMENT, } def __init__(self, data_dir): self.data_dir data_dir def extract_all(self): relations [] for filename, rel_type in self.RELATION_MAP.items(): path f{self.data_dir}/{filename} with open(path, encodingutf-8) as f: for idx, line in enumerate(f, start1): parts line.strip().split(\t) if len(parts) ! 3: print(f跳过第{idx}行字段数不对: {line}) continue head, tail, strength parts relations.append( { head: head, tail: tail, relation: rel_type, strength: strength, } ) return relations这段代码本身没有高深的技术含量但它是整条数据链路的“门卫”。文件里多了一个换行、少了一个 Tab数据处理阶段立刻会报错或者污染图谱。我把字段数不对的行直接跳过并打印行号这就是典型的“宁可报出来也不要静默出错”的思路——后面真正往图数据库里灌数据时任何一条脏数据都会变成图谱里的一根错边。关于 strength 字段构建 Cypher 时我会把它做成关系属性weight。这样在问答环节如果同一个症状匹配到多个疾病可以按 weight 排序优先返回“常见”关联的疾病而不是把所有可能性都丢给用户。2.3 实体统一与空格清洗的隐藏工作量关系抽取器解析出的实体名必须和实体词典里的叫法完全对齐否则识别器就算抓到了实体图谱里也找不到对应节点。这里最容易出问题的是全角半角空格、大小写、括号格式不统一。比如数据文件里写了“2型糖尿病”词典里却只有“II型糖尿病”识别器匹配时基本是字符串精确匹配两个字不一样就完蛋。我在数据预处理脚本中做了这些归一化把所有全角字母数字转成半角。去掉实体名首尾空格。把中文括号“”统一替换为英文括号“()”。对别名表里的实体把所有别名都映射到标准名。这套清洗逻辑不算代码分析的硬核内容但每个正儿八经的知识图谱项目里都得写一遍。强烈建议你在做关系抽取前先把这三步跑掉不然图谱构建器入库后才发现实体对不上返工成本非常高。3. 图谱写入与存储查询代码解读3.1 批量写入实体节点时索引为什么放在前面图谱不是把数据丢进图数据库就完事的。节点数量不大时无所谓但到了几千节点、上万关系时没有索引的查询会慢得让你怀疑人生。我用的图数据库是 Neo4j连接和初始化代码如下from py2neo import Graph class GraphBuilder: def __init__(self, uri, user, password): self.graph Graph(uri, auth(user, password)) self.relations [] def build(self, relations): self.create_constraints() self.create_nodes(relations) self.create_relationships(relations) def create_constraints(self): self.graph.run(CREATE CONSTRAINT disease_name IF NOT EXISTS ON (n:Disease) ASSERT n.name IS UNIQUE) self.graph.run(CREATE CONSTRAINT symptom_name IF NOT EXISTS ON (n:Symptom) ASSERT n.name IS UNIQUE) self.graph.run(CREATE CONSTRAINT drug_name IF NOT EXISTS ON (n:Drug) ASSERT n.name IS UNIQUE) self.graph.run(CREATE CONSTRAINT department_name IF NOT EXISTS ON (n:Department) ASSERT n.name IS UNIQUE)之所以先建唯一性约束是因为后续创建节点时必须依赖“存在即不重复”的特性。假如不建约束你跑两遍构建脚本图谱里就会有两套一模一样的“感冒”节点关系也会跟着翻倍。这类问题很难靠肉眼发现等到问答时返回了重复结果你才会意识到数据已经变脏了。约束是原生索引的一种它同时承担了“唯一性校验”和“按 name 查节点加速”两个职责。Neo4j 官方也建议索引命名要语义化所以我这里直接用了_name后缀查询时也能一眼分清。3.2 用 py2neo 做批量创建的正确姿势创建节点时最直接的写法是一个一个建但几千个实体逐个merge会有明显的网络往返开销。我用的是批量unwind的方式把 Python 列表丢给一条 Cypher 语句处理性能至少提升一个量级。核心代码是这样的def create_nodes(self, relations): entities set() for rel in relations: entities.add((rel[head], Disease)) entities.add((rel[tail], self._type_of_tail(rel[relation]))) # 将实体一次性写入 from py2neo import Node nodes [ Node(label, namename) for (name, label), label in [self._label_mapping(e) for e in entities] ] self.graph.merge(nodes, primary_keyname)这里涉及merge的一个重要特性它和create的区别在于merge会先检查是否存在相同标签和主键的节点存在就跳过不存在才创建。用主键name做唯一性判断正好和我们前面建的约束呼应。讲到_type_of_tail这个方法它根据关系类型反向判断尾实体的标签HAS_SYMPTOM的尾巴是症状TREAT_WITH的尾巴是药物VISIT_DEPARTMENT的尾巴是科室。这个映射逻辑不复杂但它保证了图谱里的节点标签是严格统一的不会出现“药物节点被建成了症状节点”这种错位。注意merge批量写入时Node对象列表里的标签必须一致或者个别指定我在工程里做了一个小封装让它按实际标签组装代码里为了可读性做了简化。真实项目中建议把实体去重后的数据按标签分组再一组一组merge。3.3 关系写入与 weight 属性的作用关系批量写入的代码和节点类似只不过目标是“边”而不是“点”def create_relationships(self, relations): for rel in relations: cypher ( fMATCH (a:Disease {{name: {rel[head]}}}), f(b:{self._label_of_tail(rel[relation])} {{name: {rel[tail]}}}) fMERGE (a)-[r:{rel[relation]}]-(b) fSET r.weight {self._weight_value(rel[strength])} ) self.graph.run(cypher)逐条run在数据量小时完全够用几千条关系可能也就几十秒。但如果数据量到了几万条逐条执行的时间会让人坐不住这时候应该改成UNWIND成批传入。我在项目里写了两种模式默认走批量模式只有 debug 时才用逐条模式。weight属性的取值逻辑是这样的常见映射为 1.0偶见映射为 0.5其他情况映射为 0.8。这个 mapping 不是随便拍的我参考了医疗资料里对症状和疾病关联强度的常见划分把它量化为可比较的数字。后面问答排序时这个值直接决定“最可能疾病”的排列顺序。图谱入库完成后可以用一行查询快速验证成果MATCH (n) RETURN labels(n) AS label, count(*) AS cnt如果返回结果里疾病数量、症状数量和你数据文件里的实体数对不上优先检查数据清洗时的去重逻辑和实体别名映射。4. 问答主流程代码串联与关键方法解读4.1 问答处理器的核心流程如果说图谱构建是“写数据”那问答这块就是“读数据”。这部分我从用户输入到最终答案完整走一遍。main.py里主流程长这样def answer(question): entities recognizer.recognize(question) intent intent_parser.parse(question) if not entities: return 我没听懂你说的医学实体换个说法试试 if not intent: return 我能回答症状、用药和科室问题换个问法试试 cypher query_generator.generate(intent, entities) result graph.run(cypher).data() return answer_formatter.format(question, intent, entities, result)一句话总结这个流程先找出实体再判断意图接着根据“意图实体”生成查询最后把查询结果格式化成回答。四个步骤每一个都可能让答案偏离预期所以每个模块的返回信息我都设计了兜底文案。这里有个很关键的取舍为什么要把意图判断放在实体识别之后因为实体识别会消耗较多时间如果问题本身就不是医学相关的先做意图判断能更快拒绝。但反过来如果连实体都没有意图也没有意义。我的实际调试经验是用户问的问题几乎都带至少一个实体所以先用实体过滤再判断意图命中率更高也能遮掉一些意图解析器的误判。4.2 意图解析里的模板匹配与阈值意图解析器我上一篇详细讲过这里重点看它怎么衔接查询生成。它的核心是一个二维规则表INTENT_RULES [ (symptom, [有什么症状, 有哪些症状, 表现为, 会发烧吗, 流鼻涕吗]), (treatment, [怎么治, 吃什么药, 用什么药, 如何治疗, 该怎么办]), (department, [挂什么科, 哪个科, 去看什么科, 科室]), ]规则匹配的代码不复杂遍历规则看问题文本里是否包含任何一个触发词。但“包含”不等于“一定对”我加了一个覆盖率阈值如果同一个意图被多个触发词命中或者触发词在问句中的字符占比超过一定比例才判定为该意图否则返回未知意图。这个阈值我调成 0.5也就是至少要有两个触发词命中或者一个较长触发词三个字以上命中才会判定成功。举个实际例子“感冒吃什么药好得快”里感冒是疾病实体吃什么药命中治疗触发词意图是treatment结果会去查TREAT_WITH关系。而“感冒有什么药不能吃”同样包含“什么药”但语义是禁忌而不是常规治疗目前规则表没有覆盖到“禁忌”这类细粒度意图最终会落到兜底逻辑。这种情况我如实写在了项目说明里后续可以考虑扩展禁用药关系但现阶段先把主干跑通更重要。4.3 查询生成器的动态拼查询语句意图和实体都定了之后查询生成器负责把两者翻译成 Cypher。基本写法如下class QueryGenerator: TEMPLATES { symptom: MATCH (d:Disease)-[r:HAS_SYMPTOM]-(s:Symptom) WHERE d.name $name RETURN s.name AS answer, r.weight AS weight ORDER BY weight DESC , treatment: MATCH (d:Disease)-[r:TREAT_WITH]-(drug:Drug) WHERE d.name $name RETURN drug.name AS answer, r.weight AS weight ORDER BY weight DESC , department: MATCH (d:Disease)-[:VISIT_DEPARTMENT]-(dept:Department) WHERE d.name $name RETURN dept.name AS answer , } def generate(self, intent, entities): if intent not in self.TEMPLATES: raise ValueError(f未知意图: {intent}) disease_entity next((e for e in entities if e[type] disease), None) if not disease_entity: raise ValueError(没有识别到疾病实体) return self.TEMPLATES[intent], {name: disease_entity[word]}这个生成器做了一件和教科书不太一样的事它强制要求问题里必须识别出疾病实体而不是随便哪个实体就行。因为我们的三种意图最终都围绕“某个病”展开少了疾病实体查询结果根本没有锚点。如果用户问的是“发烧应该挂什么科”实体识别器命中的是症状“发烧”这时查询生成会失败并返回提示。严格来说这种问题应该被回答成“建议先判断可能疾病再挂号”但当前项目的范围是“已知疾病查信息”所以直接把问题归为不支持类型。$name参数用了传参方式而不是字符串拼接这是刻意的。一个是避免引号、特殊字符把语句搞坏另一个是防止 Cypher 注入。虽然本地项目不太可能遇到恶意输入但这种习惯在数据库操作里必须养成。4.4 答案格式化图谱记录变成人话图数据库返回的result是字典列表形如[{answer: 发热, weight: 1.0}, {answer: 咳嗽, weight: 0.8}]。如果直接把结构体抛给用户体验太差了。答案格式化器做的事情包括按 weight 降序排列。如果答案超过 5 个只保留前 5 个。拼成“常见的症状有发热、咳嗽、流鼻涕等。”这种自然语言。class AnswerFormatter: LIMIT 5 def format(self, question, intent, entities, result): if not result: return 数据库里暂时没查到相关信息 answers [r[answer] for r in result if r.get(answer)] if len(answers) self.LIMIT: answers_text 、.join(answers[: self.LIMIT]) 等 else: answers_text 、.join(answers) return f{entities[0][word]}{self._intent_message[intent]}{answers_text}格式化器里我加了LIMIT上限原因是医疗问答场景下用户并不需要一次拿到几十种药物或几十个症状给前几个高关联的答案更符合实际使用习惯。针对“症状”意图返回多个常见症状没问题针对“治疗”意图如果结果太多说明该疾病对应的药物列表很长前 5 个通常就是首选药物或常用药。这段代码看起来简单但它承担了问答系统“最后印象”的职责。我调试时发现图数据库返回的字段名如果和格式化器里取的不一致会直接抛KeyError。所以查询返回时统一用AS answer给字段改名让格式化器不用关心具体意图的数据结构。5. 代码调试实录与踩坑清单5.1 我实际跑代码遇到的问题这个项目我前后调试了好几天把印象最深的问题记录下来给你做个参考。问题现象原因解决方法Neo4j 连接超时graph.run() 报连接错误本机 Neo4j 服务没启动或者密码不对启动服务后检查bolt://localhost:7687连通性中文乱码图谱里的节点名称是??连接配置缺少 UTF-8 编码声明数据库初始化时设置字符集Python 脚本文件统一 UTF-8实体识别漏掉别名问“高血压”命中的是空词典里只有“原发性高血压”给词典加别名映射如“高血压”-“原发性高血压”查询返回空列表明明图谱里有关系实体词和图谱里节点名不一致检查数据清洗的规范化步骤尤其是括号和空格构建图谱时重复节点merge 没有起到去重效果没有先建唯一性约束在 build 前先跑 create_constraints匹配窗口过大长句子时识别速度慢max_word_len 设置成全局最长词改成按实体类型维护各自的最大长度这里面的“中文乱码”我印象最深刻。代码层面完全看不出问题文件用 UTF-8 打开、数据库连接也没写过其他编码参数但节点名称就是乱码。最后发现是 Neo4j 服务端默认字符配置的问题在启动配置里指定了 UTF-8 之后问题消失。国内做文本类项目大概率会遇到编码坑建议建环境时就把统一编码提到首位。实体别名问题也是个典型场景。真实医疗术语非常不统一用户说“高血压”知识库里可能叫“原发性高血压”也可能叫“高血压病”。我在词典构建脚本里专门维护了一张“口语别名表”把常见口语名和标准名做映射识别器在识别阶段直接输出标准名这样查询生成器能直接用标准名匹配图谱一举两得。5.2 数据量变大后性能怎么优化当前几千条数据时代码跑起来毫无压力。但如果你想把规模扩大到几万实体、几十万关系有几个优化点务必要做。第一节点与关系的批量写入必须换成UNWIND不能再逐条MATCH。我测试过逐条执行一万条关系大概要两三分钟换成一整批UNWIND之后只要十几秒差别非常明显。建议把create_relationships改成收集参数列表后一次graph.run(cypher, parametersrelations)。第二查询侧要保证按name查询时走索引。我们的CREATE CONSTRAINT已经建立了唯一索引但如果你在某个阶段把实体名改成了前缀匹配比如WHERE d.name STARTS WITH $prefix那索引的利用率会大打折扣这属于查询模式差异需要单独评估。第三如果图谱数据全部塞在一个库里而且有大量并发查询要考虑加一层缓存。最轻量的方案是用 Redis 缓存“实体名到图谱查询结果”的映射因为医疗问答的常见问题重复度很高缓存命中率会非常理想。这个问题我在当前版本里没有上缓存但架构上已经预留了接口后续要加也只需要在answer()外层包一层装饰器。5.3 从代码到系统的扩展空间最后聊点实在的。这套代码虽然跑通了但它离一个“产品级”的医疗问答系统还有距离。我复盘下来有三个方向值得继续做。第一个方向是引入“意图槽位”机制。现在意图和实体是分开解析的但实体之间其实存在依赖关系。比如“感冒发烧吃什么药”里包含“感冒”和“发烧”两个实体一个疾病一个症状严格来说应该先确认“发烧”是不是“感冒”的常见症状再返回药物。这需要把实体关系校验接入问答流程代码上大概增加一个校验函数就行。第二个方向是增强图谱的可解释性。当前返回的答案是列表用户看不出“为什么推荐这个药”。如果能在返回答案的同时给出路径比如“感冒——常见症状——发热”“感冒——常用药物——某药”交互上会更可信。第三个方向是把规则式问答升级成模型式问答。意图解析、实体识别都能用预训练语言模型替代词典和规则表可以作为兜底方案。但这属于重投入我个人建议先在小数据量、小规模场景下把规则系统做到极致模型优化反而是后面的事。回到代码本身这套系统最值得借鉴的不是某个算法而是那种“每个环节都留了兜底、每个返回都有可读性”的工程习惯。我在跑代码的过程中踩过的坑有一半以上其实都是“数据不规范”带来的另一半是“图数据库使用习惯”带来的真正死在算法逻辑上的情况反而很少。所以说先把自己数据的命门管好图谱项目基本就稳了一半。
