电影知识问答系统实战：Neo4j图库+Flask接口+小程序前端全链路拆解

简介这份资源是面向计算机专业学生与知识图谱初学者的大作业/毕业设计参考项目围绕电影领域构建知识问答系统采用Python技术栈结合Neo4j图数据库完成实体与关系的存储和查询可帮助读者理解从数据采集到问答交互的完整链路。压缩包共55个文件约5.77MB包含11个py脚本、10个csv数据表、8个json配置、5个txt说明及svg、js、png、md等前端与文档素材覆盖后端Flask服务、爬虫模块、微信小程序前端页面与工具函数目录按backend、spider、frontend等模块划分结构清晰便于按需查阅。目前已有423人学习下载适合需要快速搭建知识图谱问答原型、参考实体识别与关系抽取实现思路、或寻找可运行课程设计案例的读者也可作为理解图数据库建模与前后端联调的实践素材。1. 电影知识问答系统拆包从 Neo4j 图库到 Flask 接口再到小程序前端拿到这个压缩包的时候我第一反应是「终于有人把知识图谱大作业做成了能跑通的三段式」。它不是那种只丢一个 Jupyter Notebook 的作业而是把后端 Flask、图数据库 Neo4j、微信小程序前端串成了一条完整链路。解压后能看到backend、spider、frontend三个目录外加两份 README结构上已经替你把「数据从哪来、存到哪、怎么查、怎么展示」四个问题都摆好了位置。适合正在做知识图谱课程设计、毕业设计或者想找一个能改造成自己领域问答系统的同学。它解决的核心问题是把电影领域的实体和关系灌进 Neo4j用 Cypher 做多跳查询再通过 Flask 暴露接口给小程序调用。下面我按实际复现顺序拆一遍重点讲清楚每个环节的参数和容易翻车的地方。2. 环境与数据链路Neo4j 装完先别急着导数据2.1 三个目录各自负责什么spider目录负责从公开电影页面抓取原始数据产出结构化字段backend是 Flask 应用里面包含图数据库连接、Cypher 查询封装和 REST 接口frontend是小程序端负责把用户输入的问题发给后端并渲染答案。这个分工意味着你不需要一次性理解全部代码可以按「先让 Neo4j 有数据 → 再让 Flask 能查 → 最后让前端能显示」的顺序推进。常见做法是先在本地把 Neo4j 跑起来确认浏览器能访问http://localhost:7474再动 Python 环境。很多人一上来就pip install一堆包结果图数据库连不上排查方向全乱。2.2 Neo4j 安装与配置里最容易忽略的两个参数Neo4j 社区版足够跑这个项目。安装后重点看conf/neo4j.conf里两行# 允许远程访问默认只监听 localhost dbms.default_listen_address0.0.0.0 # Bolt 协议端口Flask 通过它连接 dbms.connector.bolt.listen_address:7687第一行决定你能不能从虚拟机或另一台机器访问第二行是 Python 驱动实际使用的端口。改完必须重启 Neo4j 服务否则配置不生效。如果你在 Mac 上用 Homebrew 装配置文件路径通常在/opt/homebrew/var/neo4j/conf/neo4j.conf别去改安装目录里的模板文件。提示Neo4j 4.x 和 5.x 的默认认证方式不同5.x 首次登录强制改密码。如果 Flask 连接报authentication failure先确认密码不是默认的neo4j/neo4j。2.3 用 Cypher 建约束再灌数据不管spider抓下来的数据是 CSV 还是 JSON导入前先建唯一约束否则重复节点会让后续查询结果爆炸。在 Neo4j Browser 里执行// 给电影和人物分别建唯一约束 CREATE CONSTRAINT movie_title IF NOT EXISTS FOR (m:Movie) REQUIRE m.title IS UNIQUE; CREATE CONSTRAINT person_name IF NOT EXISTS FOR (p:Person) REQUIRE p.name IS UNIQUE;约束建好后用LOAD CSV导入// 导入电影节点假设 CSV 有 title、year、rating 三列 LOAD CSV WITH HEADERS FROM file:///movies.csv AS row MERGE (m:Movie {title: row.title}) SET m.year toInteger(row.year), m.rating toFloat(row.rating);MERGE而不是CREATE是关键它保证同一部电影不会因为多次执行导入脚本而产生多个节点。file:///指向 Neo4j 安装目录下的import文件夹CSV 必须放在那里不能随便放桌面。2.4 验证数据是否真的进去了导入完成后别急着启动 Flask先在 Browser 里跑一条查询MATCH (m:Movie)-[:ACTED_IN]-(p:Person) RETURN m.title, p.name LIMIT 10;如果返回空结果说明关系没建上。常见原因是 CSV 里演员名和电影名对不上或者关系导入语句里MATCH写成了MERGE导致创建了孤立节点。这一步花五分钟确认比后面调 Flask 接口时报 500 要省事得多。3. Flask 后端把 Cypher 查询封装成 HTTP 接口3.1 连接 Neo4j 的驱动初始化backend里通常有一个db.py或neo4j_conn.py核心是创建驱动实例。我一般会这样写from neo4j import GraphDatabase class Neo4jConnection: def __init__(self, uri, user, password): self.driver GraphDatabase.driver(uri, auth(user, password)) def close(self): self.driver.close() def query(self, cypher, parametersNone): with self.driver.session() as session: result session.run(cypher, parameters or {}) return [record.data() for record in result]uri填bolt://localhost:7687user和password填你改过的 Neo4j 账号。record.data()会把结果转成字典列表方便 Flask 直接jsonify。注意session.run返回的是惰性结果必须在with块内消费完否则连接释放后拿不到数据。3.2 问答接口的意图识别逻辑这个项目的问答不是自由文本理解而是基于模板匹配。用户问「周星驰演过哪些电影」后端会先匹配关键词「演过」和「哪些电影」然后拼出 Cypherdef build_cypher(question): if 演过 in question and 电影 in question: person extract_person(question) return ( MATCH (p:Person {name: $name})-[:ACTED_IN]-(m:Movie) RETURN m.title AS title, m.year AS year ORDER BY m.year DESC, {name: person} ) # 其他意图分支...extract_person通常用 jieba 分词加实体词典匹配。这里的关键是参数化查询$name由驱动做转义避免 Cypher 注入。如果你直接把用户输入拼进字符串遇到带引号的人名就会报语法错误。3.3 启动 Flask 并测试接口cd backend export FLASK_APPapp.py export FLASK_ENVdevelopment flask run --host0.0.0.0 --port5000--host0.0.0.0是为了让同一局域网内的小程序开发工具能访问。启动后用 curl 测一条curl -X POST http://localhost:5000/qa \ -H Content-Type: application/json \ -d {question: 周星驰演过哪些电影}如果返回{answer: [...]}说明链路通了。返回 500 时先看 Flask 控制台堆栈八成是 Cypher 语法或参数名对不上。3.4 跨域与小程序请求域名配置小程序开发阶段需要在微信开发者工具里勾选「不校验合法域名」否则http://localhost:5000会被拦截。正式发布则必须 HTTPS 域名并备案。frontend里的app.js或utils/request.js通常有一个baseUrl变量改成你 Flask 的实际地址即可。注意Flask 默认只监听 127.0.0.1如果小程序在真机上调试必须改成0.0.0.0并确保手机和电脑在同一网段。4. 前端小程序与查询联调从输入框到答案渲染4.1 页面结构和请求封装frontend的pages目录下一般有一个index页面包含输入框、发送按钮和结果列表。utils里封装了request方法统一处理wx.request的 success 和 fail。核心调用长这样// utils/request.js const baseUrl http://localhost:5000; function askQuestion(question) { return new Promise((resolve, reject) { wx.request({ url: ${baseUrl}/qa, method: POST, data: { question }, header: { Content-Type: application/json }, success: res resolve(res.data), fail: err reject(err) }); }); }Content-Type必须和后端request.get_json()匹配否则 Flask 收不到 body。小程序端不支持localhost以外的裸 IP 在正式环境使用开发阶段用localhost没问题。4.2 答案渲染的两种格式后端返回的answer可能是字符串列表也可能是包含title、year的对象列表。前端用wx:for渲染时要做兼容// pages/index/index.js Page({ data: { answerList: [] }, onSend() { askQuestion(this.data.inputValue).then(res { const list Array.isArray(res.answer) ? res.answer : [res.answer]; this.setData({ answerList: list }); }); } });如果后端返回的是对象数组wx:for里用item.title取值如果是纯字符串数组直接{{item}}。联调时先确认后端返回结构再改前端模板不要两边同时猜。4.3 用 Neo4j Browser 反查前端结果是否正确前端显示「周星驰演过 36 部电影」你怀疑数字不对直接在 Browser 里跑MATCH (p:Person {name: 周星驰})-[:ACTED_IN]-(m:Movie) RETURN count(m);如果 Browser 返回 36说明数据没问题前端显示异常是渲染逻辑问题如果 Browser 返回 40说明后端 Cypher 漏了条件。这种「前端 → 后端 → 图库」逐层反查的习惯能帮你快速定位问题在哪一层。4.4 多跳查询的 Cypher 写法电影问答里常见「周星驰和吴孟达合作过哪些电影」这需要两跳MATCH (p1:Person {name: 周星驰})-[:ACTED_IN]-(m:Movie)-[:ACTED_IN]-(p2:Person {name: 吴孟达}) RETURN m.title, m.year ORDER BY m.year;注意方向ACTED_IN从 Person 指向 Movie所以中间节点是 Movie两边箭头都指向它。如果方向写反查询会返回空。这是 Neo4j 新手最容易翻车的地方之一。5. 避坑与排查这五个问题我全踩过5.1 现象Neo4j Browser 能查Flask 报连接超时原因Neo4j 的 Bolt 端口没监听在0.0.0.0或者防火墙拦了 7687。解决检查neo4j.conf里dbms.connector.bolt.listen_address:7687重启服务后用telnet localhost 7687确认端口通。5.2 现象导入 CSV 后节点数翻倍原因用了CREATE而不是MERGE或者唯一约束没建就导入。解决先MATCH (n) DETACH DELETE n清空建好约束再重新导入。清空操作在生产环境慎用本地随便。5.3 现象小程序请求返回 400原因Content-Type没设成application/json或者 body 不是合法 JSON。解决在wx.request的 header 里显式声明并用JSON.stringify包一层 data。5.4 现象Cypher 查询结果顺序每次不一样原因没写ORDER BY。Neo4j 不保证默认返回顺序。解决在 Cypher 末尾加ORDER BY m.year DESC或ORDER BY m.title前端展示才稳定。5.5 现象中文人名匹配不到原因CSV 编码不是 UTF-8或者 Neo4j 导入时没指定编码。解决用file命令确认 CSV 编码必要时iconv -f GBK -t UTF-8转换后再导入。6. 进阶技巧用参数化查询和索引把响应压到 200ms 内6.1 给高频查询字段建索引电影问答里最常查的是Person.name和Movie.title。除了唯一约束自带的索引还可以显式建CREATE INDEX movie_year IF NOT EXISTS FOR (m:Movie) ON (m.year);year不是唯一字段用INDEX而不是CONSTRAINT。建完后用EXPLAIN看执行计划EXPLAIN MATCH (m:Movie {year: 1995}) RETURN m.title;如果输出里有NodeIndexSeek说明索引生效如果是AllNodesScan说明没走索引需要检查索引是否建成功。6.2 用参数化查询复用执行计划Flask 里每次拼字符串会导致 Neo4j 重新编译 Cypher响应时间波动大。改成参数化后执行计划会被缓存def query_movies_by_year(year): cypher MATCH (m:Movie {year: $year}) RETURN m.title ORDER BY m.title return conn.query(cypher, {year: year})我实测同一个查询参数化后第二次调用比第一次快 40% 左右。对于问答系统这种高频重复查询场景这个优化几乎零成本。6.3 限制返回条数防止前端卡死有些问题会命中大量结果比如「列出所有电影」。Cypher 里加LIMIT 50Flask 层再做一次截断def safe_query(cypher, params, limit50): result conn.query(cypher, params) return result[:limit]前端渲染 50 条和 500 条的体验差距很大小程序列表过长会明显掉帧。这个习惯是我被卡顿教做人之后养成的现在不管查什么先加LIMIT再说。6.4 验证优化效果的方法改完索引和参数化后用 Neo4j Browser 的PROFILE而不是EXPLAIN看实际耗时PROFILE MATCH (p:Person {name: 周星驰})-[:ACTED_IN]-(m:Movie) RETURN m.title ORDER BY m.year DESC;输出里db hits越低越好time单位是毫秒。我一般会把优化前后的db hits记下来确认真的降了再收工。从那以后我每次改完 Cypher 都强制走一遍PROFILE不然根本不知道改了个啥。希望帮到你。本文还有配套的精品资源点击获取