简介：这套Python源码包以neo4j图数据库为核心，完整实现了从医疗、犯罪等领域数据中构建知识图谱，并基于该图谱进行问答检索的系统。资源面向希望学习知识图谱构建、Neo4j应用或问答系统开发的Python开发者，涵盖数据预处理、实体关系抽取、图谱存储、问题解析与答案生成等关键环节。压缩包共40个文件，以Python脚本（.py）为主，配合txt数据字典、json图谱数据、pyc编译文件以及png运行截图、pptx讲解和markdown说明文档，整体体积仅9.18MB，便于下载与本地环境快速部署。目前已有798人学习下载，说明其具有一定的参考价值。通过阅读源码和说明，读者可以掌握基于neo4j搭建知识图谱的完整流程，了解问句分类、意图识别、查询语句构造等问答系统核心组件的实现思路，并借助截图和演示文档快速校验运行效果。

1. 基于neo4j的Python问答系统：不训练模型，靠图谱和规则跑通问答闭环

先用一个场景开场：你在对话框里输入“我最近头晕乏力，应该挂什么科”，这套系统会在几百毫秒内完成实体识别、意图分类、图谱查询和答案拼接，最终返回一句“头晕乏力可能与神经内科相关，建议您前往神经内科就诊”。这套基于neo4j的知识图谱问答系统，把医疗领域的疾病、症状、药品、检查、科室五类实体，以及疾病和它们之间的各种关系全部存进图数据库，再用词典和规则驱动问答。全程不需要训练深度学习模型，却能跑出一个形态完整的问答闭环。它适合两类人：一是想搞懂知识图谱从建图到应用完整链路的开发者，二是想快速搭一版垂直领域问答系统的后端工程师。

2. 项目结构与数据资产：先看清每个文件在链路里的位置

项目拿到手，不要急着跑，先把目录结构过一遍。这套源码的核心是“一条数据流水线”：爬取原始数据 → 清洗成词表和关系对 → 灌入neo4j图谱 → 基于图谱做问答。只要认清每个文件在哪一环，后面调参、改bug、二次开发都会顺手很多。

2.1 目录结构：一条链路上的脚本各自负责什么

展开压缩包后，我习惯先画一张文件映射图，把脚本和它对应的职责挂上钩，结构大致是这样的：

. ├── build_medicalgraph.py # 核心建图脚本，读dict数据入neo4j ├── chatbot_graph.py # 问答入口，串起分类、解析、检索 ├── question_classifier.py # 问句实体识别与意图分类 ├── question_parser.py # 把意图转成Cypher查询语句 ├── answer_search.py # 执行Cypher并组装答案文本 ├── del.py # 清空neo4j图谱数据，便于重复建图 ├── prepare_data/ │ ├── data_spider.py # 从公开医疗词条抓取原始数据 │ ├── build_data.py # 清洗数据，生成词表和关系对 │ └── max_cut.py # 最大匹配分词，辅助问句实体识别 ├── dict/ # 预置实体词典，问答和建图共用 │ ├── disease.txt # 疾病词表 │ ├── symptom.txt # 症状词表 │ ├── drug.txt # 药品词表 │ ├── check.txt # 检查项目词表 │ ├── department.txt # 科室词表 │ ├── food.txt # 食物词表 │ ├── producer.txt # 药品生产商词表 │ └── deny.txt # 否定/禁忌词表 ├── data/ # 数据备份目录 │ └── kg_crime.json ├── img/ # 作者跑出来的截图 │ ├── kg_route.png # 图谱整体概览 │ ├── qa_route.png # 问答链路图 │ └── graph_summary.png ├── document/ # 作者早期的架构讲解PPT └── README.md # 项目说明，先读它

注意document/里放了一份作者早期分享的架构讲解PPT，里面有kg_route和qa_route的完整链路图，对理解设计意图很有帮助。__pycache__下的.pyc是运行时自动生成的编译缓存，可以直接忽略，不影响项目运行。

从这条目录结构能看出作者的分层意识：数据准备、建图、问答三条链路各自独立，build_medicalgraph.py只依赖dict/，问答三个脚本也只依赖dict/和自己的类。这意味着你想换一个领域，只需要替换数据源和词表，问答层基本不用大动。

