简介：一个基于neo4j的简易医疗问答知识图谱项目，数据从ask120医疗问答平台爬取，整理疾病、症状、治疗方法等实体关系并入库，适合希望学习知识图谱建模、Neo4j使用及医疗文本处理的开发者参考。压缩包共37个文件，以Python脚本为主（13个.py），涵盖爬虫采集、数据清洗和Django后端逻辑，另有pyc编译文件、xml配置、html页面及少量工程资源，总大小仅78KB，结构紧凑便于快速浏览。已有5460人学习下载，可作为相关课程设计或入门案例。资源包含spider1.py、spider2.py等采集脚本，以及manage.py、settings.py、views.py等标准Django模块，通过qaprocess、dbopt等封装可看到从数据导入到问答接口响应的完整实现，方便直接运行、调试并扩展医疗实体关系查询功能。

1. 基于neo4j的简易医疗问答知识图谱：先别急于建图，先想清楚问什么

做过医疗问诊demo的工程师都有体会：最难的不是往neo4j里灌数据，而是把“用户想问什么”翻译成一条能查出正确答案的cypher。标题里“简易”两个字很容易让人误以为只写几个csv再import就能交差，实际上实体识别、意图分类、cypher模板这三件事没理顺，图建得再漂亮也没法回答“高血压挂哪个科”。这个项目走的正是这条务实的路子：用公开的医疗知识子集，构建包含疾病、症状、药物、科室、检查的六元组本体，再配合HanLP词典分词加规则模板实现问答。适合刚入门知识图谱的开发者、做毕业设计或转岗AI工程的人复现，不涉及大模型训练，全部代码在本地就能跑通。它能解决的核心问题只有一个：让你看清知识图谱问答系统的完整链路，知道哪一步耗时最长、哪一步最值得优化。

2. 环境选型与数据准备：neo4j版本坑比想象中多

2.1 neo4j Desktop还是Community Server：版本决定后续API

选择neo4j版本是这个项目第一个容易翻车的地方。neo4j 3.5和4.x、5.x在驱动协议、cypher语法、索引创建方式上都有差异，很多教程代码在3.5上跑得好好的，换到4.4就报错。这个项目我建议直接装neo4j Desktop桌面版，它内置了版本管理，可以在3.5和4.4之间一键切换，对学习和复现最友好。如果一定要装Community Server，注意JDK版本配套：neo4j 3.5需要JDK 8，neo4j 4.4需要JDK 11，neo4j 5.x需要JDK 17，装错版本服务根本起不来。

另一个需要决策的是Python驱动。neo4j官方提供了neo4j python driver，同时也保留了py2neo这个第三方库。对于简易项目我推荐py2neo，原因是它封装了Graph、Node、Relationship对象，写起来直观，排错也容易。代价是py2neo 2021.2版本之后更新变慢，对neo4j 5.x支持不完整，所以这个项目建议锁定neo4j 4.4 + py2neo 2021.2.3这个组合，稳定且资料最多。

启动neo4j Desktop后新建项目时，把数据库密码设置成容易记住的，比如neo4j123，后面所有连接代码都用这个凭据。很多人喜欢设成复杂密码，结果每次写连接串都要翻记事本，完全没有必要。

2.2 医疗数据从哪来：字段清理比找数据更花时间

这个项目不提供完整医疗数据包，需要自己准备。常见做法是从开放医疗知识库获取数据子集，比如CMeKG公开的疾病-症状-药物关联。数据不需要多，两三百条疾病记录配上几千条三元组关系就足够demo运行。关键是字段设计要贴近问答场景，不能直接拿原始csv往neo4j里灌。

我一般会把数据拆成三个csv：nodes.csv存实体及其类型，relations.csv存实体间关系，属性用JSON字符串存在实体节点上。字段设计如下表：

这里要特别注意一点：原始数据里“高血压”可能出现两种写法，比如“高血压病”和“高血压”，如果直接导入neo4j，图谱里会出现两个实体节点，问答时用户问“高血压挂什么科”可能匹配到空结果。解决办法是导入前先做一次归一化，把同一实体的不同别名合并成一行，这一步看着简单，实际要花掉整个数据处理时间的三成以上。

3. 本体设计与批量写入：六元组关系如何映射到cypher

3.1 医疗问答的知识本体：六个关系类型够用但不凑合

