医疗知识图谱问答系统核心代码解析：关系抽取与图谱入库

写到第四篇，前面几篇从知识图谱的基础概念聊到医疗问答系统的数据怎么整理，再到上一篇的代码分析第一弹，把这套“基于医疗知识图谱的问答系统”的骨架搭起来了。这篇继续啃代码，我默认你已经看过前三篇，至少知道我们要建的是什么样的东西：一批疾病、症状、药品、科室的医疗实体，以及它们之间“疾病-症状”“疾病-药物”“疾病-科室”的关系，构建成图谱，然后通过自然语言问句从图里查答案。

这次代码分析的重点是后面这半截：关系怎么抽、图谱怎么入库、问句怎么转成查询、答案怎么拼回来。也就是真正让系统“跑起来”的部分。我把实际能跑的代码结构、关键逻辑、参数怎么选、踩过哪些坑，都拆开写清楚，适合正在复现同类项目、或者准备从零做一个医疗图谱问答的开发者参考。

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.py

data/里是清洗后的结构化数据，kg/里的三个文件负责把数据变成图谱，qa/里的三个文件负责把问题变成图谱查询，main.py把整条链路串起来。

这套结构的核心思路是把“知识处理”和“问答处理”彻底分开。知识处理只管把数据变成图谱，问答处理只管在图谱上做检索。这样做的好处是以后想替换数据来源、扩展实体类型，或者换一种问答策略，改动范围都收敛在单个模块里，不用把整艘船拆了重造。

1.2 数据流向从原始数据到答案

整个系统跑起来是这样的链路：

1. 读取data/dict/下的实体词典，构建实体识别器的匹配库。
2. 读取data/下的三元组文件，通过关系抽取器解析成标准结构。
3. 图谱构建器连接图数据库，先把实体节点建好，再建立关系。
4. 用户输入问题，意图解析器判断问的是“症状”“用药”还是“科室”。
5. 查询生成器根据意图和实体，动态拼出一条查询语句。
6. 答案格式化器把图数据库返回的记录拼成一句人能读懂的话。

这篇我重点分析第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, encoding="utf-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, encoding="utf-8") as f: for idx, line in enumerate(f, start=1): 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型糖尿病”，识别器匹配时基本是字符串精确匹配，两个字不一样就完蛋。我在数据预处理脚本中做了这些归一化：

1. 把所有全角字母数字转成半角。
2. 去掉实体名首尾空格。
3. 把中文括号“（）”统一替换为英文括号“()”。
4. 对别名表里的实体，把所有别名都映射到标准名。

这套清洗逻辑不算代码分析的硬核内容，但每个正儿八经的知识图谱项目里都得写一遍。强烈建议你在做关系抽取前先把这三步跑掉，不然图谱构建器入库后才发现实体对不上，返工成本非常高。

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, name=name) for (name, label), label in [self._label_mapping(e) for e in entities] ] self.graph.merge(nodes, primary_key="name")

这里涉及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 = ( f"MATCH (a:Disease {{name: '{rel['head']}'}}), " f"(b:{self._label_of_tail(rel['relation'])} {{name: '{rel['tail']}'}}) " f"MERGE (a)-[r:{rel['relation']}]->(b) " f"SET 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}]。如果直接把结构体抛给用户，体验太差了。答案格式化器做的事情包括：

1. 按 weight 降序排列。
2. 如果答案超过 5 个，只保留前 5 个。
3. 拼成“常见的症状有：发热、咳嗽、流鼻涕等。”这种自然语言。

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 我实际跑代码遇到的问题

这个项目我前后调试了好几天，把印象最深的问题记录下来，给你做个参考。

这里面的“中文乱码”我印象最深刻。代码层面完全看不出问题，文件用 UTF-8 打开、数据库连接也没写过其他编码参数，但节点名称就是乱码。最后发现是 Neo4j 服务端默认字符配置的问题，在启动配置里指定了 UTF-8 之后问题消失。国内做文本类项目大概率会遇到编码坑，建议建环境时就把统一编码提到首位。

实体别名问题也是个典型场景。真实医疗术语非常不统一，用户说“高血压”，知识库里可能叫“原发性高血压”，也可能叫“高血压病”。我在词典构建脚本里专门维护了一张“口语别名表”，把常见口语名和标准名做映射，识别器在识别阶段直接输出标准名，这样查询生成器能直接用标准名匹配图谱，一举两得。

5.2 数据量变大后性能怎么优化

当前几千条数据时，代码跑起来毫无压力。但如果你想把规模扩大到几万实体、几十万关系，有几个优化点务必要做。

第一，节点与关系的批量写入必须换成UNWIND，不能再逐条MATCH。我测试过，逐条执行一万条关系大概要两三分钟，换成一整批UNWIND之后只要十几秒，差别非常明显。建议把create_relationships改成收集参数列表后一次graph.run(cypher, parameters=relations)。

第二，查询侧要保证按name查询时走索引。我们的CREATE CONSTRAINT已经建立了唯一索引，但如果你在某个阶段把实体名改成了前缀匹配，比如WHERE d.name STARTS WITH $prefix，那索引的利用率会大打折扣，这属于查询模式差异，需要单独评估。

第三，如果图谱数据全部塞在一个库里，而且有大量并发查询，要考虑加一层缓存。最轻量的方案是用 Redis 缓存“实体名到图谱查询结果”的映射，因为医疗问答的常见问题重复度很高，缓存命中率会非常理想。这个问题我在当前版本里没有上缓存，但架构上已经预留了接口，后续要加也只需要在answer()外层包一层装饰器。

5.3 从代码到系统的扩展空间

最后聊点实在的。这套代码虽然跑通了，但它离一个“产品级”的医疗问答系统还有距离。我复盘下来，有三个方向值得继续做。

第一个方向是引入“意图槽位”机制。现在意图和实体是分开解析的，但实体之间其实存在依赖关系。比如“感冒发烧吃什么药”里包含“感冒”和“发烧”两个实体，一个疾病一个症状，严格来说应该先确认“发烧”是不是“感冒”的常见症状，再返回药物。这需要把实体关系校验接入问答流程，代码上大概增加一个校验函数就行。

第二个方向是增强图谱的可解释性。当前返回的答案是列表，用户看不出“为什么推荐这个药”。如果能在返回答案的同时给出路径，比如“感冒——常见症状——发热”“感冒——常用药物——某药”，交互上会更可信。

第三个方向是把规则式问答升级成模型式问答。意图解析、实体识别都能用预训练语言模型替代，词典和规则表可以作为兜底方案。但这属于重投入，我个人建议先在小数据量、小规模场景下把规则系统做到极致，模型优化反而是后面的事。

回到代码本身，这套系统最值得借鉴的不是某个算法，而是那种“每个环节都留了兜底、每个返回都有可读性”的工程习惯。我在跑代码的过程中踩过的坑，有一半以上其实都是“数据不规范”带来的，另一半是“图数据库使用习惯”带来的，真正死在算法逻辑上的情况反而很少。所以说，先把自己数据的命门管好，图谱项目基本就稳了一半。