2.2 dict词表：给实体识别提供弹药

dict/是整个项目的“弹药库”，建图脚本和问答脚本都从这里读数据。每个txt文件一行一个词，例如disease.txt里是“感冒”“高血压”“糖尿病”这类疾病名，symptom.txt里是“发热”“咳嗽”“头晕”这类症状词。

这里有两个容易被忽略的细节。第一个是deny.txt，它存的是“不适宜”“忌口”“不能吃”这类否定表达。医学问答里经常出现“高血压患者不适宜吃什么”这种问法，系统需要靠这个词表判断句子里的否定关系，否则会把“不适宜”直接忽略，给出相反的饮食建议。第二个是producer.txt，生产商词表，它支撑的是“这个药是谁生产的”一类问法，属于补充性的实体维度。

data/下的kg_crime.json和dict/里的crime.txt、crime_big.txt是作者切换实验主题时留下的另一份数据备份，和医疗问答主线无关。跑建图脚本前建议直接忽略它们，避免把无关实体混进图谱。

2.3 数据准备脚本：从爬虫、清洗到分词

prepare_data/下三个脚本对应数据准备的三道工序。data_spider.py负责从公开的医疗词条页面抓取疾病词条，包括疾病名称、症状、治疗方式、常用药品等半结构化的文本；build_data.py把抓回来的原始文本做清洗，抽取实体名和实体关系对，输出成dict/下的词表文件；max_cut.py实现的是一个最大正向匹配分词器，留给问答阶段做问句实体匹配用。

我当时跑build_data.py的感受是，它的清洗逻辑并不复杂，核心就是定好实体和关系的抽取规则。抓回来的原始文本如果是JSON，清洗时大致长这样：

import json raw_data = json.load(open("raw_medical.json", encoding="utf-8")) disease_set = set() symptom_disease_pairs = [] for item in raw_data: disease = item.get("name", "").strip() symptoms = item.get("symptom", "").split("、") if not disease: continue # 疾病词表，去重 disease_set.add(disease) # 疾病-症状关系对，后续建图直接消费 for sym in symptoms: sym = sym.strip() if sym: symptom_disease_pairs.append((sym, disease)) with open("dict/disease.txt", "w", encoding="utf-8") as f: for d in sorted(disease_set): f.write(d + "\n")

逻辑说明：这段代码先读入爬虫产出的原始JSON，item.get("name")取疾病名，item.get("symptom")取症状字段并按顿号切分。疾病名和症状名都被汇总到集合里做去重，最终写出的是dict/disease.txt这样的词表文件。symptom_disease_pairs这个关系对列表则是建图脚本的输入。

参数说明：encode="utf-8"在Windows和Linux下都必须显式指定，否则在Windows上会被默认的GBK编码坑到，后面第5章会细说。split("、")是按中文顿号切分，不同数据源的切分字符可能不一样，处理其他领域数据时记得先确认分隔符是逗号、顿号还是空格。

3. 图谱构建核心：build_medicalgraph.py如何把数据灌进neo4j

建图是整个项目最有价值的部分。build_medicalgraph.py做的事情可以概括为三件事：定义实体类型、定义关系类型、把dict/里的词表数据和关系对灌进neo4j。图谱设计得好不好，直接决定后面问答能回答什么、不能回答什么。

3.1 实体与关系设计：决定图谱长什么样

实体就是图谱里的节点，关系就是节点之间的边，它们共同构成图谱的Schema。这个项目的Schema设计完全围绕医疗问答场景展开。

节点类型有七类：疾病、症状、药品、检查项目、科室、食物、生产商。关系类型则按“疾病和其他实体的联系”来设计，核心关系可以整理成下面这张表：

关系命名是英文驼峰风格，Cypher查询时直接使用，比如(d:Disease)-[:has_symptom]->(s:Symptom)。这套设计把医疗问答里最高频的几类问题都覆盖到了，没有过度设计。

3.2 实体入库：py2neo的Node、merge与去重

实体入库的代码是整个建图脚本的骨架，常见写法是遍历词表，为每个词创建一个Node，再用merge写入。

