Python知识图谱智能推荐系统实战：从Neo4j图谱构建到推荐算法实现

简介本资源为毕业设计Python基于知识图谱的智能推荐系统完整项目包面向计算机相关专业学生及需要完成毕设、期末大作业或课程设计的学习者帮助解决推荐系统选题难、代码无从下手、文档缺失等问题。压缩包共168个文件约200.95MB包含14个py核心源码、8个html页面、8个css与8个js前端资源以及jpg、png等界面截图和演示素材另附数据库文件、xls数据表与md说明文档代码注释齐全新手也能看懂。项目围绕知识图谱构建与智能推荐算法展开涵盖用户画像、关系抽取、推荐结果展示等模块界面美观、功能完善部署简单即可运行。已有154人学习下载作者为qq_38140936。下载后可获得完整源码、数据库与文档说明便于快速理解系统架构、复现推荐流程并作为高分毕设或课程设计参考具有较高的实际应用与学习价值。1. 从零读懂「Python 知识图谱智能推荐系统」到底在做什么很多同学做毕业设计时一看到「知识图谱 智能推荐」这两个词就头大觉得这是两个独立的大工程硬凑在一起只会让代码越写越乱。其实这套系统的核心逻辑非常清晰用知识图谱把用户、物品、属性、行为之间的复杂关系存下来再基于这些关系做推荐。相比传统的协同过滤只靠「用户-物品评分矩阵」知识图谱能引入物品的类别、标签、属性关联解决冷启动和数据稀疏的问题。这套方案适合有一定 Python 基础、想做出差异化毕设的本科生也适合想快速理解图谱推荐落地路径的开发者。你需要的技术栈并不复杂Python 做后端逻辑Neo4j 存图谱MySQL 存业务数据前端可以用简单的 Web 框架展示结果。接下来我会把整个系统拆成可复现的步骤从环境搭建到图谱构建再到推荐算法实现每一步都给出能直接抄的代码和参数说明。2. 环境搭建与数据层设计Python、Neo4j、MySQL 怎么配2.1 Python 环境与依赖安装的避坑配置先解决 Python 安装和环境配置的问题。很多教程让你直接装最新版但知识图谱相关的库对版本有要求我一般推荐 Python 3.9 或 3.10太新的版本某些图数据库驱动会报兼容错误。安装完 Python 后用虚拟环境隔离依赖这是血泪经验——不隔离的话后面装 py2neo 和 pandas 很容易冲突。# 创建虚拟环境Windows 和 Linux 通用 python -m venv kg_rec_env # 激活环境 # Windows: kg_rec_env\Scripts\activate # Linux/Mac: source kg_rec_env/bin/activate # 安装核心依赖指定版本避免兼容问题 pip install py2neo2021.2.3 pip install pandas1.5.3 pip install numpy1.24.2 pip install scikit-learn1.2.2 pip install flask2.2.3 pip install pymysql1.0.3这段命令做了三件事建虚拟环境、激活、装指定版本的库。py2neo 是 Python 操作 Neo4j 最顺手的驱动2021.2.3 这个版本对 Neo4j 4.x 支持稳定pandas 和 numpy 处理数据scikit-learn 用来做相似度计算flask 提供 Web 接口pymysql 连 MySQL。参数上注意如果你用的 Neo4j 是 5.xpy2neo 要换到 2021.2.4 以上否则连接会超时。提示装完以后用pip list检查一遍确认没有红色报错。如果 py2neo 安装失败多半是缺少 C 编译环境Windows 下装个 Visual Studio Build Tools 即可。2.2 Neo4j 图数据库的安装与知识图谱 Schema 设计Neo4j 是知识图谱的存储核心社区版免费够用。下载 Neo4j Desktop 或者直接解压社区版压缩包启动后默认端口 7474浏览器和 7687Bolt 协议。第一次登录用户名密码都是 neo4j会强制你改密码记住它。Schema 设计决定了图谱能不能支撑推荐。我一般会建这几类节点和关系节点标签属性说明Useruser_id, name, age用户基本信息Itemitem_id, title, category物品/商品信息Categorycategory_id, name类别节点Tagtag_id, name标签节点关系类型方向说明PURCHASEDUser → Item购买行为VIEWEDUser → Item浏览行为BELONGS_TOItem → Category物品属于类别HAS_TAGItem → Tag物品拥有标签SIMILAR_TOItem → Item物品相似关系这个 Schema 的好处是推荐时可以从 User 出发沿着 PURCHASED 找到物品再通过 BELONGS_TO 和 HAS_TAG 找到同类或同标签的物品形成「用户-物品-属性」的推理路径。比起只存评分矩阵这种结构能解释「为什么推荐这个」。from py2neo import Graph, Node, Relationship # 连接 Neo4j替换成你的密码 graph Graph(bolt://localhost:7687, auth(neo4j, your_password)) # 清空旧数据调试时用生产环境慎用 graph.run(MATCH (n) DETACH DELETE n) # 创建约束加速查询 graph.run(CREATE CONSTRAINT IF NOT EXISTS FOR (u:User) REQUIRE u.user_id IS UNIQUE) graph.run(CREATE CONSTRAINT IF NOT EXISTS FOR (i:Item) REQUIRE i.item_id IS UNIQUE) graph.run(CREATE CONSTRAINT IF NOT EXISTS FOR (c:Category) REQUIRE c.category_id IS UNIQUE) # 插入示例节点 user1 Node(User, user_idU001, name张三, age22) item1 Node(Item, item_idI001, titlePython编程入门, category图书) cat1 Node(Category, category_idC01, name计算机) tag1 Node(Tag, tag_idT01, name编程) graph.create(user1) graph.create(item1) graph.create(cat1) graph.create(tag1) # 建立关系 graph.create(Relationship(user1, PURCHASED, item1)) graph.create(Relationship(item1, BELONGS_TO, cat1)) graph.create(Relationship(item1, HAS_TAG, tag1))这段代码先清库、建唯一约束再插入用户、物品、类别、标签四类节点最后连关系。约束的作用是防止重复插入查询时也能走索引。参数上注意auth里的密码换成你改过的graph.run执行 Cypher 语句graph.create用于创建节点和关系。跑完可以去 Neo4j 浏览器里执行MATCH (n) RETURN n看看图谱长什么样。2.3 MySQL 业务数据库的表结构与增删改查Neo4j 存图谱关系但用户登录、物品详情、操作日志这些结构化数据还是放 MySQL 更合适。建一个库叫kg_recommend然后建三张核心表CREATE DATABASE IF NOT EXISTS kg_recommend DEFAULT CHARSET utf8mb4; USE kg_recommend; -- 用户表 CREATE TABLE users ( user_id VARCHAR(20) PRIMARY KEY, username VARCHAR(50) NOT NULL, password VARCHAR(100) NOT NULL, age INT, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ); -- 物品表 CREATE TABLE items ( item_id VARCHAR(20) PRIMARY KEY, title VARCHAR(200) NOT NULL, category VARCHAR(50), price DECIMAL(10,2), description TEXT ); -- 行为日志表 CREATE TABLE user_behavior ( id INT AUTO_INCREMENT PRIMARY KEY, user_id VARCHAR(20), item_id VARCHAR(20), behavior_type VARCHAR(20), -- view, purchase, rate behavior_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP, FOREIGN KEY (user_id) REFERENCES users(user_id), FOREIGN KEY (item_id) REFERENCES items(item_id) );建完表用 Python 做增删改查pymysql 的用法很直接import pymysql # 连接 MySQL conn pymysql.connect( hostlocalhost, userroot, passwordyour_password, databasekg_recommend, charsetutf8mb4 ) cursor conn.cursor() # 插入用户 cursor.execute(INSERT INTO users (user_id, username, password, age) VALUES (%s, %s, %s, %s), (U001, zhangsan, 123456, 22)) # 查询 cursor.execute(SELECT * FROM users WHERE user_id %s, (U001,)) result cursor.fetchone() print(result) conn.commit() cursor.close() conn.close()参数说明charsetutf8mb4支持中文和 emoji%s是占位符防止 SQL 注入conn.commit()必须调用否则插入不生效。这三张表配合 Neo4j 的图谱就构成了推荐系统的数据底座。3. 知识图谱构建从 CSV 到 Neo4j 的批量导入与实体对齐3.1 用 pandas 清洗数据并生成图谱导入文件真实场景下数据往往来自爬虫或公开数据集格式乱七八糟。我一般先用 pandas 做清洗把用户、物品、行为分别整理成 CSV再批量导入 Neo4j。假设你有一份用户行为日志behavior.csv字段是 user_id、item_id、behavior_type、timestamp。import pandas as pd # 读取原始数据 df pd.read_csv(behavior.csv) # 去重、去空 df.drop_duplicates(subset[user_id, item_id, behavior_type], inplaceTrue) df.dropna(subset[user_id, item_id], inplaceTrue) # 过滤掉行为次数少于 2 的用户冷启动用户单独处理 user_counts df[user_id].value_counts() active_users user_counts[user_counts 2].index df_filtered df[df[user_id].isin(active_users)] # 生成用户节点 CSV users df_filtered[[user_id]].drop_duplicates() users.to_csv(nodes_users.csv, indexFalse, header[user_id:ID(User)] # 生成物品节点 CSV items df_filtered[[item_id]].drop_duplicates() items.to_csv(nodes_items.csv, indexFalse, header[item_id:ID(Item)]) # 生成购买关系 CSV purchases df_filtered[df_filtered[behavior_type] purchase][[user_id, item_id]] purchases.to_csv(rels_purchased.csv, indexFalse, header[:START_ID(User), :END_ID(Item)]) print(f用户数: {len(users)}, 物品数: {len(items)}, 购买关系数: {len(purchases)})这段代码做了四件事去重、去空、过滤低频用户、分别导出节点和关系的 CSV。表头格式user_id:ID(User)是 Neo4j 批量导入工具neo4j-admin import要求的冒号前是字段名括号里是节点标签。参数上注意value_counts()统计频次阈值 2 可以根据数据量调整数据少就设 1。3.2 用 neo4j-admin import 批量导入百万级节点数据量小的时候可以用 py2neo 逐条插但上万条就会慢得让人翻车。Neo4j 自带的neo4j-admin import能秒级导入百万级数据前提是数据库必须是空的。# 先停止 Neo4j 服务 neo4j stop # 清空现有数据库谨慎操作 rm -rf data/databases/neo4j rm -rf data/transactions/neo4j # 执行批量导入 neo4j-admin import \ --databaseneo4j \ --nodesimport/nodes_users.csv \ --nodesimport/nodes_items.csv \ --relationshipsimport/rels_purchased.csv \ --delimiter, \ --skip-bad-relationshipstrue # 启动服务 neo4j start参数说明--databaseneo4j指定目标库--nodes可以跟多个文件--relationships指定关系文件--skip-bad-relationshipstrue跳过格式错误的关系避免整个导入失败。注意 CSV 文件要放在 Neo4j 安装目录的import文件夹下路径不对会报找不到文件。导入完成后去浏览器查MATCH (n) RETURN count(n)验证数量。3.3 实体对齐把 MySQL 里的业务数据和图谱节点关联起来图谱里的节点和 MySQL 里的记录需要对齐否则推荐结果没法展示详情。常见做法是用 user_id 和 item_id 作为桥梁在 Python 里做关联查询。from py2neo import Graph import pymysql graph Graph(bolt://localhost:7687, auth(neo4j, your_password)) conn pymysql.connect(hostlocalhost, userroot, passwordyour_password, databasekg_recommend, charsetutf8mb4) cursor conn.cursor() # 从图谱查出某用户购买过的物品 ID query MATCH (u:User {user_id: $user_id})-[:PURCHASED]-(i:Item) RETURN i.item_id AS item_id result graph.run(query, user_idU001).data() item_ids [row[item_id] for row in result] # 去 MySQL 查详情 if item_ids: placeholders ,.join([%s] * len(item_ids)) cursor.execute(fSELECT item_id, title, price FROM items WHERE item_id IN ({placeholders}), item_ids) for row in cursor.fetchall(): print(row) cursor.close() conn.close()这段代码先用 Cypher 查图谱拿到物品 ID 列表再用这些 ID 去 MySQL 查标题和价格。参数上注意$user_id是 Cypher 的参数化写法防止注入placeholders动态生成占位符适配不定长的 IN 查询。实体对齐做完推荐结果就能带上物品名称和价格用户体验才完整。4. 推荐算法实现基于图谱路径的协同过滤与规则推理4.1 基于 Neo4j 路径查询的相似度计算知识图谱推荐的核心思路是如果两个用户购买过相同或相似的物品那他们可能对其他相同物品也感兴趣。用 Cypher 可以直接查这种路径。from py2neo import Graph graph Graph(bolt://localhost:7687, auth(neo4j, your_password)) def recommend_by_similar_users(user_id, limit10): query MATCH (u1:User {user_id: $user_id})-[:PURCHASED]-(i:Item)-[:PURCHASED]-(u2:User) WHERE u1 u2 MATCH (u2)-[:PURCHASED]-(rec:Item) WHERE NOT (u1)-[:PURCHASED]-(rec) RETURN rec.item_id AS item_id, count(*) AS score ORDER BY score DESC LIMIT $limit result graph.run(query, user_iduser_id, limitlimit).data() return result # 调用 recs recommend_by_similar_users(U001, 5) for r in recs: print(f推荐物品: {r[item_id]}, 得分: {r[score]})这段 Cypher 的逻辑是找到和目标用户买过同一物品的其他用户再看这些用户还买了什么目标用户没买过的按共同购买次数打分排序。参数$limit控制返回数量count(*)就是简单的共现次数。这种做法的优点是解释性强——你能告诉用户「因为你买了 A和你相似的人还买了 B」。缺点是热门物品容易霸榜需要加惩罚因子。4.2 融合物品属性与标签的加权推荐光靠购买关系还不够知识图谱的优势在于能引入物品的类别和标签。我一般会在路径查询里加上属性相似度权重。def recommend_with_tags(user_id, limit10): query MATCH (u:User {user_id: $user_id})-[:PURCHASED]-(i:Item)-[:HAS_TAG]-(t:Tag) MATCH (rec:Item)-[:HAS_TAG]-(t) WHERE NOT (u)-[:PURCHASED]-(rec) AND rec i RETURN rec.item_id AS item_id, count(DISTINCT t) AS tag_score ORDER BY tag_score DESC LIMIT $limit return graph.run(query, user_iduser_id, limitlimit).data() # 结合两种策略购买相似度 标签相似度 def hybrid_recommend(user_id, limit10): sim_recs {r[item_id]: r[score] for r in recommend_by_similar_users(user_id, 20)} tag_recs {r[item_id]: r[tag_score] for r in recommend_with_tags(user_id, 20)} # 加权融合权重可调 all_items set(sim_recs.keys()) | set(tag_recs.keys()) final_scores {} for item in all_items: s1 sim_recs.get(item, 0) s2 tag_recs.get(item, 0) final_scores[item] 0.6 * s1 0.4 * s2 # 权重根据效果调 sorted_items sorted(final_scores.items(), keylambda x: x[1], reverseTrue) return sorted_items[:limit] print(hybrid_recommend(U001, 5))这段代码先分别拿到基于购买相似度和基于标签的推荐列表再用 0.6 和 0.4 的权重融合。参数上注意权重不是固定的如果你的数据里标签信息丰富可以调高标签权重如果购买行为密集就调高相似度权重。融合后的排序更稳定不会因为单一策略的偏差导致推荐质量波动。4.3 冷启动用户的推荐策略利用类别和热门度兜底新用户没有购买记录图谱路径查不出来。这时候要用类别和热门度兜底。def recommend_for_new_user(limit10): query MATCH (i:Item)-[:BELONGS_TO]-(c:Category) OPTIONAL MATCH (i)-[:PURCHASED]-(u:User) RETURN i.item_id AS item_id, c.name AS category, count(u) AS popularity ORDER BY popularity DESC LIMIT $limit return graph.run(query, limitlimit).data() # 或者根据用户注册时选的兴趣类别推荐 def recommend_by_category(category_name, limit10): query MATCH (i:Item)-[:BELONGS_TO]-(c:Category {name: $category}) OPTIONAL MATCH (i)-[:PURCHASED]-(u:User) RETURN i.item_id AS item_id, count(u) AS popularity ORDER BY popularity DESC LIMIT $limit return graph.run(query, categorycategory_name, limitlimit).data()第一个查询按全局热门度推荐第二个按用户选的类别推荐。参数OPTIONAL MATCH保证即使物品没人买过也能返回count(u)统计购买人数作为热度。冷启动策略虽然简单但能保证新用户一进来就有东西可看不至于面对空白页面。5. 避坑与排查知识图谱推荐系统最常见的 5 个翻车现场5.1 Neo4j 连接超时或认证失败现象Python 脚本报py2neo.errors.ConnectionUnavailable或AuthError。原因通常是 Neo4j 服务没启动、端口被占用、密码不对或者 Bolt 协议版本不匹配。解决先确认 Neo4j 服务在运行浏览器打开http://localhost:7474能访问检查auth里的密码是否和改过的一致如果 Neo4j 是 5.x 而 py2neo 是旧版升级 py2neo 到 2021.2.4 以上。另外防火墙可能拦了 7687 端口临时关掉试试。5.2 批量导入时 CSV 格式报错现象neo4j-admin import提示Wrong number of columns或Bad relationships。原因多半是 CSV 表头格式不对或者字段里有逗号没转义。解决严格按字段名:ID(标签)和:START_ID(标签)的格式写表头字段值里如果有逗号用双引号包起来。导入前用head -5看一眼文件内容确认列数一致。--skip-bad-relationshipstrue能跳过错误关系但最好先修数据。5.3 推荐结果全是热门物品现象不管哪个用户推荐列表前几名都是同样的爆款。原因是共现次数没有做归一化热门物品天然得分高。解决在 Cypher 里加惩罚项比如score / log(count(u))或者用 TF-IDF 思路降低热门权重。也可以在 Python 层对得分做归一化再排序。5.4 MySQL 和 Neo4j 数据不一致现象图谱里有某个物品但 MySQL 查不到详情前端展示空白。原因是两边数据没有同步机制。解决写一个定时脚本每天对比两边的 item_id 集合把缺失的补上或删掉。更稳妥的做法是业务写入时同时写 MySQL 和 Neo4j用事务保证一致性但实现复杂度高毕设场景用定时同步就够了。5.5 查询性能随数据量增长急剧下降现象数据量到十万级以后推荐查询从毫秒变成几秒。原因是没有建索引或者 Cypher 写法导致全图扫描。解决给 User.user_id、Item.item_id、Category.name 建唯一约束或索引查询时尽量用MATCH (u:User {user_id: $id})这种带标签和属性的写法避免MATCH (n)全扫描。另外LIMIT要尽早加减少中间结果集。6. 把推荐结果做成可演示的 Web 接口与效果验证毕设最终要答辩光有命令行输出不够得有个能点的页面。我用 Flask 搭一个最简单的接口把推荐结果返回成 JSON前端用 HTML 表格展示。from flask import Flask, request, jsonify from py2neo import Graph import pymysql app Flask(__name__) graph Graph(bolt://localhost:7687, auth(neo4j, your_password)) def get_item_details(item_ids): conn pymysql.connect(hostlocalhost, userroot, passwordyour_password, databasekg_recommend, charsetutf8mb4) cursor conn.cursor() placeholders ,.join([%s] * len(item_ids)) cursor.execute(fSELECT item_id, title, price FROM items WHERE item_id IN ({placeholders}), item_ids) rows cursor.fetchall() cursor.close() conn.close() return [{item_id: r[0], title: r[1], price: float(r[2])} for r in rows] app.route(/recommend) def recommend(): user_id request.args.get(user_id, U001) query MATCH (u1:User {user_id: $user_id})-[:PURCHASED]-(i:Item)-[:PURCHASED]-(u2:User) WHERE u1 u2 MATCH (u2)-[:PURCHASED]-(rec:Item) WHERE NOT (u1)-[:PURCHASED]-(rec) RETURN rec.item_id AS item_id, count(*) AS score ORDER BY score DESC LIMIT 10 result graph.run(query, user_iduser_id).data() item_ids [r[item_id] for r in result] details get_item_details(item_ids) if item_ids else [] return jsonify({user_id: user_id, recommendations: details}) if __name__ __main__: app.run(debugTrue, port5000)这段代码定义了一个/recommend接口接收user_id参数查图谱拿到推荐物品 ID再去 MySQL 补全详情最后返回 JSON。参数上debugTrue方便调试上线要关掉port5000是 Flask 默认端口冲突就换。启动后浏览器访问http://localhost:5000/recommend?user_idU001就能看到推荐结果。验证推荐效果不能只看「有没有返回」得量化。我一般用留一法从每个用户的行为里随机抽一条购买记录藏起来用剩下的数据训练看推荐列表里有没有这条藏起来的物品。命中率就是准确率。import random def evaluate_hit_rate(user_ids, graph, top_n10): hits 0 total 0 for uid in user_ids: # 查该用户所有购买物品 query MATCH (u:User {user_id: $uid})-[:PURCHASED]-(i:Item) RETURN i.item_id AS item_id items [r[item_id] for r in graph.run(query, uiduid).data()] if len(items) 2: continue # 留一法藏一个 test_item random.choice(items) train_items [i for i in items if i ! test_item] # 用训练集做推荐简化版直接查相似用户 rec_query MATCH (u1:User {user_id: $uid})-[:PURCHASED]-(i:Item)-[:PURCHASED]-(u2:User) WHERE u1 u2 MATCH (u2)-[:PURCHASED]-(rec:Item) WHERE NOT (u1)-[:PURCHASED]-(rec) RETURN rec.item_id AS item_id, count(*) AS score ORDER BY score DESC LIMIT $top_n recs [r[item_id] for r in graph.run(rec_query, uiduid, top_ntop_n).data()] total 1 if test_item in recs: hits 1 return hits / total if total 0 else 0 # 假设有 100 个用户 user_ids [fU{i:03d} for i in range(1, 101)] hit_rate evaluate_hit_rate(user_ids, graph, top_n10) print(fHit Rate10: {hit_rate:.2%})这段评估代码用留一法计算 Hit Rate10即藏起来的物品出现在前 10 个推荐里的比例。参数top_n控制推荐列表长度一般取 10 或 20。跑出来如果低于 5%说明图谱关系太稀疏或者权重不合理需要回头调整 Schema 或融合策略。我自己的习惯是每次改完算法都跑一遍这个评估不然改了半天不知道有没有变好纯靠感觉就是玄学。希望这套方案能帮你把毕设稳稳落地。本文还有配套的精品资源点击获取