简介这是一套面向高校计算机及相关专业学生如人工智能、自动化、电子信息等的毕业设计级知识图谱实践项目聚焦电影领域问答场景解决结构化语义查询与自然语言理解落地问题。资源共45个文件涵盖7个核心Python脚本含question_classification.py、process_question.py、data2neo4j.py等、5个CSV电影数据集movie.csv、genre.csv、person_to_movie.csv等、5张系统流程与界面截图png、3个GIF动图演示交互效果以及设计文档.doc、项目说明.md、前端静态资源html/css/js和依赖清单requirements.txt压缩包仅1.19MB轻量易部署。已有71人学习下载适合毕设选题、课程设计或知识图谱入门实战——提供从Neo4j图谱构建、问句分类与模板匹配、前后端联调到完整设计报告的全链路实现代码经严格测试可直接运行小白可循序上手进阶者亦可基于现有模块拓展新实体或问答逻辑。1. 为什么用知识图谱做电影问答比直接喂大模型更稳、更准、更可控你试过让 ChatGLM 或 Qwen 直接回答“《盗梦空间》里柯布的配偶叫什么她是怎么死的”——模型大概率会编造一个名字再配上一段逻辑自洽但完全虚构的死亡情节。这不是模型不强而是它没被明确告知电影世界里的人物关系是确定的、可验证的、有边界的。而知识图谱恰恰就是干这个的把《肖申克的救赎》《阿凡达》《寄生虫》这些片子拆成“导演克里斯托弗·诺兰”“主演莱昂纳多·迪卡普里奥”“类型科幻/悬疑”“上映年份2010”“关联影片《奠基者》同导演”这样的三元组存进 Neo4j 这个原生图数据库里。当用户问“诺兰导演过哪些带梦境设定的电影”系统不是靠语义猜而是走图查询MATCH (d:Director {name:克里斯托弗·诺兰})-[:DIRECTED]-(m:Movie)-[:HAS_THEME]-(:Theme {name:梦境}) RETURN m.title——结果精准、可追溯、零幻觉。这套方案不依赖 GPU 显存堆参数一台 16G 内存的笔记本就能跑通完整 pipeline它也不吃预训练语料的“玄学”所有答案都来自你亲手构建或清洗过的数据集。适合想快速落地垂直领域问答、又不想被大模型黑匣子反噬的 Python 工程师、NLP 初学者、以及需要可解释性输出的课程设计者。2. 从零搭起图谱骨架用 Python 清洗数据 Neo4j 建模 Cypher 定义 Schema2.1 数据集选型与结构解析为什么不用 IMDb API而选现成电影 CSV 包标题里带.zip说明项目附带了本地数据集——这是实操友好性的关键。常见配套数据集有两类一类是movies.csvactors.csvgenres.csv三表结构类似 Kaggle 上的 TMDB 子集另一类是已导出的 Neo4j.dump文件如 MovieLens 图谱快照。我一般会优先选前者因为可控性强字段清晰title, year, director, actors, genres, plot_summary、无 API 调用限制、可离线反复调试。比如某典型movies.csv含 5000 条记录关键字段如下字段名示例值说明movie_idm12345主键后续作为节点 IDtitle盗梦空间中文片名需统一编码UTF-8 BOM 去除year2010整型用于时间范围查询director克里斯托弗·诺兰多人用/分隔需切分后建Person节点actors莱昂纳多·迪卡普里奥/艾伦·佩吉同上注意姓名标准化“艾伦·佩吉” ≠ “埃伦·佩吉”genres科幻/动作/悬疑类型标签建Genre节点并关联plot_summary一名特工通过潜入他人梦境窃取机密…供后续 QA 检索用暂不入图提示别急着用pandas.read_csv()直读。先用fileencoding检测工具如chardet确认编码否则director字段会乱码成å¼ å°”æ–¯æ‰˜å¤«Â·è¯ºå…°——这会导致后续所有节点创建失败。2.2 Python 数据清洗脚本处理分隔符、去重、补全缺失值清洗目标不是追求完美而是让数据能“稳稳落进 Neo4j”。核心矛盾在于CSV 里actors是/分隔字符串但图谱要求每个演员是一个独立Person节点并与电影建立ACTED_IN关系。以下脚本完成三件事① 拆分多值字段② 合并同名不同写法如“周星驰”和“周星馳”③ 为缺失director的电影打上Unknown标签避免 NULL 阻断 Cypher 创建。# clean_movies.py import pandas as pd import re def normalize_name(name: str) - str: 姓名标准化去空格、统一繁体简体、删括号注释 if not isinstance(name, str): return Unknown # 去首尾空格和中间多余空格 name re.sub(r\s, , name.strip()) # 繁体转简体极简版实际项目建议用 opencc name name.replace(馳, 驰).replace(裏, 里).replace(為, 为) # 删除括号及内容如“莱昂纳多·迪卡普里奥Leonardo DiCaprio” name re.sub(r.*?|\(.*?\), , name) return name.strip() df pd.read_csv(movies.csv, encodingutf-8-sig) # 自动处理 BOM # 处理 director 字段 df[director] df[director].fillna(Unknown).apply(normalize_name) # 拆分 actors 并展开成多行关键否则无法建关系 df_actors df.assign(actorsdf[actors].str.split(/)).explode(actors) df_actors[actors] df_actors[actors].apply(normalize_name) # 同理处理 genres df_genres df.assign(genresdf[genres].str.split(/)).explode(genres) df_genres[genres] df_genres[genres].apply(lambda x: x.strip() if isinstance(x, str) else Unknown) # 保存清洗后数据 df.to_csv(movies_clean.csv, indexFalse, encodingutf-8) df_actors.to_csv(movie_actors.csv, indexFalse, encodingutf-8) df_genres.to_csv(movie_genres.csv, indexFalse, encodingutf-8) print(✅ 清洗完成生成 movies_clean.csv / movie_actors.csv / movie_genres.csv)这段代码跑完你会得到三张新 CSVmovies_clean.csv单电影一行、movie_actors.csv一电影-一演员一行、movie_genres.csv一电影-一类型一行。它们就是后续批量导入 Neo4j 的“弹药”。2.3 Neo4j Desktop 配置与 Cypher Schema 定义为什么必须先建约束Neo4j Desktop 是目前最稳的本地开发环境避开了服务器部署的权限坑。安装后新建一个 Project → Database → 选择 5.19 版本兼容最新驱动启动数据库。切记在导入任何数据前先执行约束语句——这是图谱查询性能和数据一致性的基石// 在 Neo4j Browser 中执行注意必须在空库中首次运行 CREATE CONSTRAINT ON (m:Movie) ASSERT m.movie_id IS UNIQUE; CREATE CONSTRAINT ON (p:Person) ASSERT p.name IS UNIQUE; CREATE CONSTRAINT ON (g:Genre) ASSERT g.name IS UNIQUE; CREATE INDEX ON :Movie(title); CREATE INDEX ON :Person(name);这 5 行代码的作用远超“加速查询”ASSERT m.movie_id IS UNIQUE强制每部电影只能有一个Movie节点避免同一部电影因多次导入产生重复节点ASSERT p.name IS UNIQUE让“莱昂纳多·迪卡普里奥”全球唯一后续MERGE时自动去重INDEX ON :Movie(title)是给中文片名建哈希索引否则MATCH (m:Movie {title:阿凡达})会全表扫描——10 万部电影时延迟从 200ms 涨到 3s。注意约束创建后不可删除除非删库且建约束过程会锁库 1~2 秒。生产环境建议在低峰期操作。3. 批量导入用 neo4j-admin import 和 Cypher LOAD CSV 双轨并行3.1 用 neo4j-admin import 导入主实体Movie/Person/Genre快、稳、不占内存neo4j-admin import是 Neo4j 官方推荐的离线批量导入工具速度是 Cypher 的 10 倍以上且不消耗运行时内存。但它要求 CSV 格式严格无 header、无引号、字段用逗号分隔、NULL 用\N表示。所以得先用 Python 把清洗后的 CSV 转成合规格式# generate_import_files.py import csv # 生成 Movie 节点文件movie_id,title,year with open(movies_clean.csv, encodingutf-8) as f: reader csv.DictReader(f) with open(import/movies.csv, w, newline, encodingutf-8) as out: writer csv.writer(out) for row in reader: # 注意year 可能为空转成 \N year row[year] if row[year].strip() else r\N writer.writerow([row[movie_id], row[title], year]) # 生成 Person 节点文件name persons set() with open(movie_actors.csv, encodingutf-8) as f: reader csv.DictReader(f) for row in reader: if row[actors] ! Unknown: persons.add(row[actors]) with open(import/persons.csv, w, newline, encodingutf-8) as out: writer csv.writer(out) for p in sorted(persons): writer.writerow([p])生成import/movies.csv和import/persons.csv后在终端执行路径按你的 Neo4j 安装位置调整# Windows 示例管理员权限运行 cmd cd C:\Program Files\Neo4j Desktop\Neo4jDatabases\database-xxx\installation-5.19.0\bin neo4j-admin import --nodesimport/movies.csv --nodesimport/persons.csv --nodesimport/genres.csv --relationshipsimport/acted_in.csv --ignore-missing-nodestrue --skip-bad-relationshipstrue逻辑说明--ignore-missing-nodestrue允许关系文件中引用的节点不存在时跳过防演员名拼错--skip-bad-relationshipstrue忽略格式错误的关系行。这两个参数是容错关键否则一条脏数据会让整个导入失败。3.2 用 LOAD CSV 补充关系ACTED_IN / HAS_GENRE灵活、可调试、支持条件过滤neo4j-admin import不能处理动态逻辑如“只导入票房 1 亿的电影”这时用 Cypher 的LOAD CSV更合适。它在数据库运行时执行支持WHERE过滤、FOREACH循环、CASE判断// 导入演员关系从 movie_actors.csv LOAD CSV WITH HEADERS FROM file:///movie_actors.csv AS row MERGE (m:Movie {movie_id: row.movie_id}) MERGE (p:Person {name: row.actors}) CREATE (p)-[:ACTED_IN]-(m); // 导入类型关系注意 genres 字段可能为空 LOAD CSV WITH HEADERS FROM file:///movie_genres.csv AS row WITH row WHERE row.genres Unknown AND row.genres MERGE (m:Movie {movie_id: row.movie_id}) MERGE (g:Genre {name: row.genres}) CREATE (m)-[:HAS_GENRE]-(g);执行前确认 CSV 已放在 Neo4j 的import目录下Desktop 版默认路径~/Documents/Neo4j Desktop/Application/neo4jDatabases/database-xxx/installation-5.19.0/import/且文件权限允许读取。LOAD CSV每次执行约 1000 行/秒5000 部电影的关系导入 5~10 秒搞定。3.3 验证图谱连通性三个必查 Cypher 查询导入完成后别急着写问答逻辑先用这三条语句验血// 1. 查总节点数应接近 movies.csv 行数 actors 去重数 genres 去重数 MATCH (n) RETURN count(n) AS node_count; // 2. 查一部电影的完整关系链验证是否漏边 MATCH (m:Movie {title: 盗梦空间})-[]-(related) RETURN m.title AS movie, labels(related) AS type, properties(related) AS props LIMIT 5; // 3. 查是否存在“孤岛节点”没任何关系的 Person MATCH (p:Person) WHERE NOT (p)--() RETURN p.name AS orphan LIMIT 10;如果第 3 条返回结果说明某些演员只出现在 CSV 里但没关联到电影——要回溯movie_actors.csv看movie_id是否匹配movies_clean.csv的主键。这是新手最常踩的“数据对不上”坑。4. 问答引擎实现Python Neo4j Driver 构建意图识别 Cypher 生成器4.1 问题分类器用规则 小模型区分“查人物”“查关系”“查列表”大模型做 QA 是重武器但本项目定位是“轻量可控”所以用规则引擎打底辅以关键词匹配。我们定义四类问题模板问题类型用户输入示例匹配规则对应 Cypher 模板单实体属性查询“《泰坦尼克号》的导演是谁”含“《》”“导演”MATCH (m:Movie {title:$title}) RETURN m.director关系查询“诺兰和莱昂纳多合作过哪些电影”含人名“合作”/“一起”MATCH (p1:Person {name:$p1})-[:ACTED_IN]-(m)-[:DIRECTED]-(p2:Person {name:$p2}) RETURN m.title列表查询“周星驰演过的喜剧片有哪些”含人名类型词喜剧/爱情/科幻MATCH (p:Person {name:$name})-[:ACTED_IN]-(m)-[:HAS_GENRE]-(g:Genre {name:$genre}) RETURN m.title, m.year模糊检索“讲梦境的电影”含主题词梦境/时间/平行宇宙MATCH (m:Movie) WHERE m.plot_summary CONTAINS $keyword RETURN m.title, m.year LIMIT 5Python 实现用re和jieba中文分词做基础解析# qa_engine.py import re import jieba class MovieQAEngine: def __init__(self, driver): self.driver driver def parse_intent(self, question: str): # 规则1匹配《片名》 属性词 movie_match re.search(r《(.*?)》, question) if movie_match: title movie_match.group(1) if 导演 in question: return director_query, {title: title} if 主演 in question or 演员 in question: return cast_query, {title: title} # 规则2匹配人名 合作词 person_names [w for w in jieba.lcut(question) if len(w) 2 and w not in [和, 与, 及]] if len(person_names) 2 and any(kw in question for kw in [合作, 一起, 搭档]): return collab_query, {p1: person_names[0], p2: person_names[1]} # 规则3匹配人名 类型词 genre_words [喜剧, 爱情, 科幻, 动作, 悬疑, 动画] for genre in genre_words: if genre in question and any(p in question for p in person_names): return genre_query, {name: person_names[0], genre: genre} # 默认模糊检索 return fuzzy_query, {keyword: question.replace(讲, ).replace(的电影, )} def execute_query(self, intent, params): with self.driver.session() as session: if intent director_query: result session.run( MATCH (m:Movie {title:$title}) RETURN m.director AS director, titleparams[title] ) elif intent collab_query: result session.run( MATCH (p1:Person {name:$p1})-[:ACTED_IN]-(m)-[:DIRECTED]-(p2:Person {name:$p2}) RETURN m.title AS title, m.year AS year, p1params[p1], p2params[p2] ) # ... 其他 intent 分支 return [record.data() for record in result] # 使用示例 from neo4j import GraphDatabase driver GraphDatabase.driver(bolt://localhost:7687, auth(neo4j, password)) qa MovieQAEngine(driver) print(qa.execute_query(*qa.parse_intent(《阿凡达》的导演是谁)))参数说明GraphDatabase.driver的auth参数必须与 Neo4j Desktop 中设置的密码一致默认是neo4j首次登录后需改密bolt://localhost:7687是 Desktop 的默认连接地址若改过端口需同步更新。4.2 Cypher 生成器为什么不用 f-string 拼接而用参数化查询新手常犯错误用f MATCH (m:Movie {{title:{title}}}) 拼接 Cypher——这会导致 SQL 注入式漏洞用户输入《泰坦尼克号 OR 11》就能拖库。正确做法是所有变量都用$param占位符由 driver 自动转义# ❌ 危险 query fMATCH (m:Movie {{title:{user_input}}}) RETURN m # ✅ 安全 query MATCH (m:Movie {title:$title}) RETURN m result session.run(query, titleuser_input) # driver 自动处理引号转义Neo4j Driver 对$title的处理等价于数据库的PreparedStatement无论user_input是《肖申克的救赎》还是abc; DROP DATABASE movies; --都会被当作纯字符串值传入绝不会执行恶意语句。4.3 问答接口封装Flask Web 服务 前端简易 HTML为了让非程序员也能测试加一层 Web 接口# app.py from flask import Flask, request, jsonify, render_template from qa_engine import MovieQAEngine from neo4j import GraphDatabase app Flask(__name__) driver GraphDatabase.driver(bolt://localhost:7687, auth(neo4j, your_password)) qa_engine MovieQAEngine(driver) app.route(/) def home(): return render_template(index.html) # 静态 HTML 页面 app.route(/ask, methods[POST]) def ask(): question request.json.get(question, ).strip() if not question: return jsonify({error: 问题不能为空}), 400 try: intent, params qa_engine.parse_intent(question) result qa_engine.execute_query(intent, params) return jsonify({answer: result, intent: intent}) except Exception as e: return jsonify({error: f执行失败{str(e)}}), 500 if __name__ __main__: app.run(debugTrue, host0.0.0.0, port5000)配套templates/index.html精简版!DOCTYPE html html headtitle电影知识图谱问答/title/head body h2 电影知识图谱问答系统/h2 input idq placeholder例如诺兰和莱昂纳多合作过哪些电影 stylewidth:500px button onclickask()提问/button div idans/div script function ask() { const q document.getElementById(q).value; fetch(/ask, { method: POST, headers: {Content-Type: application/json}, body: JSON.stringify({question: q}) }) .then(r r.json()) .then(data { if (data.error) throw data.error; const ans data.answer.map(a Object.entries(a).map(([k,v]) ${k}${v}).join( | ) ).join(br); document.getElementById(ans).innerHTML b答案/b${ans}; }) .catch(err document.getElementById(ans).innerHTML b❌ 错误/b${err}); } /script /body /html运行python app.py访问http://localhost:5000即可交互测试。这是验证整个 pipeline 是否打通的最后关卡。5. 避坑指南Neo4j Python 知识图谱问答的 4 个血泪经验5.1 现象neo4j.exceptions.ServiceUnavailable: Cannot connect to ...原因Neo4j Desktop 数据库未启动或防火墙阻止了 7687 端口或密码错误Desktop 默认密码是neo4j但首次登录后强制修改很多人忘了新密码。解决① 在 Neo4j Desktop 界面点击数据库旁的 ▶️ 启动按钮② 检查右下角状态栏是否显示Running③ 在 Browser 中执行:play movie-graph验证连接④ 若密码遗忘在 Desktop 设置中重置密码Settings → Change Password。5.2 现象Cypher execution error: Node with label Movie must have the property movie_id原因CREATE CONSTRAINT ON (m:Movie) ASSERT m.movie_id IS UNIQUE已建但导入 CSV 时movie_id字段为空或重复如两行movie_id都是m001。解决① 用pandas检查movies_clean.csv中movie_id是否全非空且唯一df[movie_id].isnull().sum()和df[movie_id].duplicated().sum()② 若有重复用df.drop_duplicates(subset[movie_id], keepfirst)去重③ 重新运行neo4j-admin import。5.3 现象问答返回空结果但 Cypher 在 Browser 中能查到原因Python 代码中session.run()未消费结果流忘记list(result)或for record in result导致 driver 认为查询未完成后续请求被阻塞。解决所有session.run()后必须显式获取结果哪怕只是list(result)。正确写法# ✅ 正确 result session.run(MATCH (m:Movie) RETURN m.title LIMIT 10) records list(result) # 必须这行 return [r[m.title] for r in records] # ❌ 错误看似正常实则 driver 连接泄漏 result session.run(MATCH (m:Movie) RETURN m.title LIMIT 10) return [r[m.title] for r in result] # 迭代器未完全消费5.4 现象中文搜索m.plot_summary CONTAINS 梦境返回空但 Browser 中手动查有结果原因Neo4j 默认的CONTAINS是大小写敏感且不支持中文分词梦境在plot_summary中实际存储为梦境设定但CONTAINS不做子串匹配它匹配的是 token而中文无空格分词。解决① 改用正则WHERE m.plot_summary ~ .*梦境.*② 或启用全文索引推荐// 创建全文索引一次 CALL db.index.fulltext.createNodeIndex(plotIndex, [Movie], [plot_summary]) // 查询时用 CALL db.index.fulltext.queryNodes(plotIndex, 梦境) YIELD node, score RETURN node.title, score全文索引对中文支持更好且score可排序相关性。6. 进阶技巧用 RAG 增强问答边界 图谱版本管理实战6.1 当图谱查不到时用 RAG 补位把 plot_summary 当向量库用知识图谱擅长结构化查询谁导演了什么、A 和 B 有什么关系但对开放性问题“《寄生虫》表达了什么社会隐喻”力不从心。这时不必扔掉图谱而是用 RAGRetrieval-Augmented Generation补位把plot_summary文本向量化存入 ChromaDB用户问开放问题时先查图谱查不到就查向量库再把 top-3 摘要喂给 LLM 总结。# rag_fallback.py from chromadb import Client from sentence_transformers import SentenceTransformer # 初始化向量库只需一次 client Client() collection client.create_collection(movie_plots) model SentenceTransformer(paraphrase-multilingual-MiniLM-L12-v2) # 批量插入 plot_summary 向量从 movies_clean.csv 读 df pd.read_csv(movies_clean.csv) texts df[plot_summary].fillna().tolist() embeddings model.encode(texts, show_progress_barTrue) ids [fm_{i} for i in range(len(texts))] collection.add( idsids, embeddingsembeddings.tolist(), documentstexts, metadatas[{title: t} for t in df[title].tolist()] ) # 问答时 fallback def rag_search(query: str, top_k3): query_emb model.encode([query])[0] results collection.query( query_embeddings[query_emb.tolist()], n_resultstop_k ) return [ {title: r[title], summary: r[document]} for r in results[metadatas][0] ]这样当图谱引擎返回空时调用rag_search(《寄生虫》的社会隐喻)拿到剧情摘要再用ollama run llama3生成解释——图谱管“事实”RAG 管“理解”分工明确。6.2 图谱版本管理用 git dump 文件锁定数据快照多人协作或迭代开发时图谱数据必须可回滚。Neo4j 提供neo4j-admin dump命令生成二进制快照# 导出当前图谱含 schema 和数据 neo4j-admin dump --to/path/to/movies_v1.dump # 恢复会清空当前库 neo4j-admin load --from/path/to/movies_v1.dump --force我习惯把.dump文件和对应的movies_clean.csv一起提交 Git并在 README 写明版本对应关系版本dump 文件CSV 文件更新说明v1.0movies_v1.dumpmovies_clean_v1.csv初始 5000 部电影含导演/演员/类型v2.0movies_v2.dumpmovies_clean_v2.csv新增 200 部豆瓣高分片补全 plot_summary这样同事拉代码后只需neo4j-admin load --frommovies_v2.dump --force30 秒还原完整环境不用再跑一遍清洗和导入。6.3 一个真实踩坑后的习惯每次改 Cypher 前先EXPLAIN看执行计划Neo4j 的EXPLAIN命令能显示查询如何走索引、是否触发全表扫描。比如这条慢查询// ❌ 没用索引全表扫描 MATCH (m:Movie) WHERE m.title STARTS WITH 阿 RETURN m.title执行EXPLAIN MATCH (m:Movie) WHERE m.title STARTS WITH 阿 RETURN m.title看到AllNodesScan——说明没走索引。修复方法是建前缀索引// ✅ 创建前缀索引Neo4j 5.12 CREATE TEXT INDEX movie_title_prefix ON :Movie(title) OPTIONS {indexProvider: lucenenative-2.0};现在STARTS WITH就能毫秒级响应。我现在的开发流程固定为写完 Cypher →EXPLAIN→ 看是否有NodeByLabelScan或DirectedRelationshipTypeScan健康信号→ 没有就加索引或改写。希望帮到你。本文还有配套的精品资源点击获取