from py2neo import Graph, Node g = Graph("bolt://localhost:7687", auth=("neo4j", "123456")) def create_disease_node(disease_name, desc=""): """创建一个疾病节点，name是唯一键，merge避免重复入库。""" node = Node("Disease", name=disease_name, desc=desc) g.merge(node, "Disease", "name") return node

逻辑说明：Node("Disease", name=disease_name, desc=desc)的第一个参数是节点标签，后面是键值对属性。g.merge的作用是“有则更新、无则创建”，第三个参数"name"指定了唯一键，也就是说当图谱里已存在同名疾病节点时，不会新建节点，只会更新属性。如果不加这个唯一键，每跑一次脚本就会多出一批重复节点，图谱很快就废了。

参数说明：Graph("bolt://localhost:7687", auth=("neo4j", "123456"))里，第一个参数是Bolt协议连接地址，默认端口7687；auth元组是用户名和密码。如果你本地的neo4j没有改过密码，初始密码是neo4j，第一次登录neo4j会强制要求修改。这里要特别提醒，源码里写死的密码通常和你本地环境不一致，照抄大概率会连不上。

3.3 关系构建与批量提交：Relationship与一次提交

节点建好之后是关系。关系构建的经典写法是先match到起点和终点节点，再创建关系。为了减少网络请求次数，实际项目中通常会先把一批节点和关系放进一个Subgraph，最后一次性merge。

from py2neo import Graph, Node, Relationship def build_relationships(pairs, rel_type): """ pairs是[(起点实体名, 终点实体名), ...]的列表 rel_type是关系类型，例如"has_symptom" """ nodes_to_merge = [] rels = [] for start_name, end_name in pairs: start_node = Node("Disease", name=start_name) end_node = Node("Symptom", name=end_name) nodes_to_merge.append(start_node) nodes_to_merge.append(end_node) rel = Relationship(start_node, rel_type, end_node) rels.append(rel) subgraph = Subgraph(nodes_to_merge, rels) g.merge(subgraph, "Disease", "name") g.merge(subgraph, "Symptom", "name")

逻辑说明：关系必须挂在节点上，所以这里先为每个关系对创建两个临时Node，再用Relationship(start_node, rel_type, end_node)建立关系。把节点和关系一起放进Subgraph，最后仍然走merge。注意merge时分别指定了Disease和Symptom的name作为唯一键，这样即使临时Node的属性不完整，也能正确对齐到图谱里已有的节点上。

参数说明：Relationship第二个参数是关系类型名，字符串里不能带空格或特殊符号；Subgraph接收的是节点列表和关系列表，它的merge逻辑是按标签和唯一键对齐，不是按内存地址对齐。

3.4 建图后的验证：Cypher查询与数据核对

建图脚本跑完后，第一件事是打开neo4j浏览器，用Cypher检查图谱规模。作者在img/里放了两张图：kg_route.png是图谱整体概览，graph_summary.png是节点数统计。你可以自己执行下面的Cypher验证：

MATCH (d:Disease) RETURN count(d) AS disease_cnt; MATCH (:Disease)-[r:has_symptom]->(:Symptom) RETURN count(r) AS relation_cnt; MATCH (n) RETURN labels(n) AS label, count(*) AS cnt ORDER BY cnt DESC;

逻辑说明：第一条统计疾病节点数，第二条统计“疾病-症状”关系数，第三条按标签分组统计各类节点的数量。核对数量时以dict/里的词表为基准：如果disease.txt有200个词，那图谱里Disease节点数应该也是200左右，上下浮动不超过几个，否则说明有重复节点或者有词没匹配上。

这里有个血泪经验：建图脚本如果连续跑两遍，在merge没有生效的情况下节点数会翻倍。作者特意留了del.py，它的作用就是先清空图谱再重建。所以每次重新建图前，我都会先跑一遍del.py，再跑build_medicalgraph.py，保证从干净状态开始。

4. 问答链路拆解：从一句话到查询语句再到人话

图谱只是基础设施，用户实际接触的是问答入口。整个问答链路由三个脚本串联：question_classifier.py理解用户问的是什么，question_parser.py把理解结果翻译成Cypher，answer_search.py执行查询并把结果拼成自然语言。