这个简易问答系统能回答的问题类型，在设计本体那一刻就已经决定了。我建议定义六个关系类型：症状、常用药物、就诊科室、检查项目、禁忌药物、疾病别名。对应的问题模板分别是“XX有什么症状”“XX吃什么药”“XX挂什么科”“XX需要做什么检查”“XX不能吃什么药”“XX是什么病”。

别贪多。初学者最容易犯的错误是一口气把十几类关系全部建上，结果每个关系的数据量都不足，问答准确率反而低。六个关系类型对应六类高频问题，医疗问诊场景覆盖度已经够演示。把每个关系类型写清楚它的头和尾实体类型，比如：

在neo4j里创建约束和索引是建图的第一步。以下脚本在neo4j Browser中执行，为实体ID和名称建立唯一约束，这是后续按名字查询能走索引的前提：

CREATE CONSTRAINT ON (d:Disease) ASSERT d.id IS UNIQUE; CREATE CONSTRAINT ON (d:Disease) ASSERT d.name IS UNIQUE; CREATE INDEX ON :Disease(name); CREATE INDEX ON :Symptom(name); CREATE INDEX ON :Drug(name);

这段脚本的逻辑是：唯一约束保证同一个疾病不会出现两个节点，普通索引加快按name匹配的速度。注意4.x版本的语法里CREATE CONSTRAINT ON已经弃用，应该用CREATE CONSTRAINT FOR (d:Disease) REQUIRE d.id IS UNIQUE，但py2neo的某些旧版本不兼容新语法，所以项目锁定4.4后就用老语法最省事。如果后续打算升级到neo4j 5.x，这些constraint和index脚本全部要改写，这也是我不推荐在这类入门项目里直接用5.x的原因。

3.2 py2neo批量写入：为什么逐条CREATE慢到无法接受

数据清洗完成、约束建好之后，最直接的导入方式是逐条执行CREATE语句。但几百个实体加上几千条关系，逐条写入的时间是按分钟起步的，而且如果中途报错，前面写入的数据需要手动清理。正确做法是把数据按关系类型分组，用UNWIND配合参数列表批量写入。

下面这段py2neo脚本是写入疾病到药物关系时的核心逻辑：

from py2neo import Graph graph = Graph("bolt://localhost:7687", auth=("neo4j", "neo4j123")) def batch_create_relations(rel_type, start_label, end_label, triples): """ triples 格式: [{"start": "高血压", "end": "硝苯地平"}, ...] 每条三元组先按名字找到节点ID，再建立关系 """ statement = """ UNWIND $pairs AS pair MATCH (s:{start_label} {{name: pair.start}}) MATCH (e:{end_label} {{name: pair.end}}) MERGE (s)-[r:{rel_type}]->(e) RETURN count(r) AS created """.format(start_label=start_label, end_label=end_label, rel_type=rel_type) result = graph.run(statement, pairs=triples) print(f"{rel_type} 创建 {result.value('created', 0)} 条")

这里的关键参数是UNWIND解包出来的pair列表，每条pair包含start和end两个名字。先用MATCH定位起点和终点节点，再用MERGE而不是CREATE建立关系，这样重复执行不会产生重复关系。format字符串拼接标签名是py2neo里比较常见的写法，因为标签不能作为参数传递，只能通过格式化方式嵌入cypher语句。需要留意的是，这里拼接的标签名必须来自白名单，不能用用户输入直接拼，否则存在cypher注入风险。

4. 问答实现：从问题到cypher的映射链路

4.1 实体识别：HanLP词典分词比训练模型更划算

问答系统第一步是从用户问题里抽出疾病名。这个项目不降低门槛去接大模型接口，而是用HanLP的词典分词方案。给HanLP指定一个自定义词典，把nodes.csv里所有实体名字加载进去，分词时就能优先切分出这些词。对比用BERT做序列标注，词典方案对“简易”二字最贴切，不需要GPU，不需要标注数据，分割准确率对常见疾病名足够。

HanLP加载自定义词典的标准做法是这样：

from pyhanlp import HanLP, JClass # 自定义词典路径，每行一个实体名 CustomDictionary = JClass("com.hankcs.hanlp.dictionary.CustomDictionary") with open("medical_dict.txt", encoding="utf-8") as f: for line in f: word = line.strip() if word: CustomDictionary.add(word) text = "高血压患者出现头晕应该挂什么科" terms = HanLP.segment(text) entities = [term.word for term in terms if term.word in medical_entity_set]

