Python基于Neo4j构建知识图谱问答系统：从建图到多跳查询实战

简介这份资源面向具备一定Python基础、希望入门知识图谱与智能问答开发的学习者提供了一套基于Neo4j构建知识图谱并据此实现问答系统的完整源码。包内共40个文件以py脚本、txt词表、png流程图、json数据、pyc缓存及md说明为主压缩包约9.18MB涵盖图谱构建、问题分类、问题解析、答案搜索与聊天入口等核心模块并配有医疗、犯罪等领域的实体与关系数据。目前已有797人学习下载。读者可借此理解从原始语料抽取、图谱入库到自然语言问句解析、答案检索返回的完整链路掌握问题分类器与解析器的实现思路参考图谱路由与问答路由示意图梳理系统架构并借助词表与爬虫脚本自行扩展数据。目录按数据准备、图谱构建、问答处理分层组织适合作为课程设计、毕业设计或知识图谱练手项目的参考模板。1. 从零搭一套能跑的知识图谱问答为什么 Neo4j 是绕不开的那一环很多人第一次接触「Python 基于 Neo4j 构建知识图谱并做问答系统」脑子里想的是先找个大模型接上就完事。真动手才发现问答准不准八成取决于图谱建得对不对而不是模型多强。这个标题讲的是三件事串成一条线用 Python 把结构化或半结构化数据抽成实体和关系写进 Neo4j 图数据库再基于图查询做意图识别和答案生成。它解决的是「关系型数据库查不动多跳关系、纯向量检索答不准事实类问题」这个痛点。适合谁会一点 Python、想给业务加个能解释推理路径的问答能力的后端或数据工程师。下面按建图、查询、问答、避坑、进阶的顺序把每一步落到能复现的命令和参数上。2. 建图之前先把数据模型定死节点、关系与属性怎么切2.1 为什么先画模型再写代码图数据库最怕的就是边写边改模型。Neo4j 里节点和关系一旦写入改标签或关系类型意味着要重跑全量数据。我一般会先在纸上画三样东西实体类型节点标签、关系类型有向边、每个实体挂哪些属性。以常见的领域问答为例实体可能是「疾病」「症状」「药品」「科室」关系是「疾病-有症状-症状」「疾病-用药品-药品」「疾病-挂科室-科室」。属性只放会被查询或展示的字段比如疾病名、药品的通用名和商品名。判断模型合不合理有个土办法把你预期用户会问的 20 个问题列出来逐个在模型上走一遍看能不能用 2 到 3 跳查出来。查不出来的要么缺关系要么关系方向反了。这一步花半小时能省后面几天的返工。2.2 用 Python 把原始数据转成 Cypher 能吃的结构假设原始数据是一份 CSV两列头实体、尾实体、关系类型。先做实体去重和关系归一化再批量写入。下面是最小可跑的建图脚本用官方驱动连库。from neo4j import GraphDatabase import csv # 连接参数bolt 协议默认 7687改过端口要同步 URI bolt://localhost:7687 AUTH (neo4j, your_password) driver GraphDatabase.driver(URI, authAUTH) # 用 MERGE 而不是 CREATE避免重复节点 CREATE_REL MERGE (h:Entity {name: $head}) MERGE (t:Entity {name: $tail}) MERGE (h)-[r:REL {type: $rel}]-(t) def load_csv(path): with open(path, encodingutf-8) as f: reader csv.DictReader(f) for row in reader: yield row[head], row[tail], row[rel] def build(rows): with driver.session() as session: for head, tail, rel in rows: session.run(CREATE_REL, headhead, tailtail, relrel) if __name__ __main__: build(load_csv(triples.csv)) driver.close()逻辑说明MERGE保证同名实体只建一次关系用REL统一类型、把具体关系名放进属性type这样模型不用随关系种类膨胀。参数说明URI走 bolt 协议AUTH是用户名密码元组如果 Neo4j 装在远程把 localhost 换成实际地址并确认 7687 端口放行。数据量大时逐条session.run会很慢常见做法是每 1000 条包一个事务或者用LOAD CSV直接在 Cypher 里导入。2.3 建完图先做一次完整性校验写完不等于写对。跑几条统计查询确认节点数、关系数、有没有孤立节点。// 节点总数 MATCH (n) RETURN count(n) AS node_count; // 关系总数 MATCH ()-[r]-() RETURN count(r) AS rel_count; // 找孤立节点没有任何关系的实体多半是脏数据 MATCH (n) WHERE NOT (n)--() RETURN n.name LIMIT 20;如果孤立节点占比超过 5%回去查 CSV 里是不是有拼写不一致的实体名。这一步不做后面问答会莫名其妙答不出来还找不到原因。3. 从图里查答案Cypher 多跳查询与意图映射3.1 多跳查询是知识图谱问答的命根子关系型数据库做两跳以上关联要写一堆 JOIN图数据库一条 Cypher 就够。比如「得了某病该吃什么药」本质是「疾病节点出发走用药品关系拿到药品节点」。MATCH (d:Entity {name: $disease})-[:REL {type: 用药品}]-(m:Entity) RETURN m.name AS drug参数说明$disease是用户问题里抽出来的实体名type要和建图时写入的关系类型字符串完全一致大小写都不能错。这是最常见的翻车点——建图写「用药品」查询写「用药」结果永远返回空。3.2 把自然语言问题映射成 Cypher 的三种做法第一种是模板匹配把「X 用什么药」「X 有哪些症状」这类句式做成规则抽实体填模板。优点是可控、可解释缺点是覆盖不了没见过的问法。第二种是意图分类加槽位填充用一个小模型或关键词规则判断意图再抽实体。第三种是让大模型直接生成 Cypher灵活但容易生成语法错误或幻觉关系。我一般用前两种打底第三种只作为兜底。原因很实际问答系统上线后用户最不能忍的是答错而不是答不出。模板答不出可以回「暂不支持」大模型乱答会直接毁掉信任。import re # 极简模板匹配意图 - Cypher 模板 TEMPLATES { drug: MATCH (d:Entity {name: $e})-[:REL {type: 用药品}]-(m:Entity) RETURN m.name AS ans, symptom: MATCH (d:Entity {name: $e})-[:REL {type: 有症状}]-(m:Entity) RETURN m.name AS ans, } def detect_intent(question): if re.search(r吃什么药|用什么药, question): return drug if re.search(r什么症状|有哪些表现, question): return symptom return None def answer(question, entity): intent detect_intent(question) if not intent: return 暂不支持该问法 with driver.session() as session: result session.run(TEMPLATES[intent], eentity) return [r[ans] for r in result]逻辑说明detect_intent用正则做粗分类answer按意图取模板执行。参数说明$e是实体名必须和库里存的完全一致实际项目里实体抽取要单独做可以用词典匹配或 NER 模型这里为了聚焦查询逻辑先省略。3.3 查询性能的三个必调参数图小的时候感觉不到数据上百万节点后查询会明显变慢。第一给实体名建索引CREATE INDEX FOR (n:Entity) ON (n.name)否则每次MATCH都是全表扫描。第二控制跳数超过 3 跳的查询要么拆成多次要么加LIMIT。第三用PROFILE看执行计划确认走的是索引而不是 AllNodesScan。CREATE INDEX entity_name_index IF NOT EXISTS FOR (n:Entity) ON (n.name);建完索引再跑一次PROFILE对比 db hits 数量通常能降一个数量级。4. 问答系统落地从查询结果到自然语言回答4.1 答案生成不是简单拼接查到结果只是半成品。用户问「感冒吃什么药」返回[感冒灵, 布洛芬]直接丢出去体验很差。常见做法是套一层回答模板「针对感冒常用的药物有感冒灵、布洛芬。」如果结果为空要区分是「实体没找到」还是「关系没匹配上」前者提示换个说法后者说明图谱覆盖不足。def render(intent, entity, results): if not results: return f没有查到关于「{entity}」的相关信息换个说法试试 joined 、.join(results) if intent drug: return f针对{entity}常用的药物有{joined}。 if intent symptom: return f{entity}的常见症状包括{joined}。 return joined逻辑说明render按意图选模板空结果单独处理。参数说明results是查询返回的列表实际项目里还要做去重和排序比如按关系权重或出现频次排。4.2 多跳问题的拆解策略「某药的副作用对应的症状该挂什么科」这种问题要三跳药→副作用→症状→科室。直接写一条 Cypher 也能查但意图识别会很难。我一般把复杂问题拆成子问题链前一步的结果作为后一步的输入实体逐跳查询。这样每步都可解释出错也知道断在哪一跳。def multi_hop(start_entity, hops): # hops 是 [(关系类型, 返回字段), ...] current [start_entity] for rel_type, _ in hops: nxt [] for e in current: with driver.session() as session: res session.run( MATCH (a:Entity {name: $e})-[:REL {type: $t}]-(b:Entity) RETURN b.name AS n, ee, trel_type) nxt.extend([r[n] for r in res]) current nxt return current逻辑说明逐跳展开每跳的结果集合作为下一跳的起点。参数说明hops是关系类型序列顺序不能乱实际用的时候要加去重和最大跳数限制防止图里出现环导致死循环。4.3 把问答接成服务本地跑通后用 Flask 或 FastAPI 包一层 HTTP 接口前端或其它系统就能调。核心是把 driver 做成全局单例别每次请求都新建连接。from flask import Flask, request, jsonify app Flask(__name__) app.route(/qa, methods[POST]) def qa(): data request.get_json() question data.get(question, ) entity data.get(entity, ) intent detect_intent(question) if not intent: return jsonify({answer: 暂不支持该问法}) with driver.session() as session: result session.run(TEMPLATES[intent], eentity) answers [r[ans] for r in result] return jsonify({answer: render(intent, entity, answers)}) if __name__ __main__: app.run(host0.0.0.0, port5000)逻辑说明接口收问题、抽好的实体走意图识别和查询返回渲染后的答案。参数说明host设0.0.0.0才能被外部访问生产环境要换成 gunicorn 之类的 WSGI 服务器别用 Flask 自带的那套。5. 避坑与排查建图和问答里最容易翻车的五件事5.1 现象查询永远返回空但数据明明写进去了原因关系类型字符串不一致建图写「用药品」查询写「用药」或「治疗」。Cypher 对字符串是精确匹配差一个字都不行。解决统一用常量管理关系类型建图和查询引用同一个常量别手写字符串。5.2 现象Neo4j 本地能连远程连不上原因默认配置只监听 localhost或者防火墙没放行 7687。解决改neo4j.conf里的dbms.default_listen_address重启服务再确认端口放行。这是 neo4j 不能通过 ip 访问 这类搜索里最高频的问题八成出在监听地址上。5.3 现象数据量一大写入慢到无法接受原因逐条session.run每次都开事务开销全花在事务管理上。解决批量提交每 1000 到 5000 条一个事务或者干脆用LOAD CSV让 Neo4j 自己导入速度差几十倍。5.4 现象问答结果里出现重复答案原因图里同一对实体之间存在多条同类型关系或者多跳查询没去重。解决建图时用MERGE保证关系唯一查询结果用DISTINCT或 Python 侧set()去重。5.5 现象复杂问题答非所问原因意图识别把问题分错了类或者实体抽取抽错了。解决先把意图分类的准确率打上去宁可多一类「不确定」走兜底也别硬猜。实体抽取加一层词典校验抽出来的实体在库里不存在就直接提示别往下查。6. 进阶让图谱问答从能用到好用前面搭的是能跑通的最小闭环真要用起来还得补几块。第一块是实体链接用户说的「感冒」和库里的「普通感冒」得对上常见做法是建同义词表加模糊匹配或者用向量相似度做召回。第二块是混合检索纯图查询覆盖不了所有问法把图查询结果和向量检索结果做融合能明显提升召回率这也是 langchain-chatchat 问答检索集成 neo4j 三路混合检索 这类方案火起来的原因。第三块是图谱的可视化前端用现成的图可视化插件把查询路径画出来用户能看到推理链路信任感完全不一样。验证做得好不好我有个习惯准备 50 个真实问题人工标注标准答案每次改完模型或查询逻辑就跑一遍看准确率和召回率的变化。别凭感觉说「好像变好了」数据不会骗人。# 极简评测脚本对比预期答案和实际答案 def evaluate(test_cases): hit 0 for q, entity, expected in test_cases: intent detect_intent(q) if not intent: continue with driver.session() as session: res session.run(TEMPLATES[intent], eentity) actual set(r[ans] for r in res) if set(expected) actual: # 命中任一即算对 hit 1 return hit / len(test_cases)逻辑说明evaluate遍历测试集命中任一预期答案就算对。参数说明test_cases是 (问题, 实体, 预期答案列表) 的列表规模不用大50 条就能看出趋势。我自己踩过最深的坑是早期图省事没建索引数据到十万节点后查询慢到没法演示临时加索引又赶上演示前夜。从那以后建图脚本里索引语句永远放在第一批执行。希望帮到你。本文还有配套的精品资源点击获取