4.1 实体识别和意图分类：question_classifier.py的核心逻辑

意图分类是整个问答系统的灵魂。它要回答两个问题：这句话里出现了哪些实体？用户到底想问什么？这个项目采用的是词典匹配加正则模板的方案，没有用训练模型。

class QuestionClassifier: def __init__(self): self.disease_words = self.load_words("dict/disease.txt") self.symptom_words = self.load_words("dict/symptom.txt") self.drug_words = self.load_words("dict/drug.txt") self.intent_patterns = { "disease_symptom": re.compile(r"(.+?)有什么症状|症状是(.+?)[吗呢？]"), "disease_cureway": re.compile(r"(.+?)怎么治|怎么治疗|如何治疗"), "disease_drug": re.compile(r"(.+?)吃什么药|推荐什么药"), "disease_belongs_to": re.compile(r"(.+?)挂什么科|去什么科室"), "disease_food": re.compile(r"(.+?)吃什么好|不能吃什么"), } def classify(self, question): entities = self.find_entities(question) for intent, pattern in self.intent_patterns.items(): match = pattern.search(question) if match: return intent, entities return "unknown", entities

逻辑说明：classify方法先调用find_entities找出句子里的疾病、症状、药品等实体，然后按顺序用正则模板匹配意图。(.+?)是非贪婪分组，用来捕获问句里的主体词。比如“高血压应该挂什么科”会命中disease_belongs_to模板，同时find_entities会识别出“高血压”这个疾病实体。

这里的执行顺序是有讲究的：先匹配实体再匹配意图，两个结果都保留。意图决定了走哪条查询路径，实体决定了查询条件里的具体值。如果所有正则都没命中，就返回unknown，由下游做兜底处理。

参数说明：self.load_words是作者封装的一个读词表方法，内部把txt文件的每行读取为列表元素；re.compile预编译了每个意图对应的正则，避免每次调用都重新编译，这个习惯在高并发问答场景里能省不少开销。

4.2 查询模板生成：question_parser.py如何构造Cypher

意图和实体都有了，接下来要生成Cypher。question_parser.py的核心是一个字典工厂，根据不同的question_type输出不同的Cypher模板，并且要求每种类型都必须返回一个固定的结构。

class QuestionPaser: def build_cypher(self, intent, entities): disease = entities.get("disease", "") if intent == "disease_symptom": cypher = ( f"MATCH (d:Disease)-[:has_symptom]->(s:Symptom) " f"WHERE d.name = '{disease}' " f"RETURN s.name" ) elif intent == "disease_drug": cypher = ( f"MATCH (d:Disease)-[:recommend_drug]->(m:Drug) " f"WHERE d.name = '{disease}' " f"RETURN m.name" ) elif intent == "disease_belongs_to": cypher = ( f"MATCH (d:Disease)-[:belongs_to]->(dept:Department) " f"WHERE d.name = '{disease}' " f"RETURN dept.name" ) else: cypher = "" return [{ "question_type": intent, "queries": [{"query_type": intent, "cypher": cypher}] }]

逻辑说明：build_cypher把Cypher语句包装成一个两层结构，外层是question_type和queries列表。queries里每一项都有query_type和cypher两个字段。这样设计的价值在于：一个问句可能对应多条查询链，比如“胃炎有什么症状，该挂什么科”需要查两个子问题，queries数组正好可以放多条Cypher。

参数说明：Cypher里的'{disease}'是直接字符串拼接。这在演示项目里很常见，但如果做成生产系统，最好换成参数化查询，否则用户输入的引号可能会破坏Cypher结构。我在二次开发时会用$disease参数替掉这段拼接逻辑。

4.3 答案组装和对话入口：answer_search.py与chatbot_graph.py

Cypher生成后交给answer_search.py执行，执行结果不能直接返回，要拼装成正常人能读的话。答案组装规则按query_type区分，不同的意图有不同的答案模板。