这段逻辑里medical_entity_set是预先加载到Python内存的实体名字集合，目的是过滤掉自定义词典切出来的非实体词，比如“出现”“应该”这类词也会被切出来，但不在实体集合里就不会被当成疾病名。词典文件medical_dict.txt每行一个词，utf-8编码，不要带BOM头，否则第一个词会多一个不可见字符导致匹配失败。

实体识别跑通后用Neo4j查询确认实体在图谱里存在，这是排错的好习惯。查询语句很简单：

MATCH (n {name: "高血压"}) RETURN labels(n), n.name LIMIT 1

如果返回空，说明数据写入阶段有问题，或者用户问题里的疾病名和图谱里的名字不一致。应对办法是维护一个同义词映射表，比如“高血压”和“高血压病”都映射到图谱里的标准名“高血压”。

4.2 意图分类与cypher模板：参数化查询是底线

识别出实体之后，紧接着要做意图分类。这个项目的实现方式不复杂，用关键词规则匹配。按问题里是否出现“什么药”“挂什么科”“什么症状”“不能吃”等词，分别映射到不同cypher模板。

规则表可以做成Python字典：

INTENT_TEMPLATES = { "症状": { "patterns": ["什么症状", "表现", "会有哪些"], "cypher": "MATCH (d:Disease {{name:$name}})-[:症状]->(s:Symptom) RETURN s.name LIMIT 10" }, "常用药物": { "patterns": ["什么药", "吃什么", "怎么治"], "cypher": "MATCH (d:Disease {{name:$name}})-[:常用药物]->(drug:Drug) RETURN drug.name LIMIT 10" }, "就诊科室": { "patterns": ["挂什么科", "去哪个科", "看什么科"], "cypher": "MATCH (d:Disease {{name:$name}})-[:就诊科室]->(dep:Department) RETURN dep.name LIMIT 5" }, "检查项目": { "patterns": ["检查", "查什么", "做哪些"], "cypher": "MATCH (d:Disease {{name:$name}})-[:检查项目]->(c:CheckItem) RETURN c.name LIMIT 10" }, "禁忌药物": { "patterns": ["不能吃", "禁用", "忌用"], "cypher": "MATCH (d:Disease {{name:$name}})-[:禁忌药物]->(drug:Drug) RETURN drug.name LIMIT 10" } }

执行时先判断用户问题命中了哪个意图，再用参数化方式执行cypher。参数化是必须养成的习惯，原因有两个：一是防止cypher注入，二是neo4j对参数化查询有缓存，重复执行相同模板时性能明显提升。

def answer_question(question): entity = extract_entity(question) if not entity: return "没认出疾病名，换个说法试试" intent = classify_intent(question) if not intent: return "这个问题我没学会，试试问症状或用药" template = INTENT_TEMPLATES[intent]["cypher"] result = graph.run(template, name=entity).data() if not result: return f"图谱里没找到{entity}的{intent}信息，可能数据还没补齐" answers = [record[list(record.keys())[0]] for record in result] return "、".join(answers[:5])

注意cypher模板里用了{{name:$name}}这种双层花括号写法，这是因为外层.format()转义会导致单括号失效，所以在Python字符串里需要用{{ }}。$name是cypher的参数占位符，执行时graph.run的第二个参数传入实际值。返回结果为空时，要区分是实体不存在还是该实体缺少这个关系的数据，这两类问题提示语不一样，用户能直观感受到系统是否可靠。

5. 常见问题与避坑记录：这些坑我逐个踩过

5.1 neo4j启动失败：版本与JDK不匹配

现象：neo4j Desktop点击start后，数据库状态一直显示installing或直接变回stop，浏览器打开bolt端口无响应。

原因：绝大多数情况是neo4j版本需要特定JDK版本，项目第一次跑时用了默认JDK 17去启动neo4j 3.5，导致启动脚本直接报错。另外Community Server如果之前装过其他版本的neo4j服务，注册的Windows服务可能会抢占7687端口。

解决：neo4j Desktop里进入项目Settings面板，确认当前使用的JAVA_HOME路径指向匹配的JDK版本。neo4j 3.5用JDK 8，4.4用JDK 11。手动检查端口占用：netstat -ano | findstr 7687，找到PID后taskkill /PID，再重启neo4j服务。

5.2 中文乱码问题：csv文件保存格式的坑

现象：用py2neo写入中文数据后，在neo4j Browser里查询节点，显示的是\uXXXX这样的转义字符，或者直接是乱码方块。

