简介面向知识图谱与医疗信息检索初学者这份资源利用ask120平台真实问答数据完整演示了从网络爬虫采集、数据清洗与建模到导入Neo4j图数据库再借助Django实现简易医疗问答应用的开发链路。包体共37个文件以Python源码为核心既有spider1.py、spider2.py等爬虫模块也有manage.py、settings.py、urls.py、views.py等Django工程文件搭配HTML页面与JavaScript交互、XML工程配置以及PNG示意图整体仅78KB轻量小巧结构简洁便于逐文件对照学习。资源还一并提供了pyc编译文件与.idea工程配置帮助使用者理解项目运行环境与调试入口。内容预览中的目录结构清晰明了直接运行或改造即可快速掌握医疗实体抽取、关系链接和Cypher查询等关键技能尤其适合课程设计与毕业设计参考。已有5461人学习下载对于想从零搭建医疗问答知识图谱的开发者具有不错的参考价值。1. 医疗问答别急着上模型先问知识图谱怎么组织做医疗问答最容易被卡住的不是模型而是知识怎么组织。某开发者用一个模拟项目X试过把疾病、症状、药物、科室存进三张 MySQL 表第一个问题能硬写 JOIN 答上来第二个问题一来又要加字段。换成 Neo4j 知识图谱后同一份三元组数据只改查询模板就能覆盖「症状、药物、科室」三类问答。这份基于 Neo4j 的简易医疗问答知识图谱本质是一套可复现的工程链路图建模、数据导入、问句解析、Cypher 生成、答案返回每一步都有对应的脚本和数据文件。适合想把知识图谱落到代码里的后端开发者也适合需要医疗导诊 Demo 基座的同学。2. Neo4j 图建模疾病、症状、药物、科室的四类节点三种关系2.1 为什么这个场景更适合 Neo4j 而不是 MySQL医疗常识天然是图。拿流感举例「流感 → 发热 → 布洛芬 → 发热门诊」这条链在 MySQL 里需要疾病表、症状表、药物表、科室表和至少三张关联表。回答「流感有什么症状」要做两次 JOIN回答「发热可以看哪个科」要先从症状反查疾病再关联科室。需求一多SQL 的组合数量会快速膨胀。图数据库的思路完全不同节点代表实体关系代表语义查询只在路径上做局部计算。Neo4j 的MATCH (d:疾病 {name:流感})-[:HAS_SYMPTOM]-(s:症状) RETURN s.name与人的问法几乎一一对应。图遍历的性能特征非常适合关系密集、查询路径超过一跳的场景。实体总量到十万级以后这种结构优势会更明显。不过这里要说清一个容易误用的点不是所有业务都要搬进 Neo4j。知识图谱适合「关系密集、路径查询多」的部分用户账号、操作日志这类事务型数据留在原库更稳。本项目的取舍是只把医疗实体与关系做成图问答服务通过 Python 驱动连 Neo4j其余逻辑保持轻量。这样即便是几百条数据的小图谱也能完整展示一套生产可用的交互链路。如果一开始就想着把所有数据都塞进图里模型会变得非常臃肿查询性能也未必比 MySQL 好。2.2 节点与关系定义数据规模做得很克制但结构完整。节点一共四类每类都有 name 属性作为全局唯一标识其余属性按场景补充| 节点标签 | 主要属性 | 来源文件 | | 疾病 | name, desc | disease.csv | | 症状 | name, organ | symptom.csv | | 药物 | name, dosage, usage | drug.csv | | 科室 | name, desc | department.csv |关系只建三类方向统一从疾病出发避免双写导致不一致| 关系类型 | 起点 | 终点 | 语义 | | HAS_SYMPTOM | 疾病 | 症状 | 该疾病会有某症状 | | USES_DRUG | 疾病 | 药物 | 该疾病常使用某药 | | VISIT_DEPT | 疾病 | 科室 | 该疾病应就诊科室 |为什么方向全从疾病出发因为问题最常见的形态是「疾病 → 结果」问症状、问药物、问科室。而「什么病会有发热」这种反查用反向遍历-[:HAS_SYMPTOM]-就能覆盖不需要额外存一条反向关系。只存一套关系的好处是更新时不会出现两条数据状态不一致代价是查询时要稍微留意方向。还有个属性归属问题要单独说。像「头痛这个症状在哪些疾病里出现频率最高」这类问题如果把 frequency出现频率放在症状节点上就会丢失「这个频率是相对哪种疾病」的上下文正确做法是把它放到 HAS_SYMPTOM 关系上作为关系的属性。关系属性和节点属性的选用规则本项目的经验是跟两个实体同时相关的信息放关系只属于单个实体的信息放节点。2.3 数据文件与三元组格式导入脚本读 CSV所以先要把知识整理成统一格式。节点文件直接以标签命名每行一个实体。以 disease.csv 为例name,desc 支气管炎,气管黏膜的急性或慢性炎症 偏头痛,反复发作的一侧搏动性头痛 胃溃疡,胃壁黏膜出现溃疡性缺损关系统一放在 relation.csv 里用四个字段描述一条三元组source_type,source_name,rel_type,target_type,target_name 疾病,支气管炎,HAS_SYMPTOM,症状,咳嗽 疾病,支气管炎,USES_DRUG,药物,头孢克洛 疾病,偏头痛,VISIT_DEPT,科室,神经内科这种通用列名有一个好处从公开医疗术语集整理数据时只需要做一次字段映射不用为每个数据集重写导入脚本。我一般会把原始来源留一个 source 字段方便将来回溯错误。文件编码统一用 UTF-8带 BOM 的 Excel 导出文件也能被脚本直接吞掉。2.4 建模阶段就要想到的三个查询动手写导入脚本之前我建议先列出三个必须能答上的查询相当于给图结构做验收测试。第一个是「某疾病有什么症状」第二个是「某症状可能是什么病」第三个是「某疾病吃什么药、挂什么科」。这三个查询分别对应正向关系、反向遍历和多类型输出。如果某个查询需要跨三种关系才能回答那就要回到 2.2 的表格再检查是不是漏建了关系。这一点比建模本身更重要图结构不是一次设计完的而是由查询倒推出来的。项目里当初就是先写了这三个查询才定下现在的三类关系没做多余的边。等问答层出现问题比如查询需要「疾病 → 症状 → 科室」这种二跳路径再考虑新增关系类型。前期把关系控制在必要范围内后期维护成本会低很多。3. 数据导入 Neo4j用 Python 驱动把三元组批量写进图3.1 三种导入方式怎么选Neo4j 写入数据的常见路径有三条Cypher Shell 一条条执行、neo4j-admin import离线导入、Python/Java 驱动调用事务 API。对这份医疗问答资源来说数据量通常在万条以内而且可能反复调整所以驱动方式更合适。| 方式 | 适合场景 | 缺点 | | Cypher Shell | 手工验证几条数据 | 无法做复杂清洗逻辑 | | neo4j-admin import | 首次全量导入几百万级数据 | 需要停库重建图 | | Python 驱动 UNWIND | 增量维护、问答联动调试 | 需要管理事务批次 |我一般会用 Python 驱动因为 CSV 清洗、同义词映射和写入可以在同一个脚本里完成跑完还能立刻调用问答接口验证。后面如果数据量确实到十亿级再考虑换成 admin import但那是另一个工程话题了。3.2 连接参数与唯一性约束from neo4j import GraphDatabase URI bolt://localhost:7687 AUTH (neo4j, replace-me) DRIVER GraphDatabase.driver(URI, authAUTH) def init_schema(session): session.run(CREATE CONSTRAINT disease_name IF NOT EXISTS FOR (n:疾病) REQUIRE n.name IS UNIQUE) session.run(CREATE CONSTRAINT symptom_name IF NOT EXISTS FOR (n:症状) REQUIRE n.name IS UNIQUE) session.run(CREATE CONSTRAINT drug_name IF NOT EXISTS FOR (n:药物) REQUIRE n.name IS UNIQUE) session.run(CREATE CONSTRAINT dept_name IF NOT EXISTS FOR (n:科室) REQUIRE n.name IS UNIQUE)参数说明URI默认指向本地默认端口生产环境记得改成内网地址并配套鉴权AUTH写你自己的密码。四条约束分别给四类节点建唯一索引后续 MERGE 的查重都靠这个约束。如果不建约束MERGE 只是尽量找并不能真正防止重复。3.3 批量写入节点接下来是三段核心代码。第一段读 CSV 建节点import csv from typing import List, Dict def load_nodes(session, label: str, filepath: str, batch_size: int 1000): with open(filepath, encodingutf-8-sig) as f: rows list(csv.DictReader(f)) for start in range(0, len(rows), batch_size): chunk rows[start:start batch_size] query ( fUNWIND $rows AS row fMERGE (n:{label} {{name: row.name}}) fSET n row ) session.run(query, rowschunk)逻辑说明MERGE按 name 找节点找不到就创建找到就在原节点上补充属性。SET n row会把 CSV 里除 name 外的所有字段写进节点属性新增列时不用改 Cypher。batch_size控制每个事务的行数行数少可以一次跑完行数多就分片提交避免事务过大。为什么用utf-8-sig因为 Excel 导出的 CSV 经常带 BOM直接用utf-8读会把第一个字段名读成\ufeffname后面的映射全错。这一点不处理第一批数据导入后大概率会得到两个不一样的节点。3.4 批量写入关系第二段代码写关系REL_CYPHER ( UNWIND $rows AS row MATCH (a:{source_type} {{name: row.source_name}}) MATCH (b:{target_type} {{name: row.target_name}}) MERGE (a)-[rel:{rel_type}]-(b) ) def load_relations(session, relations: List[Dict]): session.run( REL_CYPHER.format( source_type疾病, target_type症状, rel_typeHAS_SYMPTOM ), rows[r for r in relations if r[rel_type] HAS_SYMPTOM] ) session.run( REL_CYPHER.format( source_type疾病, target_type药物, rel_typeUSES_DRUG ), rows[r for r in relations if r[rel_type] USES_DRUG] ) session.run( REL_CYPHER.format( source_type疾病, target_type科室, rel_typeVISIT_DEPT ), rows[r for r in relations if r[rel_type] VISIT_DEPT] )逻辑说明每类关系单独跑一个 UNWIND中间按rel_type过滤这样如果某类关系缺数据能直接定位到是源文件还是过滤条件的问题。MATCH之前依赖 3.2 的唯一性约束生成索引所以按 name 找起点的速度有保证。MERGE 关系同样是为了防止重复跑脚本时插入两条一样的边。一个容易忽略的细节这里的方向是(a)-[rel]-(b)意味着 a 是关系的起点。写 relation.csv 时我刻意把 source 都设为「疾病」所以三类关系才能复用同一个模板。如果你在本地扩展了新的关系类型比如「药物-CAN_BE_USED_FOR-疾病」记得把格式化模板里的起点终点类型一起换掉不然新建的关系方向会是反的。3.5 导入完的校验查询跑完脚本不要急着写问答先用两条 Cypher 给图谱做体检MATCH (n) RETURN labels(n)[0] AS label, count(*) AS cnt ORDER BY label;MATCH (d:疾病)-[r]-(x) RETURN d.name AS disease, type(r) AS rel, head(labels(x)) AS target LIMIT 20;第一条看四类节点的数量是否和 CSV 行数一致。不一致的典型原因重复实体、约束未生效、编码问题。第二条看关系是否都从疾病出发target 类型是否符合三类关系的预期。如果某类关系数量为 0优先检查 relation.csv 里的 rel_type 有没有大小写不一致Cypher 的关系类型是区分大小写的has_symptom和HAS_SYMPTOM是两个不同的边类型。注意校验查询要放到每次导入之后执行不要只在第一次做。图数据这种东西错误往往是后期增量更新时带进来的。4. 问答链路从问句到 Cypher 的四步实现4.1 问答整体流程用户输入自然语言问句后要做四件事先做文本预处理把问句中的标点和语气词去掉然后在词典里找到实体接着判断用户问的是症状、药物还是科室最后按意图模板生成 Cypher执行后拼装成中文回答。这一步一步走完才叫一个完整的问答闭环。不少教程只讲到这里但实际工程里还要加两个兜底实体匹配不到时不能报错要回一句「没听懂」查询结果为空时不能直接把空列表抛给前端要告诉用户图谱里暂时没有这个实体的数据。这两个兜底能让 Demo 在展示时不露馅在真实试用时也不会一碰就崩。4.2 实体识别词典匹配与长词优先本项目数据量小没有直接上命名实体识别模型而是把 Neo4j 里所有节点名拉出来组成词典再用正向最大匹配方式从问句中抠实体。这是一种成本最低、效果最可控的方案等数据量变大或问法变野之后再考虑迁移到序列标注模型。def build_entity_dict(session): result session.run(MATCH (n) RETURN labels(n)[0] AS label, n.name AS name) entities [] for record in result: entities.append(record[name]) return entities def longest_entity_match(text: str, entities: List[str]): for name in sorted(entities, keylen, reverseTrue): if name in text: return name return None逻辑说明build_entity_dict把图中所有节点的 name 收集起来组成候选词典。longest_entity_match按名称长度从长到短尝试匹配这是实体识别里最关键的排序如果不排词典顺序靠前的短词会抢先命中比如词典先出现「头痛」问句里的「偏头痛」就会匹配成「头痛」后续查询全歪。长词优先是这里最容易踩的坑后面第 5 章还会专门讲。同义词也要在这一步处理。比如用户说「胃发炎」但图谱里的标准名是「胃炎」匹配不到。常见做法是准备一个synonym_map在做长词匹配前先把问句替换一遍SYNONYM_MAP { 胃发炎: 胃炎, 闹肚子: 腹泻, } def normalize_alias(text: str) - str: for alias, standard in SYNONYM_MAP.items(): text text.replace(alias, standard) return text这个映射表可以直接放在配置里不需要改代码。数据量大了以后可以基于现有节点的别名属性做自动生成但起步阶段手写几十条完全够用。4.3 意图识别关键词模板与优先级意图识别负责回答「用户想查什么」。这里没有训练分类器而是用关键词模板匹配因为在固定三段式问答里意图种类很少规则法的准确率和可解释性都更好。我把意图分成三类| 意图 | 触发词 | 问句示例 | | symptom | 症状、表现、反应 | 支气管炎有什么症状 | | drug | 药、吃什么、治疗 | 偏头痛吃什么药 | | dept | 科室、挂号、看哪个科 | 胃溃疡挂什么科 |实现代码如下INTENT_KEYWORDS { symptom: [症状, 表现, 反应], drug: [药, 吃什么, 治疗], dept: [科室, 挂号, 看哪个], } def infer_intent(text: str) - str: hits [] for intent, words in INTENT_KEYWORDS.items(): if any(word in text for word in words): hits.append(intent) for intent in (drug, dept, symptom): if intent in hits: return intent return unknown逻辑说明先用所有关键词做命中收集再按固定优先级返回。优先级为什么这么排因为「胃溃疡挂什么科」里同时有「溃疡」和「科室」的关键词symptom 关键词也可能被触发如果把 symptom 排前面这个问句就会被错判成查症状。把最具体的意图排在前面是这类规则引擎的基本功。4.4 Cypher 模板与答案返回实体和意图都确定后用模板拼 Cypher。下面是核心问答函数QUERY_TEMPLATES { symptom: ( MATCH (d:疾病 {{name: $ename}})-[:HAS_SYMPTOM]-(s:症状) RETURN s.name AS name ), drug: ( MATCH (d:疾病 {{name: $ename}})-[:USES_DRUG]-(m:药物) RETURN m.name AS name ), dept: ( MATCH (d:疾病 {{name: $ename}})-[:VISIT_DEPT]-(k:科室) RETURN k.name AS name ), } def ask(session, text: str) - str: text normalize_alias(text) entity longest_entity_match(text, ENTITY_LIST) if entity is None: return 没有识别到已知疾病或症状换个说法试试。 intent infer_intent(text) if intent unknown: return 我能回答症状、药物、科室三类问题。 cypher QUERY_TEMPLATES[intent].replace({{, {).replace(}}, }) result session.run(cypher, enameentity) names [record[name] for record in result] if not names: return f图谱里还没有「{entity}」的{intent}数据可以去扩展 CSV 后重新导入。 return f{entity}的常见{症状 if intent symptom else 用药 if intent drug else 科室}{、.join(names)}逻辑说明QUERY_TEMPLATES里的{{name: $ename}}是给 Cypher 留的参数占位符Python 端不需要做字符串替换直接通过session.run(cypher, enameentity)把实体名传进去。这样做能避开所有因引号、特殊字符导致的查询语法问题也是第 5 章要强调的参数化原则。replace({{, {).replace(}}, })只是把模板字典里的双大括号还原成单大括号避免和 Python 的format语法干扰。如果你用 f-string 直接写格式化的 Cypher碰到带引号的实体名就会翻车。这些细节都是实际运行几轮后沉淀下来的经验。4.5 完整跑一遍看链路假设用户输入「偏头痛吃什么药」normalize_alias没有命中同义词原样返回longest_entity_match在长词排序下命中「偏头痛」不会错误匹配「头痛」infer_intent命中 drug 关键词且 drug 优先级最高意图判定为 drugask拼出MATCH (d:疾病 {name:偏头痛})-[:USES_DRUG]-(m:药物) RETURN m.name执行后得到药物列表最终回答「偏头痛的常见用药布洛芬、对乙酰氨基酚」。这个例子走完基本可以确认导入脚本、图谱结构、问答模板三层都是通的。如果某一步断了就用刚才的分步函数单独调不要直接调ask否则错误信息会被兜底逻辑吞掉。5. 避坑五个让知识图谱问答翻车的常见问题这五个坑在本地小数据集里不一定触发一旦触发就会让问答结果看起来像玄学查半天也查不出头绪。我按实际排错顺序把它们列出来每条都按「现象 → 原因 → 解决」讲透。5.1 实体识别把「偏头痛」匹配成「头痛」现象用户问「偏头痛吃什么药」系统却按「头痛」返回结果甚至返回空列表。原因实体词典按原始顺序遍历短词「头痛」排在「偏头痛」前面先被命中。普通词典匹配不会自己区分词长只要in判断命中就立即返回这是最典型的朴素实现缺陷。解决匹配前必须按名称长度倒序排序保证长词优先。实际操作时我在longest_entity_match里用sorted(entities, keylen, reverseTrue)生成候选列表。如果已经匹配错可以再加一层子串保护匹配到实体后检查它是不是更长实体的子串如果是就继续向更长的候选匹配。这样即使词典顺序变化也不会把「偏头痛」拆成「头痛」。5.2 关系方向写反了图里数据正常但问答全空现象Neo4j Browser 里能看到节点和关系数量和类型都对但ask函数永远返回空。这是最折磨人的一个问题因为浏览器的可视化界面不会主动提醒你箭头方向。原因导入时把(疾病)-[USES_DRUG]-(药物)写成了(药物)-[USES_DRUG]-(疾病)查询模板里匹配不到任何边。数据没有丢只是方向反了。解决导入脚本里固定 direction写成MERGE (a)-[rel]-(b)后再用一条汇总查询检查方向。我当时是用MATCH (d:疾病)-[r]-() RETURN type(r), count(*)发现三种关系数量都对但目标类型对不上最后定位到是方向问题。从那以后导入完先跑这条校验再往下走不然后面所有排查都是在猜。5.3 Cypher 里拼字符串导致特殊字符翻车现象问句里带引号、括号时查询直接抛语法错误不带这些字符时又是好的看起来像偶发 bug。原因用 f-string 把用户文本拼进 Cypher例如MATCH (d:疾病 {name:{entity}})。遇到中文引号或英文撇号时Cypher 字符串被提前截断语法就崩了。解决一律用参数化查询session.run(query, enameentity)实体名交给驱动处理。节点标签这类不能参数化的部分用内部白名单变量拼绝不让用户输入直接进标签位置。参数化之后问句里出现什么字符都不用担心这是最省心的方案。5.4 没建唯一约束MERGE 也救不了重复节点现象导入脚本跑完发现「胃炎」这个节点有两个每个节点下只挂了一部分关系问答结果时好时坏。原因第一次导入后重新跑脚本MERGE 没有按预期匹配到已有节点。MERGE 只是尽量查重但如果没有唯一性约束并发或重复执行时依然可能生成重复节点。解决在导入任何数据前先执行 3.2 的约束初始化代码让name字段唯一。先跑约束再跑 MERGE重复脚本最多是幂等更新如果反过来脏数据就落进库里了。已经出现重复节点的要么写一段合并脚本把关系迁移到其中一个节点上要么直接删库重导。小数据量下删库重导往往比重写合并脚本更划算。5.5 大批量导入慢到像死机其实是事务粒度问题现象往图里写几万条关系跑了十几分钟没结束以为是死机。原因每条关系单独 commit 一个事务事务提交开销远超写入本身。Neo4j 的写事务要处理日志、锁、索引同步单条提交一万次和分批提交十次性能差距是数量级的。解决用 UNWIND 把批量写入改成一批一个事务。前面load_relations里每类关系就是一个 UNWIND几千条数据一两个事务就完成了。如果数据量再大可以按 500 到 1000 条分片提交不要开太小的批次也不要一个事务塞几十万条容易把内存打满。出现慢导入时先看事务日志里有没有大量单条 commit 的痕迹再改批次大小。6. 落地前的自测用回归脚本给问答系统兜底6.1 用断言集固定预期前几步做完系统看起来能跑了但这还不够。真正让问答稳定的是回归测试。我的习惯是把高频问题整理到一个小文件里每个问题记录问句、预期意图、预期实体三个字段CASES [ (支气管炎有什么症状, symptom, 支气管炎), (偏头痛吃什么药, drug, 偏头痛), (胃溃疡挂什么科, dept, 胃溃疡), (频繁咳嗽是什么原因, symptom, None), ]这里的None是故意设计的反例它要求系统绝不能把一些常见短句误识别成某个实体避免模糊输入给出错误答案。问句覆盖不到的地方可以按意图再补几条同义问法检验同义词表和优先级是否真正生效。6.2 跑一个最小回归函数不用引额外测试框架一个普通脚本就能完成三层断言def run_regression(cases): passed 0 for text, expected_intent, expected_entity in cases: entity longest_entity_match(text, ENTITY_LIST) intent infer_intent(text) answer ask(session, text) intent_ok (intent expected_intent) entity_ok (entity expected_entity) or (expected_entity is None and entity is None) pass_flag intent_ok and entity_ok and (expected_entity is None or answer ! ) if pass_flag: passed 1 else: print(f失败: {text} | 实体 {entity} | 意图 {intent} | 答案 {answer}) print(f通过率 {passed}/{len(cases)})这个函数把「实体识别、意图识别、查询执行」三层串在一起断言每次改完词典或模板就跑一遍。通过率低于 90% 就不要急着接前端先把实体词典和同义词表补齐。医疗问答这类场景宁可让系统说「不知道」也不能给错答案这是知识型问答的一条底线。如果你手里正好缺一套能跑通的数据和脚本把这份基于 Neo4j 的简易医疗问答知识图谱资源下载下来按第 2 章到第 4 章的顺序过一遍比自己重头造数据省不少时间。那次我把关系方向弄反图里怎么看都有数据问答就是空最后花了两小时才靠回归脚本定位到关系类型上。从那以后每次导入完我都会先跑一遍MATCH (d:疾病)-[r]-() RETURN type(r), count(*)确认方向对再进问答层。规则问答系统不怕笨怕的是数据没对齐。希望帮到你。本文还有配套的精品资源点击获取