class AnswerSearcher: def search_main(self, sqls): final_answers = [] for sql in sqls[0]["queries"]: cypher = sql["cypher"] query_type = sql["query_type"] if not cypher: continue results = g.run(cypher).data() if results: answer = self.convert_answer(results, query_type) final_answers.append(answer) if not final_answers: return "这个问题我暂时没有查到答案，换个问法试试" return "\n".join(final_answers) def convert_answer(self, results, query_type): names = [row["s.name"] for row in results] if query_type == "disease_symptom": return f"{names[0]}的常见症状包括：{', '.join(names[:5])}" elif query_type == "disease_belongs_to": return f"该疾病建议前往{names[0]}就诊"

逻辑说明：g.run(cypher).data()返回的是列表，每个元素是一个字典，字典key是Cypher里RETURN的别名。convert_answer根据query_type决定答案模板，names[:5]做了截断，避免症状过多导致答案又臭又长。整个结构是纯函数的风格，输入Cypher列表，输出字符串，方便单元测试。

chatbot_graph.py是最后的主入口，它把三个类串起来，形成完整的处理流水线：

class ChatBotGraph: def __init__(self): self.classifier = QuestionClassifier() self.parser = QuestionPaser() self.searcher = AnswerSearcher() def chat_main(self, sentence): intent, entities = self.classifier.classify(sentence) if intent == "unknown": return "我不太明白你的意思，换个问法试试" sqls = self.parser.build_cypher(intent, entities) answers = self.searcher.search_main(sqls) return answers

逻辑说明：chat_main接收用户原始句子，第一步做分类，第二步生成Cypher，第三步搜索答案。unknown意图在这里被兜底拦截，直接返回提示语，不会往下走。这套分层设计把“理解、查询、表达”三个关注点拆开了，任何一个环节出问题都能单独排查，这在实际调试中非常省力。

5. 踩坑与排查：neo4j连接、版本兼容、问答答非所问

这个项目我前前后后跑过三遍，每一遍都在不同地方翻车。这里挑五个最典型的坑，按“现象、原因、解决”写清楚，帮后来的人省点时间。

5.1 py2neo版本不一致导致API报错

现象：安装依赖后直接跑build_medicalgraph.py，报错要么是ImportError: cannot import name 'authenticate'，要么是Graph()的username/password参数不被识别。

原因：py2neo的API在4.x和5.x之间变化很大。老版本用from py2neo import authenticate，然后用Graph("http://localhost:7474", username="neo4j", password="123456")连接；5.x版本移除了authenticate，连接方式改成了Graph("bolt://localhost:7687", auth=("neo4j", "123456"))。源码很可能是基于旧版本写的，你直接用pip install py2neo装的是最新版，接口对不上。

解决：先看自己装的是哪个版本，再来决定改代码还是锁版本。最省事的做法是统一改成py2neo 5.x的写法，把建图的连接部分替换掉：

from py2neo import Graph # py2neo 5.x 推荐写法 g = Graph("bolt://localhost:7687", auth=("neo4j", "123456"))

如果项目里其他地方用了authenticate，直接删除那行，改上面这一个连接就够。血的教训：不要升级py2neo到6.x再尝试跑老项目，接口变动只会更大。

5.2 中文乱码：Windows下读词表全是乱码字符

现象：dict/disease.txt里的中文词读进Python后变成“鎰熷啋”一类的乱码，建出来的图谱节点名全是问号。

原因：Windows下用记事本另存txt时，默认编码可能是ANSI/GBK，而Python的open()函数默认按系统编码读取。如果代码里没显式encoding="utf-8"，在以UTF-8为默认编码的Linux或新版macOS上，GBK文件读出来必然乱码。

解决：统一两件事。第一，把所有txt文件转成UTF-8编码；第二，只要是读文本文件，一律带上encoding="utf-8"参数：

def load_words(filepath): words = [] with open(filepath, "r", encoding="utf-8") as f: for line in f: line = line.strip() if line: words.append(line) return words

5.3 图谱越建越大：重复实体和重复关系

现象：跑了几遍build_medicalgraph.py之后，用Cypher统计Disease节点数，发现比词表行数多出好几倍。

原因：建图脚本用了create而不是merge，或者merge时没有指定唯一键。每跑一次脚本，所有节点都重新创建一遍，旧的没清掉，新的又进来。

