企业官网建设流程全解析

简介：这是一份基于Neo4j构建知识图谱并在此基础上实现问答系统的完整源码包，主要面向自然语言处理初学者、知识图谱方向研究生以及医疗问答等垂直领域开发者。资源以疾病、症状、药品、食物、检查项等医疗实体为语料，完整覆盖数据爬取与预处理、分词与实体抽取、图谱构建、问句意图分类、问题解析、答案检索与对话生成链路。压缩包共四十个文件，包含九个源码脚本、十一个文本词典、两个图谱备份、六个缓存文件、一个说明文档、一份演示文稿及多张运行截图，整体约九点一八兆字节；目录按功能划分为图谱构建、问答核心、数据与备份等模块，便于对照调试。现有七百九十八人学习下载。读者可借助其中的图谱构建脚本搭建医学知识图谱，通过问句分类器与问句解析模块理解短文本分类和语义槽填充，再通过答案搜索模块理解问答语句到图查询语言的转换；另附完整医学词典与演示材料，是一套适合教学演示与二次开发的知识图谱问答练习工程。

1. 这份源码包，把知识图谱问答从"存数据"推到了"答提问"

很多人用 Python 和 Neo4j 构建知识图谱时，做完节点导入就用 Cypher 裸查，一旦走到"自然语言提问 → 自动生成查询 → 拼出人话答案"这层就卡住了。这份源码包是一套完整跑通的问答闭环：build_medicalgraph.py负责把疾病、症状、药品、食物等词典数据写入 Neo4j，question_classifier、question_parser、answer_search三个模块分别完成意图识别、Cypher 生成和话术组装，最后由chatbot_graph.py和用户一问一答。看这份源码的价值不在于把代码抄一遍，而在于看清楚"图谱模型设计 → 问题分类 → 查询模板"这三件事怎么串起来。适合三类人：做知识图谱课程设计的开发者、刚接触 Neo4j 的 Python 从业者、想给内部知识库加一个问答入口的工程师。

2. 架构与数据流：先看懂实体模型，再看三层流水线怎么协作

2.1 实体-关系模型：七类节点、四组关系，图谱设计的核心

源码的dict目录下放了 7 个词典文件：disease.txt、symptom.txt、drug.txt、check.txt、food.txt、producer.txt、department.txt。图谱构建脚本按这 7 类实体建节点，再按语义关系建边。实体类型本身不复杂，真正决定问答效果的是关系设计——关系名直接对应到后面question_parser.py里的 Cypher 模板，关系名越多，模板数量跟着涨，所以这份源码刻意把关系收敛到固定的几组：

这种"少关系 + 多实体属性"的设计有一个好处：问答系统不需要理解复杂语义路径，它只需要把用户问句里的实体抽出来，再套上一条固定路径的 Cypher 就能返回结果。比如用户问"高血压不能吃什么"，系统抽到实体"高血压"和意图"忌食"，直接走(Disease)-[:NO_EAT]->(Food)这条边查询，不用临时推理多跳路径。在数据量几千条的规模下，这种简化不是偷懒，而是工程上最稳定的选择。

kg_crime.json的存在说明这套实体-关系模型并不绑定医疗领域。它是一份已结构化的图谱导出数据，把医疗词典换成任何一组实体词典与关系映射，同一套构建脚本和问答链路就能迁移到另一类事实问答场景。后面第 6 章会展开讲换领域的具体步骤。

2.2 三层流水线：分类、解析、检索，解耦是抄它的重点

这套问答系统没有用深度学习模型，流程是经典的规则式三层流水线。第一层由question_classifier.py完成意图识别和实体抽取：用词典对用户问句做包含匹配，命中"什么症状""忌吃""推荐药"这类关键词就把意图打上标签，同时把问句里的疾病词、症状词、药名提取出来。第二层由question_parser.py把意图和实体组合成一条或多条 Cypher 查询语句。第三层由answer_search.py执行查询并把结果包装成自然语言回复。

三层解耦的好处是改起来互不干扰：想加一种新问法，只需要在分类器里加一个意图模板，在解析器里加一条 Cypher 模板，不需要改动检索层。想换数据源，只需要改图谱构建脚本和词典，问答模板一层不用碰。第一次看这份源码时建议按这个顺序读文件：question_classifier.py→question_parser.py→answer_search.py→chatbot_graph.py，从主程序往回追，比按文件名字面顺序看更容易建立整体感。

