简介：面向计算机相关专业毕业生的知识图谱电影问答系统项目，基于Python与Neo4j实现，覆盖后端服务、爬虫采集、小程序前端与知识图谱展示，能够帮助快速搭建完整可演示的毕业设计，适合正在准备毕业设计、课程设计或期末大作业的学生参考模仿。压缩包共59个文件，约6.28MB，主要包含11个Python脚本、10个CSV数据文件、8个JSON配置、小程序前端相关JS/WXML/WXSS，以及PNG/SVG架构图和MD说明文档，目录结构清晰便于按模块阅读。目前已有77人浏览学习。项目为导师指导下的高分毕业设计，评审获得99分，代码完整可直接运行；随附文档对数据爬取、知识图谱构建、问答匹配等核心环节有清晰交代，配合项目架构图与工作流程示意图，能帮助读者快速理解项目全貌，节省从零搭建环境与调试的时间。

1. 知识图谱电影问答系统：拿 Python 和 Neo4j 把电影数据做成能对话的知识引擎

这套知识图谱电影问答系统，是典型的 Python 毕业设计完整源码：后端用 Flask 提供接口，数据层用 Neo4j 存储电影实体与关系，前端还带一个微信小程序。它的亮点不是做个电影列表展示，而是把用户问句翻译成图谱查询，比如“某位导演拍过哪些电影”能直接返回结构化结果，看起来像在跟一个懂电影的知识库对话。适合正在选 Python 毕设题目的学生，以及想练图数据库落地的开发者。我拆完一遍后发现，真正的难点不在爬虫，而在实体关联和问答模板的设计，整套代码把这条链路完整串起来了。

2. 先把数据灌进图谱：爬虫与 Neo4j 数据层的一次完整拆分

2.1 三个模块的职责边界：spider、backend、frontend

拿到压缩包解压后，主目录是MovieKGQA-master，里面三个一级目录刚好对应数据层、服务层、展示层。我一般会先看目录结构再决定从哪开始跑，避免一上来就瞎改代码。

这个拆分是典型的“数据采集-数据处理-前端展示”三层结构。最容易被忽略的是backend/flask和spider之间的数据格式约定：爬虫抓下来的是什么字段，后端 Neo4j 查询时就要用什么属性名。如果两者字段对不上，最典型的翻车就是前端问“某导演拍过哪些电影”，后端返回空数组，但 Neo4j 里明明有数据。

我拆解时先看 README 里对数据字段的描述，再去spider目录找存储字段的字典，最后核对 Neo4j 的节点属性。这个顺序能省掉大量瞎猜时间。源码包里还带了assets目录和几张工作流程图，例如MovieKGQA_workflow_graph.png，说明文档和流程图画得比较完整，适合答辩时直接讲架构。

2.2 用 Python 把电影数据写进 Neo4j：实体、关系与属性设计

电影问答系统之所以选 Neo4j 而不是 MySQL，是因为问答场景里大量查询是“查关系”的，比如“演员 A 和导演 B 合作过哪些电影”。图数据库把实体和关系直接存成边，查询深度不受多表 JOIN 限制。这个选型是毕设里最站得住脚的技术亮点。

用 Python 操作 Neo4j 常见有两种驱动：py2neo和官方neo4j-driver。这个项目源码倾向用后者，因为 Neo4j 4.x 之后官方驱动对事务支持更稳，而py2neo在 4.x 以上版本兼容性比较玄学。我一般建议用官方neo4j包，连接串写成常量放在配置文件里。

from neo4j import GraphDatabase # 连接参数：bolt 默认端口是 7687，不是 7474 uri = "bolt://localhost:7687" user = "neo4j" password = "123456" # 换成你 Neo4j 里实际设置的密码 driver = GraphDatabase.driver(uri, auth=(user, password)) # 导入一条电影-演员关系 def create_movie_actor(session, movie_title, actor_name): query = """ MERGE (m:Movie {title: $movie_title}) MERGE (a:Person {name: $actor_name}) MERGE (a)-[:ACTED_IN]->(m) """ session.run(query, movie_title=movie_title, actor_name=actor_name) with driver.session() as session: create_movie_actor(session, "电影A", "演员A")

代码里的MERGE很关键，它避免重复创建同一个节点。如果换成CREATE，跑两次爬虫就会得到两套重复节点，问答结果也会跟着翻倍。这里的实体标签用了Movie和Person，关系类型用ACTED_IN，是 Neo4j 电影知识图谱里的经典命名方式。