解决：第一，先给实体加唯一约束，从数据库层面拦住重复。在neo4j浏览器里执行：

CREATE CONSTRAINT IF NOT EXISTS FOR (d:Disease) REQUIRE d.name IS UNIQUE;

第二，每次建图前先跑del.py清库。我现在的习惯是：清库、建约束、建图、统计数量，四步走完才敢进行问答测试。

5.4 neo4j连接不上：端口、认证、Driver不匹配

现象：Graph("bolt://localhost:7687", auth=("neo4j", "123456"))一直报ConnectionError或Unauthorized。

原因：常见三处。一是neo4j服务没启动；二是默认密码没改，初始密码是neo4j不是123456；三是Bolt端口和HTTP端口搞混，neo4j的默认端口是7474（HTTP）和7687（Bolt），代码里用的协议和端口不匹配就会连不上。

解决：先用浏览器打开http://localhost:7474确认neo4j进程正常，再手动登录一次验证账号密码。如果密码没改过，先用默认密码登录并按提示修改，然后把代码里的auth改成你自己的密码。端口问题很简单：用了bolt://就配7687，用了http://就配7474，别混。

5.5 意图识别不准：词典缺失与分词切碎实体

现象：输入“我最近老咳嗽，吃什么药好的快”，系统回答“我不太明白你的意思”，或者把“咳嗽”识别成别的实体。

原因：两种可能。第一，dict/symptom.txt里没有“咳嗽”这个词，实体识别直接漏掉；第二，max_cut.py的最大匹配分词在切分时，受语气词干扰，把“咳嗽”从句子中错误切分出去。

解决：先在代码里单独调试实体识别这一步，看find_entities输出了什么。如果词表缺词，直接在dict/symptom.txt追加一行。如果分词被干扰，可以先做一轮轻量的问句清洗，把“我最近”“好的快”这类无关片段去掉，再进入实体匹配。我一般会在分类器里加一个预处理步骤：

def clean_question(self, question): # 去掉口语语气词，减少对实体匹配的干扰 question = re.sub(r"我最近|请问|一下|谢谢|好的", "", question) return question.strip()

参数说明：re.sub的第二个参数替换成空字符串就等于删除匹配片段。这里列举的干扰词是按用户真实问法统计出来的，跑问答测试时你会发现不同的问法干扰词不一样，这份清单要长期维护。

6. 二次开发：把医疗图谱换成你自己的领域数据

理解了这个项目的分层结构，二次开发其实不难。核心思路是：只要你的领域仍然可以用“实体+关系+问句模板”来表达，这套框架就能复用。我拿一个模拟场景来演示改造过程，比如要把医疗问答改成商品知识问答。

第一步，换数据。把dict/disease.txt换成商品类目词表，symptom.txt换成商品属性词表，department.txt换成服务渠道词表。建图脚本里的节点标签和关系类型也要跟着改，比如Disease换成Category，has_symptom换成has_attribute。

第二步，改意图模板。商品问答最关心的是“这个商品有什么功能、适合什么人、价格区间是什么”，所以question_classifier.py里的正则模板要重新写。这一步的核心是把意图枚举和你业务里的真实问法对齐。

第三步，改Cypher模板。question_parser.py里的build_cypher方法和convert_answer是强耦合的关系，二者必须同步改。改完后用一批真实问句做回归验证，确保每个意图都有产出：

test_cases = [ ("这款手机有什么功能", "has_attribute"), ("这款手机适合什么人", "target_user"), ("这款手机多少钱", "price_range"), ] for question, expected in test_cases: intent, entities = classifier.classify(question) assert intent == expected, f"{question} 意图识别失败" sqls = parser.build_cypher(intent, entities) answers = searcher.search_main(sqls) assert answers, f"{question} 查询结果为空"

这段验证脚本每改一次分类器就跑一遍，成本很低，能防住大多数改出来的低级错误。从那以后，我每次拿到这类图谱问答项目，都强制先跑一遍建图脚本确认关系数能和词表对得上，再做问答测试，最后才动意图模板——关系对不上，后面所有模板都是白搭。希望帮到你。

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