2.3 源码目录逐文件对照：哪个文件管哪件事

目录里还有编译过的__pycache__缓存文件，从cpython-36和cpython-39混存能看出源码在 Python 3.6 和 3.9 上都跑过。这本身说明兼容性没问题，但有个小隐患：两个版本的 pyc 混在同一个目录里，首次启动解释器会因为缓存版本不一致而跳过部分编译缓存，反应是启动慢一两秒。拿到源码后建议直接删掉__pycache__文件夹，让它按当前环境重新编译，省得排查时被误导。整体看，这套源码的路径依赖很短，核心逻辑就五个 Python 文件加上一个构建脚本，适合完全读懂后按自己的数据改造成通用问答系统。

3. 图谱构建实战：从词典文本到 Neo4j 节点，一条命令跑通

3.1 数据准备：build_data.py怎么把词典加工成三元组

图谱构建的第一步不是写 Cypher，而是准备结构化数据。prepare_data/build_data.py读入dict目录下的纯文本词典，每个词典文件一行一个实体名，输出成带标签的实体列表和头尾关系列表。max_cut.py在这里承担分词工作，它不依赖外部分词库，而是用自定义词典做最大匹配切分，为的是让"过敏性鼻炎"这种长词不被拆成"过敏""性""鼻炎"。

# prepare_data/build_data.py 核心逻辑 import json from max_cut import MaxCut def build_entities(disease_file="dict/disease.txt"): entities = [] with open(disease_file, encoding="utf-8") as f: for line in f: name = line.strip() if not name: continue entities.append({ "name": name, "label": "Disease", "desc": f"{name}相关知识条目" }) return entities def build_relations(disease_list, symptom_list): relations = [] for disease in disease_list: for symptom in symptom_list: # 简化规则：同一句话里两个词都出现则判定存在关联 if MaxCut.match(disease, symptom): relations.append({ "head": disease, "tail": symptom, "rel": "HAS_SYMPTOM" }) return relations

逻辑说明：build_entities把词典里的每一行变成一个字典，label对应 Neo4j 节点标签，name是实体的唯一标识，desc是展示用描述。build_relations的MaxCut.match用共现规则判断两个实体是否存在关系，这是一种弱规则，通常是给后续人工校对提供一个候选集。

参数说明：encoding="utf-8"必须显式声明，否则在 Windows 默认 GBK 环境下读词典文件会直接抛UnicodeDecodeError。head和tail分别对应关系起止点，rel是关系类型名，这三字段的命名要和第 3.2 节里的导入脚本严格保持一致。

这份源码里真正的三元组数据来自data/kg_crime.json这类已校对文件，结构是{"head": "实体A", "tail": "实体B", "rel": "关系类型"}的数组。如果你要换成自己的数据，最省事的做法是照这个 JSON 结构生成一份，而不是去跑共现规则——规则生成的关系噪声很高，我在实际使用中通常是"爬虫 / 公开结构化数据 → 规则生成候选 → 人工抽检校对 → 导出 JSON"四步走，build_data.py只承担前两步。

3.2 批量导入：build_medicalgraph.py的直白写法与事务写法

很多人第一次跑通这套源码时，会直接用graph.create(node)逐条插入。这种写法在几百条数据时没问题，数据量到上万级就会明显变慢，Neo4j 的 HTTP 事务提交次数太多，每条建节点请求都要走完一次完整的网络往返。源码里给的示例更工程化：先delete_all()清库，再分批提交。

# build_medicalgraph.py 核心逻辑 from py2neo import Graph, Node, Relationship def create_all_batch(graph, entities, relations, batch_size=500): # 先清库，避免重复运行导致节点冗余 graph.delete_all() tx = graph.begin() nodes = {} for i, ent in enumerate(entities): node = Node(ent["label"], name=ent["name"], desc=ent.get("desc", "")) tx.create(node) nodes[ent["name"]] = node if (i + 1) % batch_size == 0: tx.commit() tx = graph.begin() tx.commit() for rel in relations: head = nodes.get(rel["head"]) tail = nodes.get(rel["tail"]) if head and tail: tx = graph.begin() tx.create(Relationship(head, rel["rel"], tail)) tx.commit()