导入大批量数据时，不要用循环一个 session 跑几千次，那样每次都要开启事务，速度很慢。常见做法是分 batch 提交，比如每 500 条用一个事务。属性类型也要提前想清楚：电影评分在 Neo4j 里最好存成 Float，上映年份存成 Integer，这样 Cypher 里能用m.score > 8.0做数值筛选，而不是对字符串做转换。

2.3 Flask 后端访问 Neo4j：连接配置与 Cypher 查询封装

后端backend/flask里的主文件一般是app.py，它要做三件事：初始化 Flask、连接 Neo4j、提供问答接口。源码里把数据库连接写成了独立函数，这是值得借鉴的地方，后续扩展接口时不用重复创建驱动实例。

from flask import Flask, request, jsonify from flask_cors import CORS from neo4j import GraphDatabase app = Flask(__name__) CORS(app) # 微信小程序调试时必须放开跨域 URI = "bolt://localhost:7687" AUTH = ("neo4j", "123456") driver = GraphDatabase.driver(URI, auth=AUTH) def run_query(query, params=None): with driver.session() as session: result = session.run(query, params or {}) records = [record.data() for record in result] return records @app.route("/api/qa", methods=["GET"]) def qa(): question = request.args.get("question", "") if not question: return jsonify({"code": 400, "msg": "question 不能为空", "data": []}) # 这里接入问答转换逻辑，下一章详细说 cypher = "MATCH (m:Movie) RETURN m.title AS title LIMIT 5" data = run_query(cypher) return jsonify({"code": 200, "msg": "ok", "data": data}) if __name__ == "__main__": app.run(host="0.0.0.0", port=5000, debug=True)

代码里把run_query单独抽出来，好处是对接 Neo4j 的逻辑只藏在一个地方，后续所有接口都复用这个函数。参数传入用params字典，不要直接拼字符串进 Cypher，否则容易产生注入问题，虽然毕设系统不会真被攻击，但是答辩时这么说加分。

CORS(app)是另一个容易漏的点。微信小程序的wx.request请求后端接口时，如果后端不返回跨域头，浏览器或者开发者工具会直接拦截，页面表现就是接口报错。这里用flask_cors一次性解决，比手动加after_request处理更省事。

3. 问答系统核心：把一句话转成 Cypher 的三步套路

3.1 问句分类与模板设计：先定意图，再找实体

问答模块是整套系统最出彩的地方。它的逻辑不是 NLP 模型，而是“模板匹配 + 实体提取”。先分析用户问题属于哪种类型，再把问题里的电影名、人名、类型名提取出来，最后拼出 Cypher 查询语句。

常见的问题类型有：查询某演员演过的电影、查询某电影的导演、查询某类型的电影推荐、查询评分大于某个值的电影、查询某部电影的演员阵容。每种类型对应一个 Cypher 模板，模板里留出实体名的位置。

在设计模板时，我习惯把一个问句拆成两部分：意图关键词和实体锚点。比如“周星驰主演的电影有哪些”，“主演”是意图，“周星驰”是实体。更稳的方案是先用意图关键词确定模板范围，再用正则把实体抓出来，避免“主演”和“导演”混淆。

3.2 生成查询的 Python 实现：正则提取与 Cypher 拼接

这一部分源码的核心是一个qa_parser.py或者类似文件，里面定义了一套规则列表。每条规则包含一个正则模式和对应的 Cypher 模板。

import re RULES = [ { "intent": "actor_movies", "pattern": r"(.+?)主演的电影有哪些", "cypher": "MATCH (p:Person {name: $name})-[:ACTED_IN]->(m:Movie) RETURN m.title AS title, m.score AS score" }, { "intent": "movie_director", "pattern": r"电影《?(.+?)》?的导演是谁", "cypher": "MATCH (m:Movie {title: $title})<-[:DIRECTED]-(p:Person) RETURN p.name AS director" }, { "intent": "genre_movies", "pattern": r"推荐(.+?)类型的电影", "cypher": "MATCH (g:Genre {name: $genre})<-[:BELONGS_TO]-(m:Movie) RETURN m.title AS title ORDER BY m.score DESC" } ] def parse_question(question): for rule in RULES: match = re.search(rule["pattern"], question) if match: entity = match.group(1).strip() return rule["intent"], entity, rule["cypher"] return None, None, None

