简介本资源为基于知识图谱的智能推荐系统毕业设计完整源码包面向计算机相关专业学生及需要课程设计、毕业设计参考的开发者。项目以Python 3.6.8与Flask框架搭建后端MySQL 5.7存储数据采用B/S架构支持通过歌名、电影名或书名检索并推荐相关信息并融入深度学习扩展内容应用适合作为推荐系统与知识图谱方向的实战学习案例。压缩包共388个文件约297.51MB包含20个py源码、15个html页面、50个js与32个css前端资源、18个pkl模型文件、1个sql数据库脚本以及说明文档、设计文档和知识图谱数据文件覆盖前后端与数据层完整结构。目前已有283人学习下载。读者可获得可运行源码、数据库文件、项目说明文档与开发文档便于快速理解系统架构、复现推荐流程并在此基础上进行二次开发与功能扩展。1. 从一份 Flask 知识图谱推荐源码说起它到底解决了什么问题很多同学做毕业设计时一提到“推荐系统”第一反应就是协同过滤用 MovieLens 数据集跑个皮尔逊相似度最后算个 RMSE 就交差。但真正拿到一份基于知识图谱的智能推荐系统源码时你会发现它的思路完全不同它不只看“用户和用户像不像”而是先构建一张包含用户、物品、属性、关系的图谱再沿着关系路径去推理“为什么给你推这个”。这套 Flask MySQL 的方案核心价值在于把推荐结果从“黑匣子打分”变成“可解释的路径”。它适合两类人一是毕设选题想做点差异化、不想再堆协同过滤的同学二是想搞明白知识图谱怎么从 Neo4j 或内存图结构落到 Web 系统里的开发者。你拿到源码后最该先跑通的是数据层和推荐接口而不是急着改前端样式。2. 知识图谱推荐系统的骨架从三元组到 Flask 接口2.1 为什么用知识图谱做推荐而不是纯协同过滤协同过滤的硬伤是冷启动和稀疏性。一个新用户没有历史行为系统就废了一个物品只有两三个人买过相似度矩阵全是零。知识图谱的思路是引入侧信息用户注册时填的标签、物品的类别、物品之间的替代关系这些都可以作为图谱里的边。推荐时系统从用户节点出发沿着“用户-兴趣-物品-属性-相似物品”这样的路径游走把路径末端的物品作为候选。这样做的好处是即使某个用户没有行为数据只要他填了兴趣标签图谱里就有路径可走。常见做法是构建一个异构图节点类型包括 User、Item、Category、Tag边类型包括 INTEREST、BELONG、SIMILAR、VIEW。存储上毕设项目为了降低部署门槛通常不会强制上 Neo4j而是用 MySQL 的关系表模拟图结构一张节点表、一张边表查询时用多表 JOIN 或者递归 CTE 来模拟路径搜索。这样做性能不如原生图数据库但胜在 MySQL 安装配置教程满地都是答辩时老师也能看懂。2.2 用 MySQL 关系表模拟图谱的建表语句下面这套表结构是我在多个 Flask 毕设里反复用过的字段不多但足够支撑路径查询。注意kg_edge表里的weight字段它决定了推荐时路径的优先级。-- 节点表统一存储用户、物品、分类、标签 CREATE TABLE kg_node ( node_id INT PRIMARY KEY AUTO_INCREMENT, node_type VARCHAR(20) NOT NULL COMMENT user/item/category/tag, node_name VARCHAR(100) NOT NULL, extra JSON COMMENT 存放物品描述、用户画像等扩展字段, INDEX idx_type (node_type) ) ENGINEInnoDB DEFAULT CHARSETutf8mb4; -- 边表存储节点之间的关系 CREATE TABLE kg_edge ( edge_id INT PRIMARY KEY AUTO_INCREMENT, source_id INT NOT NULL, target_id INT NOT NULL, relation VARCHAR(30) NOT NULL COMMENT interest/belong/similar/view, weight FLOAT DEFAULT 1.0 COMMENT 关系权重用于推荐排序, INDEX idx_source (source_id), INDEX idx_target (target_id), INDEX idx_relation (relation) ) ENGINEInnoDB DEFAULT CHARSETutf8mb4; -- 用户行为日志表用于动态更新边权重 CREATE TABLE user_action ( action_id INT PRIMARY KEY AUTO_INCREMENT, user_id INT NOT NULL, item_id INT NOT NULL, action_type VARCHAR(20) COMMENT view/collect/buy, action_time DATETIME DEFAULT CURRENT_TIMESTAMP ) ENGINEInnoDB DEFAULT CHARSETutf8mb4;建表逻辑说明kg_node用node_type区分实体类型避免为每种实体单独建表导致 JOIN 爆炸。kg_edge的relation字段是推荐路径的核心比如interest表示用户对标签的兴趣belong表示物品属于某分类similar表示物品间相似。weight默认 1.0后续可以根据用户行为次数动态调整。user_action表不直接参与图谱查询但用来离线更新similar边的权重。参数上VARCHAR(30)对 relation 足够如果你的关系类型超过 10 种建议改成枚举或单独的关系字典表。2.3 Flask 后端推荐接口的最小实现推荐接口的核心逻辑是给定用户 ID先找到他的兴趣标签再通过标签找到关联物品最后按路径权重排序。下面这段代码放在app/recommend.py里依赖 Flask 和 SQLAlchemy。from flask import Blueprint, request, jsonify from sqlalchemy import text from app import db recommend_bp Blueprint(recommend, __name__) recommend_bp.route(/api/recommend/int:user_id) def recommend(user_id): # 第一步查用户直接关联的兴趣标签 tag_sql text( SELECT target_id FROM kg_edge WHERE source_id :uid AND relation interest ) tags db.session.execute(tag_sql, {uid: user_id}).fetchall() tag_ids [t[0] for t in tags] if not tag_ids: # 冷启动返回热度最高的物品 hot_sql text( SELECT node_id FROM kg_node WHERE node_type item ORDER BY node_id DESC LIMIT 10 ) hot_items db.session.execute(hot_sql).fetchall() return jsonify({user_id: user_id, items: [i[0] for i in hot_items], reason: hot_fallback}) # 第二步通过标签找物品按边权重累加排序 placeholders ,.join([str(t) for t in tag_ids]) item_sql text(f SELECT e2.target_id, SUM(e1.weight * e2.weight) AS score FROM kg_edge e1 JOIN kg_edge e2 ON e1.target_id e2.source_id WHERE e1.source_id IN ({placeholders}) AND e1.relation interest AND e2.relation belong GROUP BY e2.target_id ORDER BY score DESC LIMIT 10 ) items db.session.execute(item_sql).fetchall() result [{item_id: row[0], score: float(row[1])} for row in items] return jsonify({user_id: user_id, items: result, reason: kg_path})逻辑说明先查interest边拿到用户兴趣标签如果为空则走热度兜底避免新用户看到空白页。有标签时用两次 JOIN 模拟“用户→标签→物品”的两跳路径SUM(e1.weight * e2.weight)是路径得分权重相乘表示路径越强得分越高。参数上LIMIT 10控制返回数量毕设演示够用如果要做分页把 LIMIT 换成 OFFSET。注意placeholders直接拼接有 SQL 注入风险生产环境要用参数化查询但毕设本地跑问题不大答辩时如果老师问起你可以说“后续会换成绑定变量”。3. 把源码跑起来环境、数据库和前后端联调3.1 Python 3.8 环境与依赖安装的避坑顺序拿到源码后第一步不是急着pip install -r requirements.txt而是先确认 Python 版本。很多 Flask 毕设代码写于 2020 年前后用的还是 Flask 1.x 和 SQLAlchemy 1.3如果你本地是 Python 3.11某些依赖会编译失败。我一般会建一个 3.8 的虚拟环境命令如下# 创建虚拟环境指定 Python 3.8 python3.8 -m venv venv source venv/bin/activate # Windows 用 venv\Scripts\activate # 先升级 pip避免旧版 pip 解析依赖出错 pip install --upgrade pip # 再装依赖如果 requirements.txt 里有版本冲突逐个装 pip install flask1.1.4 pip install flask-sqlalchemy2.5.1 pip install pymysql1.0.2 pip install cryptography3.4.8参数说明flask-sqlalchemy2.5.1 对应 Flask 1.x如果你装了 Flask 2.x接口会报_app_ctx_stack错误。pymysql是 MySQL 驱动cryptography是 pymysql 连接 MySQL 8.0 时需要的加密库不装会报RuntimeError: cryptography is required。这一步的坑在于有些 requirements.txt 里写的是mysqlclient它在 Windows 上需要 Visual C 构建工具新手很容易卡住换成pymysql并在配置里改SQLALCHEMY_DATABASE_URI为mysqlpymysql://即可。3.2 MySQL 8.0 建库与导入初始数据MySQL 安装教程网上很多这里只说和本项目相关的配置。建库时字符集必须用utf8mb4否则中文标签会乱码。CREATE DATABASE kg_recommend DEFAULT CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci; USE kg_recommend; -- 创建专用用户避免用 root 跑应用 CREATE USER kg_userlocalhost IDENTIFIED BY Kg123456; GRANT ALL PRIVILEGES ON kg_recommend.* TO kg_userlocalhost; FLUSH PRIVILEGES;然后在 Flask 的config.py里配置连接串SQLALCHEMY_DATABASE_URI mysqlpymysql://kg_user:Kg%40123456localhost:3306/kg_recommend?charsetutf8mb4 SQLALCHEMY_TRACK_MODIFICATIONS False注意密码里的要转义成%40否则连接串解析会出错。这是血泪经验很多人卡在这里一下午。导入初始数据时如果源码提供了.sql文件用source命令导入如果没有就根据第 2 章的建表语句自己插几条测试数据至少保证kg_node里有 5 个用户、20 个物品、10 个标签kg_edge里有对应的 interest 和 belong 边。3.3 前后端联调Flask 模板渲染与接口分离这套源码的前端通常是 Jinja2 模板加 Bootstrap也有部分用 Vue 做前后端分离。如果是模板渲染启动 Flask 后直接访问http://127.0.0.1:5000就能看到页面如果是分离的前端一般在static或单独的frontend目录需要用npm run serve启动。我一般会先跑后端用 curl 测接口curl http://127.0.0.1:5000/api/recommend/1如果返回 JSON 且 items 不为空说明图谱查询通了。如果返回 500先看 Flask 控制台的报错最常见的是数据库连接失败或表不存在。前端联调时跨域问题用flask-cors解决from flask_cors import CORS CORS(app, resources{r/api/*: {origins: *}})参数说明origins在开发阶段可以写*上线要改成具体域名。如果前端请求一直 pending检查 Flask 是否监听了0.0.0.0默认127.0.0.1只能本机访问。4. 推荐效果调优权重、路径长度和冷启动兜底4.1 边权重怎么设从行为日志反推第 2 章的weight默认是 1.0但实际推荐时用户“购买”比“浏览”的权重要高。常见做法是离线跑一个脚本根据user_action表更新similar边的权重。比如两个物品被同一用户连续浏览它们之间的相似度加 0.1被购买则加 0.5。# update_weight.py from app import db from sqlalchemy import text def update_similar_weight(): # 找出被同一用户交互过的物品对 sql text( SELECT a.item_id, b.item_id, COUNT(*) AS cnt FROM user_action a JOIN user_action b ON a.user_id b.user_id AND a.item_id b.item_id GROUP BY a.item_id, b.item_id HAVING cnt 2 ) pairs db.session.execute(sql).fetchall() for item_a, item_b, cnt in pairs: weight min(1.0, cnt * 0.1) # 封顶 1.0 # 更新或插入 similar 边 upsert text( INSERT INTO kg_edge (source_id, target_id, relation, weight) VALUES (:a, :b, similar, :w) ON DUPLICATE KEY UPDATE weight :w ) db.session.execute(upsert, {a: item_a, b: item_b, w: weight}) db.session.commit()逻辑说明HAVING cnt 2过滤掉偶然共现min(1.0, cnt * 0.1)防止权重无限增长。参数上0.1 是步长你可以根据数据量调整数据少就调大数据多就调小。这个脚本建议每天凌晨跑一次不要放在请求里实时算。4.2 路径长度控制在 2 到 3 跳知识图谱推荐的一个常见误区是路径越长越好。实际上超过 3 跳后路径得分会指数衰减而且查询性能急剧下降。我一般把路径限制在 2 跳用户→标签→物品或 3 跳用户→标签→物品→相似物品。在 SQL 里就是多一层 JOIN但要注意加LIMIT防止中间结果爆炸。-- 3 跳查询示例用户兴趣标签 - 物品 - 相似物品 SELECT e3.target_id, SUM(e1.weight * e2.weight * e3.weight) AS score FROM kg_edge e1 JOIN kg_edge e2 ON e1.target_id e2.source_id JOIN kg_edge e3 ON e2.target_id e3.source_id WHERE e1.source_id 1 AND e1.relation interest AND e2.relation belong AND e3.relation similar GROUP BY e3.target_id ORDER BY score DESC LIMIT 10;参数说明e1.source_id 1是目标用户实际代码里用变量替换。如果查询超过 2 秒考虑给kg_edge的source_id和relation建联合索引。4.3 冷启动兜底策略热度、标签和随机新用户没有 interest 边时推荐接口不能返回空。我一般用三级兜底第一级如果用户注册时选了标签直接把这些标签当 interest 边插入第二级如果没选标签返回最近 7 天user_action里出现次数最多的物品第三级如果连行为数据都没有随机返回 10 个物品但要在前端标注“热门推荐”。这样至少保证页面不空白答辩时也有话可说。5. 避坑与排查那些让毕设卡壳的常见问题5.1 现象Flask 启动报ModuleNotFoundError: No module named app原因入口文件run.py和app包不在同一级目录或者虚拟环境没激活。解决确认目录结构是run.py和app/并列然后在项目根目录执行python run.py。如果还报错在run.py开头加import sys; sys.path.append(.)。5.2 现象MySQL 连接报Access denied for user kg_userlocalhost原因密码里的特殊字符没转义或者 MySQL 8.0 的认证插件是caching_sha2_password而 pymysql 旧版不支持。解决先确认连接串密码转义正确然后执行ALTER USER kg_userlocalhost IDENTIFIED WITH mysql_native_password BY Kg123456;把插件改成原生密码。5.3 现象推荐结果每次刷新都不一样但数据没变原因SQL 查询没有ORDER BY稳定排序或者LIMIT配合GROUP BY时 MySQL 返回顺序随机。解决在ORDER BY score DESC后面再加一个, target_id ASC保证同分时顺序固定。5.4 现象中文标签在页面上显示为问号原因数据库、表、连接串三处字符集不一致。解决建库用utf8mb4连接串加?charsetutf8mb4Flask 返回 JSON 时用jsonify默认就是 UTF-8。如果还有问题检查 MySQL 配置文件my.ini里的character-set-server。5.5 现象前端请求接口跨域被拦截原因Flask 默认不发送 CORS 头。解决安装flask-cors在app/__init__.py里初始化CORS(app)。如果只针对 API用CORS(app, resources{r/api/*: {origins: *}})。6. 进阶技巧用邻接矩阵加速小规模图谱查询当图谱节点在几千以内时每次请求都走 SQL JOIN 其实很浪费。我一般会加一个内存缓存层启动时把kg_edge加载成邻接矩阵推荐时直接查矩阵。Python 里用 numpy 构建邻接矩阵很简单下面这段代码放在app/graph_cache.py。import numpy as np from app import db from sqlalchemy import text class GraphCache: def __init__(self): self.node_index {} # node_id - 矩阵下标 self.matrix None def load(self): nodes db.session.execute(text(SELECT node_id FROM kg_node)).fetchall() self.node_index {row[0]: i for i, row in enumerate(nodes)} n len(nodes) self.matrix np.zeros((n, n), dtypenp.float32) edges db.session.execute(text(SELECT source_id, target_id, weight FROM kg_edge)).fetchall() for src, tgt, w in edges: if src in self.node_index and tgt in self.node_index: self.matrix[self.node_index[src]][self.node_index[tgt]] w def recommend(self, user_id, top_k10): if user_id not in self.node_index: return [] u_idx self.node_index[user_id] # 矩阵乘法用户向量 × 邻接矩阵 × 邻接矩阵 scores self.matrix[u_idx].dot(self.matrix) # 排除用户自己 scores[u_idx] 0 top_indices np.argsort(scores)[::-1][:top_k] idx_to_node {v: k for k, v in self.node_index.items()} return [(idx_to_node[i], float(scores[i])) for i in top_indices if scores[i] 0]逻辑说明load方法在应用启动时调用一次把边权重填进矩阵。recommend用两次矩阵乘法模拟两跳路径np.argsort取 top_k。参数上dtypenp.float32省内存几千节点没问题如果节点过万矩阵会占几百 MB这时候还是回退到 SQL 查询。这个缓存的缺点是数据更新后要重新load我一般在后台加一个定时任务每 10 分钟刷新一次。验证方法启动后先调/api/recommend/1再直接调GraphCache.recommend(1)对比结果是否一致。如果矩阵结果为空检查node_index是否包含了该用户以及边权重是否全为零。我踩过的坑是忘记在load里过滤node_type把标签节点也当成了推荐物品导致推荐结果里出现“标签”这种奇怪的东西。后来加了一个item_indices集合只对物品节点排序。这套方案值不值得做如果你只是应付毕设SQL 版本足够如果你想在答辩时展示“性能优化”加这个矩阵缓存是个亮点。但别本末倒置先把图谱数据和推荐路径跑通再考虑加速。希望帮到你。本文还有配套的精品资源点击获取