逻辑说明：graph.begin()开启一个事务，tx.create()把节点创建操作挂进事务，tx.commit()一次性提交。nodes字典以实体名为键暂存节点对象，建关系时直接用内存里的节点对象创建边，不需要再查库。

参数说明：batch_size=500表示每积累 500 个节点就提交一次。这个值不是拍脑袋定的，实践里 200~500 是平衡网络往返和事务锁粒度的常用区间。值太小会退化成逐条插入，值太大会让单个事务持锁时间变长，并发写入时容易触发LockAcquireTimeout。建关系时单独为每一条开事务，在图谱只有几千条边时是可接受的，如果边超过几万条，建议把关系创建也放进分批事务里。

连接 Neo4j 的地方要注意版本差异：源码早期写法是Graph("http://localhost:7474", username="neo4j", password="neo4j")，这是 py2neo 4.x 的 API。新版 py2neo 改成了auth=("neo4j", "neo4j")。我一般建议统一写成下面这种显式声明协议的方式：

from py2neo import Graph graph = Graph("bolt://localhost:7687", auth=("neo4j", "neo4j"))

bolt://走 7687 端口，性能优于 HTTP 端口。这个修改在第 5 章排查部分还会再提一次，因为它是换环境复现时最常遇到的第一个报错点。

3.3 构建结果验证：用 Cypher 确认节点数和关系干净

导入完成后不要急着跑问答，先用 Cypher 做三个基础校验：

-- 检查实体总数是否与词典去重后一致 MATCH (n) RETURN count(n); -- 检查疾病与症状关系是否成规模 MATCH (d:Disease)-[:HAS_SYMPTOM]->(s:Symptom) RETURN d.name, s.name LIMIT 20; -- 检查孤立节点数量 MATCH (n) WHERE NOT (n)--() RETURN count(n) AS orphan_count;

第一条count(n)如果和词典实体去重后的数量对不上，多半是build_data.py的输出有重复或空行没过滤；第二条抽样看关系是否合理，比如"感冒"连接了"咳嗽""流鼻涕"这类症状才算正常；第三条孤立节点数如果很高，说明关系构建规则覆盖太差，后面问答时用户问疾病相关症状会频繁得到空结果。这三个查询在 Neo4j Browser 里跑一遍就行，不用写脚本，这也是排查阶段最顺手的手段。

4. 问答链路实现：意图识别、Cypher 组装与话术生成

4.1 问题分类器：用词典命中而不是训练模型，数据量小的时候规则更稳

question_classifier.py没引入任何机器学习库，它做的事是两件：从问句里抽实体，给问句打意图标签。实体抽取用词典包含匹配，意图识别用关键词模板匹配。数据量在千条级时，这种方式比训练一个意图分类模型更可靠，因为医疗问句的表达相对固定，"忌吃""症状是""推荐用"这类短语覆盖了绝大多数实际提问。

# question_classifier.py 核心结构 class QuestionClassifier: def __init__(self): self.entity_dicts = { "disease": self._load("dict/disease.txt"), "symptom": self._load("dict/symptom.txt"), "drug": self._load("dict/drug.txt"), "food": self._load("dict/food.txt"), "department": self._load("dict/department.txt"), "check": self._load("dict/check.txt"), } self.intent_patterns = { "disease_symptom": ["什么症状", "症状是什么", "表现是什么"], "symptom_disease": ["得了什么病", "可能是什么病", "什么疾病"], "disease_no_eat": ["忌吃", "不能吃", "别吃", "不宜吃"], "disease_recommend_eat": ["宜吃", "推荐吃", "可以吃"], "disease_recommend_drug": ["推荐什么药", "用什么药", "吃什么药"], } def classify(self, question): matched_entities = {} for label, words in self.entity_dicts.items(): for word in words: if word in question: matched_entities[label] = word for intent, patterns in self.intent_patterns.items(): if any(pattern in question for pattern in patterns): return {"intent": intent, "entities": matched_entities} return {"intent": "undefined", "entities": matched_entities}