这里的正则顺序就是匹配顺序，我一般会把带“推荐”的规则放在前面，因为它对实体的限制更宽松，放太靠后容易被前面的规则误吃。(.+?)是非贪婪匹配，能尽量少吞字符，但也会出现边界问题，比如“电影《霸王别姬》的导演是谁”可能匹配到“电影《霸王别姬》”，所以模板里我特意写了《?(.+?)》?来兼容带书名号和不带书名号两种情况。

实体提取出来之后还要做一步归一化。用户输入“周星驰”和数据库里的“周星驰”如果完全一致就没事，但有时候导入数据时名字前后带了空格，或者用了繁体，这就会导致查不到。源码里常见做法是统一strip()和标准化，再没有结果时可以试一下模糊匹配，但毕设阶段通常直接要求用户输入库里的标准名。

3.3 调试问答准确率：优先匹配顺序与实体归一化

模板匹配最大的坑是规则覆盖不全。我测试时发现，用户不会按你写好的问题格式去问，比如“电影A的导演是谁”和“谁导演了电影A”是同一个意思，但需要两条规则。所以源码里的 RULES 列表是可扩展的，新增一种问法就加一条规则，不需要改核心逻辑。

调试的时候要关注两点：一是接口日志里parse_question返回的intent是否为空，二是生成的 Cypher 参数名和模板里$name是否一致。如果正则匹配到实体，但 Cypher 里参数写成了$actor，Neo4j 会报参数错误。常见做法是在parse_question里把entity一并返回，再在查询函数里组装参数。

4. 小程序端联调：从接口到页面的请求链路

4.1 小程序页面结构与参数传递

frontend目录是一个标准的微信小程序工程，包含app.json、pages、utils、app.js。页面目录通常只有两个：index输入问题，result展示答案。app.json里注册了页面路由，小程序启动时默认加载第一个页面。

这个小程序的交互逻辑比较简单：用户在输入框输入一个问题，点击“提问”，把问题字符串通过wx.request发到 Flask 后端，后端返回 JSON，前端用setData更新data里的列表，渲染到页面。

4.2 封装请求工具函数并解析后端数据

utils里一般会放一个request.js，把wx.request封装成 Promise。这个封装对项目联调很有帮助，因为不用在每个页面重复写失败回调。

// utils/request.js function request(url, params) { return new Promise((resolve, reject) => { wx.request({ url: url, data: params, method: 'GET', success: (res) => { if (res.statusCode === 200 && res.data.code === 200) { resolve(res.data.data); } else { reject(res.data.msg); } }, fail: (err) => reject(err) }); }); } module.exports = { request };

这里对返回结构的约定很关键：后端返回code、msg、data三层，前端只拿data。如果后端直接返回数组，前端解析时就要多做一层判断。保留统一外层结构，后续加分页参数、错误提示都方便。

在index.js里，请求代码大致是：

// index.js const { request } = require('../../utils/request.js'); Page({ data: { question: '', answerList: [] }, onQuestionInput(e) { this.setData({ question: e.detail.value }); }, async onSubmit() { const question = this.data.question; if (!question) return; const data = await request('http://localhost:5000/api/qa', { question }); this.setData({ answerList: data }); } });

4.3 联调排错：域名校验、端口、返回结构

我见过很多跑不起来的情况，不是后端代码错，而是小程序端请求不到本地服务。微信开发者工具有一个“不校验合法域名…”的开关，调试本地 IP 时必须在详情里勾选，否则wx.request会被拦。

另外要注意模拟器里的localhost指向的是开发者工具所在的电脑，不是手机。如果用真机预览，需要改成电脑的局域网 IP，比如http://192.168.1.10:5000/api/qa。Flask 启动时要监听0.0.0.0，不能只监听127.0.0.1，否则真机访问不到。

还有一个容易出问题的点是网络超时。Neo4j 查询如果数据量大、模板没命中，返回可能超过小程序默认 60 秒超时，但实际现象是请求直接失败。排查时先在后端接口用浏览器访问一下，看返回时间，再决定要不要优化 Cypher 查询加索引。

5. 避坑手册：六个最容易让项目跑不起来的问题与排查

5.1 Neo4j 连不上：启动、地址与认证

现象：Flask 启动时报neo4j.exceptions.ServiceUnavailable: Failed to establish connection。

原因：Neo4j 默认端口是7687，很多人会记成7474，后者是浏览器管理界面的 HTTP 端口。另一个常见原因是 Neo4j 桌面版启动后数据库还没完全 ready，立刻连就会失败。