原因：csv文件保存时用了默认的UTF-8带BOM格式，或者用了GBK编码。py2neo读取csv时按UTF-8解析，BOM头会混入第一个字段名，导致neo4j里创建的属性名变成空字符加原名。GBK编码的文件更严重，中文字符直接解析失败。

解决：所有csv文件统一用UTF-8无BOM编码保存。用Python的open函数读取时显式指定encoding="utf-8"，如果发现第一个字段名有异常，用csv.reader读取前先用codecs模块去掉BOM。这个小问题在数据量小的时候不明显，一旦塞进几百条记录，排查起来相当费时间。

5.3 cypher查询走不上索引：为什么越查越慢

现象：实体数据量到了3000以上，按name查询某个疾病，从图数据库查询要2秒以上，明显不正常。

原因：唯一约束创建在id字段上，name字段只建了普通索引，但cypher查询里用了name属性直接匹配。某个查询语句没有带上索引提示，neo4j优化器选择了全表扫描。另一个是被忽略的原因是py2neo执行graph.run时，传参类型不是字符串而是Python的整数或布尔值，导致类型不匹配无法走索引。

解决：用EXPLAIN或PROFILE先看执行计划，确认是不是出现了NodeByLabelScan。执行PROFILE MATCH (d:Disease {name:"高血压"}) RETURN d查看输出部分是否显示IndexScan。如果没用索引，尝试改写查询：MATCH (d:Disease) WHERE d.name = $name，强制让优化器考虑name索引。参数类型保持str，不要传int进去。

5.4 py2neo关系写入报错：烘焙节点与游离节点

现象：用graph.create()一个节点对象时报错，提示“node is not attached to a graph”或者“Cannot create node, node already exists”。

原因：py2neo的Node对象在create之后，会被附加到一个Graph实例。第二次运行同一个脚本时，仍然用同一个Node对象去create，py2neo会认为节点已存在而抛出异常。很多人中招是因为把节点创建和关系创建写在同一个循环体里，第二次迭代复用了上一次的Node引用。

解决：所有Node和Relationship都通过merge或者match返回新引用，不要复用。稳妥写法是在循环体内部每次调用NodeMatcher去匹配目标节点，匹配不到才新建。批量导入时直接用UNWIND方式，绕开Node对象的生命周期管理，这也说明第3节里推荐UNWIND不是没有理由的。

6. 进阶：让问答具备多轮对话能力与查询性能验证

问答基础版做好了，再往前一步就是多轮会话。最简单也最实用的做法是在当前会话里缓存上一轮识别出的实体，如果用户新问题里没有识别出实体，就默认沿用上一轮的实体。比如用户先问“高血压有什么症状”，系统回复之后用户再问“那吃什么药”，此时新问题里没有疾病名，程序自动把实体替换为“高血压”，意图切换到常用药物模板。这个逻辑用Flask写接口时尤其好实现，在session里存一个last_entity字段即可。

除了多轮，性能验证环节也值得做。知识图谱问答上线前至少跑一次批量压测，准备五十个问题，统计每个问题的响应时间。我习惯把结果按查询类型分组，观察哪类cypher模板耗时最长。如果“症状”类查询普遍超过800毫秒，优先看是不是索引缺失。用下面这段py2neo脚本可以快速收集慢查询：

import time test_questions = [ ("高血压有什么症状", "症状"), ("高血压吃什么药", "常用药物"), ("高血压挂什么科", "就诊科室"), ] for question, intent in test_questions: start = time.time() answer = answer_question(question) elapsed = (time.time() - start) * 1000 print(f"{intent} | {elapsed:.0f}ms | {answer[:20]}")

响应时间超过1秒的，用neo4j Browser的PROFILE命令手动执行对应cypher，重点看是不是出现Expand(All)这种全库扩展。经验之谈是，医疗图谱数据量如果在几千节点级别，走name索引的查询应该在100毫秒内完成，超出这个量级基本可以断定索引没生效或者查询写法有问题。

最后一件事，也是我每次做完这类项目都会强制自己走一遍的流程：把第一次跑通的全部脚本整理成一个build_all.py，从csv清理到图谱构建再到问答验证一条龙执行。从那以后，每次改数据字段或者换neo4j版本，我都强制走一遍这个脚本，能在五分钟内确认整套链路是否还健康。希望这个流程和这篇笔记帮到你。

本文还有配套的精品资源，点击获取