逻辑说明：entity_dicts从 5 个文本词典里加载实体词表，classify先遍历所有词表做子串匹配，命中就记为实体；再遍历intent_patterns，按关键词模板匹配意图。意图字典的 key 就是后面question_parser.py要处理的事务类型。

参数说明：intent_patterns的字典遍历顺序就是优先级顺序。把更具体的表达放在前面，比如"不能吃"要放在"吃"前面，否则"不能吃什么"会被先匹配到"可以吃"意图导致结果反了。这是这套源码里最容易被忽略的细节，我在扩展意图时吃过亏，后面第 5 章会单列一条排查记录。

词典加载函数_load建议做两个处理：一是给每个词做strip()去掉换行符，二是跳过空行，否则词典文件末尾多一个空行会导致匹配时把空字符串当成实体名，虽然查询不出错，但容易在日志里看到奇怪的空键。

4.2 问题解析器：把意图和实体翻译成 Cypher

question_parser.py是问答链路里最关键的一层。它接收分类器输出的意图和实体，拼接出可执行的 Cypher。源码里的做法是按意图类型维护一组模板，每个模板用字符串格式化把实体名填进去。

# question_parser.py 核心逻辑 class QuestionParser: def transfer(self, intent, entities): cqls = [] if intent == "symptom_disease": symptom = entities.get("symptom") if symptom: cqls.append( "MATCH (d:Disease)-[:HAS_SYMPTOM]->(s:Symptom) " "WHERE s.name = '%s' RETURN d.name AS d" % symptom ) elif intent == "disease_no_eat": disease = entities.get("disease") if disease: cqls.append( "MATCH (d:Disease)-[:NO_EAT]->(f:Food) " "WHERE d.name = '%s' RETURN f.name AS f" % disease ) elif intent == "disease_recommend_drug": disease = entities.get("disease") if disease: cqls.append( "MATCH (d:Disease)-[:RECOMMENDED_WITH]->(dr:Drug) " "WHERE d.name = '%s' RETURN dr.name AS dr" % disease ) return cqls

逻辑说明：每个意图分支都会先判断需要的实体是否抽到，实体不存在就直接跳过，避免生成一条必空结果的 Cypher。%s字符串拼接是源码原生的写法，实体名来自受控词典，理论上没有注入风险，但如果把问答服务暴露成 Web 接口，一定要换成参数化查询，用graph.run(cql, name=symptom)替代字符串拼接。

参数说明：这里的 Cypher 模板有几个命名约定。关系名HAS_SYMPTOM、NO_EAT、RECOMMENDED_WITH必须和图谱构建脚本里建关系时用的大小写完全一致，Neo4j 的关系类型是大小写敏感的；RETURN d.name AS d的别名要和answer_search.py里取字段用的 key 对齐，否则.data()返回的字典里没有d这个键，会直接 KeyError。这两类问题在换数据、改模板时几乎必定遇到，我自己的经验是每次改动后先在 Neo4j Browser 里手跑一遍生成出的 Cypher，确认有结果再放回代码。

4.3 答案检索与话术组装：让查询结果说人话

answer_search.py负责把解析器生成的 Cypher 逐条执行，再把结果拼成可读的回复。这一步的知识点在于处理"多结果去重"和"空结果兜底"。

# answer_search.py 核心逻辑 class AnswerSearch: def __init__(self, graph): self.graph = graph def search_main(self, cqls): final_answers = [] for cql in cqls: if not cql or not cql.strip(): continue result = self.graph.run(cql).data() if not result: final_answers.append("暂时没查到相关记录，换个问法试试。") continue answer_items = [] for row in result: for key, value in row.items(): if value and value not in answer_items: answer_items.append(value) final_answers.append( "根据知识图谱，找到 " + "、".join(answer_items) ) return final_answers

逻辑说明：graph.run(cql).data()返回的是列表套字典结构，字典的 key 对应 Cypher 里的别名。这里对结果做了两层防御：一是空结果返回兜底话术；二是按 value 去重，因为同一实体可能通过多条路径被查到，比如"感冒"同时出现在两个科室下。chatbot_graph.py主程序就是一个while True循环，把三个模块串起来：