解决：启动时先确认 Neo4j 的 Bolt 端口状态，可以用cypher-shell或者浏览器打开bolt://localhost:7687。密码不对时，把连接常量里的密码改成 Neo4j 安装时设置的密码，不要用 README 里的默认值强连。

5.2 中文乱码与关系导入不完整

现象：查询结果里电影名变成???，或者某个演员的关系只有几条，明显少了很多。

原因：大部分爬虫保存 CSV 时没有指定utf-8-sig，或者 Neo4j 导入时默认编码不一致。关系不完整则很可能是MERGE时节点标签或名称属性有细微差异，比如一个叫“导演A ”，另一个叫“导演A ”带空格。

解决：爬虫写入 CSV 时统一加encoding="utf-8-sig"；导入前对属性做strip()。导入后跑一条MATCH (n) RETURN count(n)看节点总数是否合理，再随机抽查一条电影的全部关系。

5.3 模板匹配不到、返回为空

现象：接口返回 200，但data是空数组。

原因：正则模板没匹配上问句，或者匹配上了但是实体不在库里。常见原因是用户输入了全角括号、冒号，或者把“哪些”说成“有哪些”就不匹配。

解决：在parse_question里加日志，打印匹配到的intent和entity。如果entity为 None，说明是正则问题；如果实体有值但查不到，需要查 Neo4j 里实体名称是否一致。另外一个稳妥做法是在RULES最后加一条兜底模板，返回“暂不支持该问题”，而不是给用户空列表。

5.4 Flask 返回中文乱码和跨域

现象：浏览器请求接口返回的中文是乱码，小程序被跨域拦截。

原因：Flask 默认 JSON 序列化可能没有指定ensure_ascii=False；跨域头没有配置。

解决：在app.py里配置app.config["JSON_AS_ASCII"] = False，并显式加上CORS(app)。如果不想引入flask_cors，可以自己写after_request加头，但容易漏掉预检请求。

5.5 重复数据导致问答结果翻倍

现象：问“演员A演过哪些电影”，返回里同一部电影出现两次。

原因：导入脚本用了CREATE而不是MERGE，或者MERGE没匹配到唯一属性导致创建了重复节点。

解决：把导入语句统一改为MERGE，并给Movie.title、Person.name建唯一约束。建约束的 Cypher 是CREATE CONSTRAINT ON (m:Movie) ASSERT m.title IS UNIQUE。

5.6 微信小程序请求一直 pending

现象：模拟器里请求状态一直在pending，很久之后报超时。

原因：可能是后端启动失败，也可能是前端请求地址写错，还可能是因为端口被防火墙挡了。

解决：先确认 Flask 终端有没有收到请求日志。如果没收到，说明请求根本没到后端，检查小程序地址；如果收到了但响应慢，主要查 Cypher 查询和 Neo4j 索引，尝试加CREATE INDEX FOR (m:Movie) ON (m.title)。

6. 进阶实践：用 pytest 给问答接口做回归验证

如果说运行源码是及格线，那么给问答系统加一组自动化测试就是能把项目分数往上拉的做法。我拆这个项目时，第一件事不是直接跑前端，而是把后端接口可能遇到的问题先列成用例，用pytest跑一遍，这样后面改模板、加规则都不会把已有功能弄坏。

import pytest import requests BASE_URL = "http://localhost:5000" def ask(question): resp = requests.get(f"{BASE_URL}/api/qa", params={"question": question}) return resp.json() def test_actor_movies(): result = ask("演员A主演的电影有哪些") assert result["code"] == 200 assert len(result["data"]) > 0 assert all("title" in item for item in result["data"]) def test_unknown_template(): result = ask("今天吃什么") assert result["data"] == [] # 或者返回到提示信息 def test_empty_question(): resp = requests.get(f"{BASE_URL}/api/qa", params={"question": ""}) assert resp.json()["code"] == 400

这份测试用例的价值是让模板匹配的改动变得可验证。我测试时会故意写几个近似问法，比如“演员A的电影”和“演员A主演的电影有哪些”，看返回差异。如果某条规则把不该匹配的问题吞掉了，断言会立刻失败，这样就能快速发现正则优先级问题。

在此之前，我拿到类似源码总是先跑通再说，后来在某个问答系统上翻了一次车：改了一个正则，结果所有“导演”问题误判成“主演”，接口全错。从那以后，我每次动模板或数据导入逻辑，都会强制走一遍 pytest 回归用例，再启动小程序联调。希望这套系统的拆解和排坑记录能帮到你少走一段弯路。

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