# chatbot_graph.py 交互主循环 class ChatBotGraph: def __init__(self): self.classifier = QuestionClassifier() self.parser = QuestionParser() self.answer = AnswerSearch() def chat(self, question): res = self.classifier.classify(question) if res["intent"] == "undefined": return "我暂时不太理解这个问题，可以换成带疾病名或症状名的问法。" cqls = self.parser.transfer(res["intent"], res["entities"]) answers = self.answer.search_main(cqls) return "\n".join(answers) if answers else "没有找到相关答案。"

chat方法的输出会被主循环打印。交互层写得很薄，好处是方便改成 Flask 接口或者其他对话框架的对接层，只要输出这个字符串，上层展示方式随意换。资源里附带的chat1.png、chat2.png就是这套交互的输出截图，可以作为回归测试的预期对照。

5. 排查手记：换环境复现这套源码时最容易踩的五个坑

这套源码在作者原本的环境里跑得很顺，但换一台机器、换一个 Neo4j 版本，甚至换一个 Python 小版本，就会冒出各种看似玄学的问题。下面按出现频率从高到低记录五个坑，每条都是现象到根因到解决的完整链路。

5.1 py2neo 连不上 Neo4j 4.x / 5.x

现象：运行build_medicalgraph.py直接报ValueError: Invalid port，或者连接时报Unauthorized，有时候还会提示缺少auth参数。

原因：源码里连接 Neo4j 的写法是Graph("http://localhost:7474", username="neo4j", password="neo4j")，这是 py2neo 4.x 的 API 风格。py2neo 升级后把用户名密码两个参数合并成了auth=(user, passwd)，默认连接协议也从 HTTP 切换成了 Bolt。老代码在 4.x 驱动下运行，连接参数被当作 HTTP 端口解析，自然失败。

解决：统一改成Graph("bolt://localhost:7687", auth=("neo4j", "neo4j"))。如果不想改源码，也可以固定安装旧版 py2neo，但我不推荐，旧版对 Python 3.9 以上的兼容性有问题。用新版驱动的好处是以后换 Neo4j 5.x 也不需要再动连接层代码。

5.2 中文实体词被拆碎，意图识别翻车

现象：问"过敏性鼻炎患者不能吃什么"，分类器把实体抽成了"过敏"而不是"过敏性鼻炎"，结果查询返回空。

原因：两个层面。一是如果分类器用了 jieba 分词且没加载自定义词典，"过敏性鼻炎"会被标准词典切成"过敏"+"性"+"鼻炎"，实体匹配时只命中短词；二是即使不用 jieba，直接做字符串包含匹配时，代码是从词表顺序遍历的，短词先命中就把长词覆盖了。

解决：两种手段都补上。一是在_load里把词典按词长倒序排序，让长词优先参与匹配；二是把dict目录下所有词合并成一个自定义词典，在分类器初始化时执行jieba.load_userdict("dict/all_words.txt")。从那以后我凡是做规则分类器，第一个动作就是检查"最长词优先"是否生效，这个排序优先级比意图模板顺序更容易被忽略。

5.3 Cypher 返回的键名和你取数的别名对不上

现象：answer_search.py里row["d"]报 KeyError，偶尔不报错但结果列表全是空字符串。

原因：graph.run("MATCH ... RETURN d.name").data()返回的字典键是"d.name"而不是"d"。只有写成RETURN d.name AS d时，data 里的键才是d。源码模板里写了AS d，但如果自己改模板时漏掉别名，就会踩这个坑。

解决：给每个 RETURN 字段起别名，同时把取数代码写成防御式：row.get("d") or row.get("d.name")。这个防御代码看起来很丑，但在多轮迭代改模板时能省掉大量排查时间。正规做法是用参数化查询避免拼接，但防御取数仍然建议保留，因为 Neo4j 驱动对返回字段名的处理在不同版本间有过细微差异。

5.4 Windows 下读取词典文件乱码导致构建失败

现象：build_data.py读dict/disease.txt时报UnicodeDecodeError: 'gbk' codec can't decode byte，或者构建出的节点名是一堆乱码。

原因：Windows 下 Python 的默认编码是 GBK，而源码里的词典文件是 UTF-8 无 BOM 格式。直接open(path)不带编码参数，Python 会用 GBK 去解 UTF-8 字节流，遇到多字节字符就崩。

解决：所有读取文件的open()都显式加encoding="utf-8"。更稳妥的做法是把词典文件统一转成 UTF-8 with BOM，这样 Python 和记事本都能正确识别，代码里即使漏写encoding参数也大概率能读对。源码里build_data.py已经写了encoding="utf-8"，但如果你在question_classifier.py之外自己写了读词典的逻辑，很容易忘掉这一行，这是我实际踩过两次的坑。

5.5 重复导入让图谱节点翻倍，问答结果出现重复

现象：同一个build_medicalgraph.py跑了两遍，count(n)从 4000 变成了 8000，问答里同一答案重复出现。

原因：构建脚本开头虽有delete_all()，但如果手动中断了第一轮导入，第二次运行时清库逻辑没执行就被跳过；另外源码没有给name字段建唯一约束，Neo4j 不会主动拒绝重名校节点的创建。

解决：两条腿走路。构建脚本开头强制graph.delete_all()，不依赖手动清理；再给每个实体类型的name字段建唯一约束，这样就算清库失败，重复导入也会报错而不是静默产生冗余节点。约束语法注意版本差异：

-- Neo4j 3.x CREATE CONSTRAINT ON (n:Disease) ASSERT n.name IS UNIQUE; -- Neo4j 4.x 及以上 CREATE CONSTRAINT FOR (n:Disease) REQUIRE n.name IS UNIQUE;

版本差异在 Neo4j 4.x 之后尤其明显，关键字从ON变成了FOR，断言语法从ASSERT变成了REQUIRE。跑约束语法报错时，先确认自己装的 Neo4j 大版本，别急着怀疑源码数据有问题。

6. 换领域验证：把同一套问答链路迁移到自己的数据集上

源码里那份kg_crime.json就是为换领域准备的现成样本。它是一组已经结构化好的三元组数据，格式和build_medicalgraph.py期望的输入完全一致：每个对象包含head、tail、rel三个字段。把它当成验证集来跑一遍，能同时检验图谱导入、意图模板、Cypher 映射三件事是否都改对了。

换领域时只需要动三个位置。第一是dict目录下的词典文件，把医疗实体换成目标领域实体，这一步决定实体抽取能不能命中。第二是build_medicalgraph.py里的标签和关系名映射，比如把Disease换成Case、把HAS_SYMPTOM换成HAS_FACT，保证建图时用的关系名和问答模板里的一致。第三是question_parser.py里的 Cypher 模板，把节点标签和关系名替换成新领域的名字。常规做法是先用一份十几条的小数据验证图谱构建，再把对话模板逐一跑通，最后才灌全量数据。

# 自定义关系映射示例 REL_MAP = { "case_factor": "HAS_FACT", "case_law": "REFERS_TO", "case_person": "INVOLVES", } def transfer_custom(intent, entities): cqls = [] rel = REL_MAP.get(intent) if not rel: return cqls head = entities.get("case") if head and rel: cqls.append( "MATCH (c:Case)-[r:%s]->(n) " "WHERE c.name = '%s' RETURN n.name AS n" % (rel, head) ) return cqls

注意这里的关系类型名是动态拼进 Cypher 的，Python 字符串拼接时要保证REL_MAP里的值不包含用户输入，只来自代码常量，否则有注入风险。

每次改完词典或模板，我会强制走一遍"最小回归五问"：用症状查疾病、用疾病查忌食、用疾病查推荐药、用科室查疾病、问一句不在任何模板里的随机话。五条跑完，能确认实体抽取、意图优先级、Cypher 别名、话术兜底四个环节都没坏，才会继续加新意图。这套习惯是从一次翻车里学来的——当时只验证了新加的意图，没回头跑旧用例，结果发现已有一个模板的新别名没同步到answer_search.py，静态证据根本看不出来，跑一轮对话立刻暴露。从那以后我每次拿到知识图谱问答源码，都会先删掉__pycache__，按 README 顺序跑一遍最小回归，再动任何代码。希望你也能用这套流程，少走我走过的弯路。

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