☰
协同过滤图书推荐系统:Python实现从相似度到个性化书单
2026/10/18 19:56:09 网站建设 项目流程

简介:一套基于Python实现的协同过滤图书推荐系统完整资料包,主要面向推荐系统研究者、图书机构开发者及毕业设计学生。内容涵盖源码、论文与界面截图,从课题背景、协同过滤算法原理到系统设计、测试均有完整呈现,可帮助读者掌握基于物品的协同过滤推荐实现流程,并直接用于图书个性化推荐场景。压缩包共68个文件,大小3.02MB,包含Python源码、SQL数据库脚本、前端js/css/html页面、xml配置、论文docx及多张系统运行截图,覆盖开发、部署与文档各个层面。已有156人学习下载。资源最大价值在于提供了一条从理论到落地的完整链路:论文阐明推荐机制,源码展示核心算法与数据库操作,截图直观还原图书评分、推荐列表、管理页面等真实交互,便于快速理解整体架构并二次开发。

1. 协同过滤图书推荐系统:从用户评分到个性化书单

这套基于 Python 实现的协同过滤图书推荐系统,最让我意外的不是它的准确率,而是它完全不用分析图书内容,只凭用户的历史评分,就能给不同读者算出各自感兴趣的书单。对图书馆、在线书店这类场景,这意味着给每个用户维护个性化推荐不再依赖人工打标签,冷门但优质的书也有机会被翻出来。项目包内同时给了源码、论文和全套界面截图,从算法原理到部署运行是一条线,既可以作为推荐系统学习的练手项目,也能直接当毕业设计的完整骨架。我拆完代码后最大的感受是:它把离线构建相似度矩阵和在线实时推荐两个阶段切得很干净,读通 itemcf.py 一个文件,协同过滤的核心就没有黑匣子了。

2. 核心算法拆解:从用户评分到相似度矩阵的完整计算链路

2.1 算法选型:基于物品的协同过滤为什么比基于用户更合适当下场景

推荐系统里两条主流路线,一条是基于用户的协同过滤(UCF),一条是基于物品的协同过滤(ICF)。这套项目选的是后者,而且选得很有道理。UCF 的思路是“跟你口味相似的人喜欢什么,就推给你什么”,它的问题在于用户量上来之后,用户之间的相似度矩阵变得极其稀疏,计算成本高,而且新用户进来几乎没有行为数据,很难找到“相似的人”。ICF 的思路则是“你喜欢的书,和哪些书像,就把这些像的书推给你”,它计算的是物品与物品之间的相似关系,而图书这种商品相对于用户要稳定得多,《三体》和《球状闪电》的相似关系不太会随用户量增长而剧烈变化。

具体到图书推荐场景,ICF 还有一个隐藏优势是结果可解释性强。给用户推一本书时,系统可以明确告知“因为你给《深入理解计算机系统》打了 5 分,所以推荐《程序员的自我修养》”,这种解释在毕业设计答辩或者实际运营里都非常加分。论文的需求分析部分也确认了这一点,它的功能模块图里推荐模块是核心,而推荐模块依赖的正是物品相似度预计算。

对比维度基于用户的协同过滤基于物品的协同过滤
计算主体用户之间的相似度物品之间的相似度
数据规模影响用户增长导致矩阵急剧膨胀物品数量相对稳定
推荐解释解释起来绕解释直观,容易展示推荐依据
适用场景新闻、短视频等物品快速变化的场景图书、电商等物品相对稳定的场景

2.2 用户-物品矩阵构建:把评分记录变成算法吃得下的数据结构

算法的第一层输入是用户的评分记录。在数据库里,评分表存的是一行行的三元组(user_id, book_id, score),但算法没法直接消费这种平铺结构,需要把它转成“每个用户评过哪些书”的嵌套字典。项目里这一步的产物就是user_item_matrix_str.txt这个文件,说明作者在离线阶段就把矩阵导出成了文本,线上服务启动时直接加载,而不是每次推荐都从数据库现算。

# itemcf.py from collections import defaultdict def build_user_item_matrix(rows): # rows 是数据库查询结果,每条记录形如 (user_id, book_id, score) user_items = defaultdict(dict) for user_id, book_id, score in rows: user_items[user_id][book_id] = float(score) return user_items

这段代码看着简单,但它是整条推荐链路的起点。defaultdict(dict)解决了一个初学容易踩的坑:如果直接初始化普通dict,每遇到一个新用户都要先判断 key 是否存在,否则会抛 KeyError。user_items[user_id][book_id] = float(score)把评分统一转成浮点型,避免后续分数计算时出现字符串拼接这种隐蔽问题。

另一个值得注意的地方是这里刻意保留了用户 id 而不是用户名。算法层面只关心数字 id,用户名展示是 Web 层的事,这个职责拆分在源码里做得比较干净。如果你拿到的数据是用户名,建议在预处理阶段先做一层映射表,别让算法的数据结构里掺入展示字段。

2.3 物品相似度计算:余弦相似度的实现与矩阵导出

有了user_items之后,核心任务是计算任意两本书之间的相似度。经典的余弦相似度公式是:两本书的相似度等于同时给它们评过分的用户数,除以各自被评分数乘积的平方根。这个公式的直觉是:一本书越常和另一本书一起出现在同一批用户的评分列表里,它们就越相似;同时,被所有人评分的热门书会被分母惩罚,避免它跟谁都很相似,这一点在实站里很关键。

# itemcf.py import math def calc_item_similarity(user_items): # 第一步:统计每本书被哪些用户评过 item_users = defaultdict(set) for uid, items in user_items.items(): for book_id in items: item_users[book_id].add(uid) # 第二步:统计共现次数,即两本书同时被一个用户评分的次数 co_count = defaultdict(int) for uid, items in user_items.items(): book_ids = list(items.keys()) for i in range(len(book_ids)): for j in range(i + 1, len(book_ids)): co_count[(book_ids[i], book_ids[j])] += 1 # 第三步:套用余弦相似度公式 item_sim = {} for (book_i, book_j), count in co_count.items(): denom = math.sqrt(len(item_users[book_i]) * len(item_users[book_j])) item_sim[(book_i, book_j)] = count / denom if denom else 0.0 return item_sim

这一步有个性能细节值得琢磨:共现矩阵的双层循环只遍历了每个用户评分列表内的组合,没有全局笛卡尔积,这是一个合格实现该有的样子。如果用户量很大,这步计算会非常耗时,所以项目里把它放在离线阶段做,算完之后把结果序列化成itemsim_matrix_str.txt,每行存一段“图书A,图书B,相似度”,推荐服务启动时直接读文件建索引。

参数上,item_users是一个以 book_id 为 key 的 set 集合,set 保证了同一个用户不会重复计数;co_count记录的是无方向的共现次数,所以内层循环从i + 1开始,避免同一对书算两次。如果后续你想改成 Jaccard 相似度,只需要把分母换成双方被评集合的并集大小,改动非常小。

2.4 推荐生成:怎么把相似度结果变成推荐列表

相似度算完,推荐就变成了一件查表累加的事。系统拿到一个用户的历史评分,遍历用户评过的每本书,从相似度表里找出和它最像的一批书,把“用户对已评书的评分 × 两本书的相似度”累加到候选书的得分上,最后按总分排序取前 N 个,同时要把用户已经读过的书过滤掉。

# itemcf.py def load_sim_table(path): # 从离线文件加载相似度表,按源书聚合 sim_table = defaultdict(list) with open(path, 'r', encoding='utf-8') as f: for line in f: parts = line.strip().split(',') if len(parts) < 3: continue book_i, book_j, sim = parts[0], parts[1], float(parts[2]) sim_table[book_i].append((book_j, sim)) # 每个源书的相似书按相似度降序排列,方便截断 for book_id in sim_table: sim_table[book_id].sort(key=lambda x: x[1], reverse=True) return sim_table def recommend_from_table(user_id, user_items, sim_table, top_k=10): ranked = defaultdict(float) user_rated = user_items.get(user_id, {}) for book_id, score in user_rated.items(): # 只取相似度最高的前 20 个候选,控制计算量 for sim_book, sim_val in sim_table.get(book_id, [])[:20]: if sim_book in user_rated: continue # 不让已读书目再次进入列表 ranked[sim_book] += score * sim_val return sorted(ranked.items(), key=lambda x: x[1], reverse=True)[:top_k]

这个实现里有三个细节对应了真实的推荐工程经验。第一是截断:每个源书只取前 20 个相似物品,避免长尾物品的微弱相似信号无限累加;第二是去重过滤:sim_book in user_rated保证推荐列表里不出现用户已经评过分的书,这是推荐可用性的底线,截图里的“推荐列表与用户评分表”页面能直接看到这个效果;第三是分数归一化:如果用户评分习惯偏保守,普遍只打 2 到 3 分,那么加权得分会整体偏低,但这不影响排序,因为是相对比较。

3. 工程代码解析:Flask Web 层与数据层是怎么串起来的

3.1 目录结构与请求流转:一个推荐请求从进入到返回的完整路径

整个项目不是只有一个算法文件,而是一套完整的 Flask Web 应用。从资源清单看,根目录下同时存在app.py、models.py、db.py、dbapis.py、itemcf.py、templates/、static/和 SQL 脚本,分工一目了然。我拆包之后的目录结构认知大致如下:

文件/目录职责
app.pyFlask 入口,路由注册与请求分发
models.py数据类定义,对应数据库表结构
db.py数据库连接配置与连接对象
dbapis.py数据访问层,封装所有 SQL 操作
itemcf.py协同过滤算法,矩阵构建与推荐
user_item_matrix_str.txt离线导出的用户-物品评分矩阵
itemsim_matrix_str.txt离线导出的物品相似度表
bookrecommend.sql建库建表脚本与种子数据
templates/页面模板
static/CSS、JS、图片等静态资源

一个典型请求的流转是这样的:用户在前端“推荐列表”页面点击某个用户,浏览器发起/recommend/<user_id>请求,Flask 路由接收到参数后调用dbapis.py里的查询函数拿到数据,再让itemcf.py里的推荐函数基于内存中已加载的相似度表生成结果,最后把列表交给templates/下的模板渲染成 HTML 返回。注意整个过程中算法模块不直接依赖 Flask,它只接受干净的 Python 数据结构,这种解耦让算法可以脱离 Web 层做单元测试,我实际测算法时就是直接调用函数,没起服务。

3.2 数据访问层:db.py 里的连接配置与 dbapis.py 的 SQL 封装

数据库层是所有数据的来源,db.py在最简单的情况下封装了一个连接函数,配置集中在一个字典里。我拿到源码时第一件事就是改这里的密码,猜也能猜到这是跑通的第一个拦路虎。

# db.py import pymysql from pymysql.cursors import DictCursor DB_CONFIG = { "host": "localhost", # 数据库地址,远程部署时改成服务器 IP "port": 3306, "user": "root", "password": "你的密码", # 改成你本机 MySQL 的密码 "database": "bookrecommend", "charset": "utf8mb4", # 用 utf8mb4,兼容表情符号和生僻字符 } def get_conn(): return pymysql.connect(**DB_CONFIG, cursorclass=DictCursor)

这里提醒一句:cursorclass=DictCursor很关键。设置了它之后,查询返回的每一行都是字典,可以用row["book_id"]取值,而默认的元组游标只能靠下标取,代码可读性差一个量级。项目里models.py的数据类就是从这类字典行构造出来的。

dbapis.py则负责把所有 SQL 收拢到一个文件里,而不是散落在路由中。常见的设计是它导出fetch_all_ratings()、get_all_books()、save_rating(...)这类函数,每个函数内部完成“拿连接、执 SQL、关连接”的标准三步。

# dbapis.py from db import get_conn def fetch_all_ratings(): conn = get_conn() try: with conn.cursor() as cur: cur.execute("SELECT user_id, book_id, score FROM ratings") rows = cur.fetchall() return [(row["user_id"], row["book_id"], row["score"]) for row in rows] finally: conn.close()

这段代码采用了try/finally结构确保连接一定被关闭,避免 MySQL 连接泄漏——Flask 调试模式开了重载之后,泄漏最容易在不知不觉中发生。返回的列表直接喂给itemcf.build_user_item_matrix,与算法模块的数据格式对齐,这也是两个模块能解耦的原因。如果你要改数据源,比如换成 SQLite 或 CSV,只需要把这一层的返回格式保持成(user_id, book_id, score)三元组列表,算法部分完全不用动。

3.3 Web 视图层:路由、模板和管理页面的组织方式

Web 层相对套路化,但它是系统可用性的直接体现。app.py里通常会注册这样几个核心路由:首页渲染最近热门图书,图书列表页分页展示书目,图书详情页展示信息与历史评分,评分接口接收 POST 请求写入评分,推荐页针对指定用户展示结果,管理页面维护图书和用户数据。截图里看到的“图书列表”“图书详情”“历史评分列表与推荐列表”“管理页面”正好一一对应这些路由。

# app.py 路由节选 from flask import Flask, render_template, request, redirect from dbapis import get_all_books, fetch_ratings_by_user, save_rating from itemcf import generate_recommendation app = Flask(__name__) @app.route("/") def index(): books = get_all_books() return render_template("index.html", books=books) @app.route("/book/<int:book_id>") def book_detail(book_id): book = get_book(book_id) ratings = fetch_ratings_by_user(book_id) return render_template("book_detail.html", book=book, ratings=ratings) @app.route("/recommend/<int:user_id>") def recommend(user_id): rec_list = generate_recommendation(user_id) return render_template("recommend.html", recommendations=rec_list) @app.route("/rate", methods=["POST"]) def rate(): user_id = request.form.get("user_id", type=int) book_id = request.form.get("book_id", type=int) score = request.form.get("score", type=float) if user_id and book_id and score: save_rating(user_id, book_id, score) return redirect(request.referrer or "/")

这里有两个值得注意的参数设计。request.form.get("score", type=float)直接用 Flask 提供的类型强转,如果前端传了非数字,type=float会把它转成默认值,避免后端手写一堆异常判断;redirect(request.referrer or "/")做的是评分后的回跳,用户从哪个页面进来评分,评完就回哪去,这个体验细节很多新手项目不做,用户会被硬生生丢回首页。

模板层在templates/下,static/里放的是页面样式和截图里那些界面的前端资源。这套代码没有把前后端分离,用的是 Jinja2 服务端渲染,对单机部署、毕业设计、小型图书馆内部系统来说,维护成本远低于 Vue+Flask 分离架构。

4. 跑通全流程:环境准备、数据库初始化与页面验证

4.1 环境准备与依赖安装

跑这套项目的软硬件门槛不高,一台普通电脑就够。Python 用 3.7 以上版本即可,我本地用 3.8 跑通的,如果你的机器装了 Anaconda 也不需要额外建虚拟环境,直接装依赖就行。数据库侧需要 MySQL 5.7 或 8.0,确保服务已经启动。

代码依赖只有两个核心包,Flask 负责 Web 服务,PyMySQL 负责连数据库。安装命令很简单:

pip install flask pymysql

装完可以用python -c "import flask, pymysql"验证,不报错就说明依赖就绪。这里我一般会顺手看一眼app.py的 import 列表,确认它是否还引用了其他第三方库——如果有,再逐个补装,这个项目我拆的时候没发现额外依赖,干净得让人舒服。

4.2 数据库初始化:导入 bookrecommend.sql

数据库的初始化是整套流程里最“玄学”的一步,出问题的大部分原因都出在字符集和 MySQL 版本差异上。项目里打包的bookrecommend.sql是建库脚本,同时包含了表结构和初始种子数据。导入方式是在命令行里执行重定向:

mysql -u root -p < bookrecommend.sql

执行后输入密码,如果没有报错,进入 MySQL 确认一下表结构:

SHOW TABLES;

正常情况下能看到用户表、图书表、评分表若干张,常见的命名是users、books、ratings,具体表名以你的 SQL 脚本为准。评分表是核心,里面的user_id、book_id、score三列就是算法消费的原始数据。种子数据一般包含几十个用户和几百条评分记录,数量虽然不大,但足够把推荐算法跑出肉眼可见的效果,比空表启动有意义得多。

导入之后建议抽查一条数据,比如查某本书的评分分布,确认分数范围确实是 1 到 5,这关系到后面推荐结果的合理性。如果查到极端数据,大概率是评分接口没做校验,可以在后续二次开发里补上。

4.3 配置修改与启动

接下来改db.py。打开文件,把DB_CONFIG里的password改成自己本机 MySQL 的密码,如果 MySQL 不是跑在默认端口 3306,把port一并改掉。这一步改完,系统的数据通道就通了。

然后在项目根目录启动:

python app.py

看到类似Running on http://127.0.0.1:5000的输出说明 Flask 已经起来了。浏览器访问http://127.0.0.1:5000,如果首页能显示图书列表,数据库连接就已经验证通过。

管理员入口在路由里单独注册,路径一般是/admin或/manage。初始账号密码不要指望代码注释里有,我当时的做法是直接在 MySQL 的users表里把一条用户记录的密码改成已知值,再登录管理页,比猜默认密码省时间。当然前提是你已经理解了这张表的字段结构。

4.4 页面功能验证与论文测试章节对照

系统跑起来之后,不能只看首页能开就收工。我习惯按照截图里的功能把每个页面都点一遍:图书列表页翻几次分页,图书详情页确认能看到评分和简介,在详情页打个分提交,再去用户评分页确认这条评分落库,最后打开推荐列表,看推荐结果有没有把已读书目过滤掉。

这一步本质上就是论文里“系统测试”章节做的功能测试。论文把这部分写成了几个模块的测试用例,我在实际验证时发现一个有用的技巧:找一个评分记录最多的用户点进推荐页,如果这个用户的推荐列表里一本他评过的书都没有,说明算法和过滤逻辑都正常工作;如果推荐列表为空,大概率是评分数据太稀疏,说明系统的冷启动问题在真实数据上会暴露出来。

还有一个值得做的验证是重启服务看推荐是否仍在。因为相似度矩阵是从itemsim_matrix_str.txt预加载的,重启之后推荐结果应该保持一致,这说明离线计算结果被正确缓存了,这也是这套架构相比每次实时重算的优势所在。

5. 避坑记录:从 SQL 导入到推荐为空的高频问题

5.1 SQL 导入报错:字符集与版本不匹配

现象是执行mysql -u root -p < bookrecommend.sql时出现ERROR 1064语法错误,或者是中文数据变成乱码。

原因大多出在两方面:一是 SQL 文件头部的建库语句指定了字符集,但本机 MySQL 版本或配置不支持;二是脚本里用了低版本 MySQL 不认识的新语法。

解决方法是先确认 MySQL 版本,再用客户端工具(Navicat 或命令行)新建一个utf8mb4字符集的空库,然后手动执行 SQL 文件里的建表语句。如果表格中的中文仍然是问号,检查连接参数是否带上了charset=utf8mb4,db.py里的配置和导入工具需要保持一致。

5.2 矩阵文件读取乱码或解析失败

现象是启动报UnicodeDecodeError,或者推荐结果全是零。

原因是user_item_matrix_str.txt和itemsim_matrix_str.txt这两个文本文件的编码可能不是你当前系统的默认编码。Windows 下用记事本保存的文本常带 BOM,Linux/Mac 下直接按 UTF-8 读会撞上 BOM 头,导致第一行解析出的第一列包含不可见字符。

解决方法是读取时统一指定容错编码,常见写法是open(path, 'r', encoding='utf-8-sig'),它能同时兼容带 BOM 和不带 BOM 的文件。如果还不行,用十六进制编辑器看一眼文件头,确认真正的编码再改读取参数。我拆这套项目时就是被这一个小问题卡了半天,最后发现是编码参数写成了utf-8,换掉就一切正常。

5.3 推荐列表为空:冷启动与数据稀疏

现象是页面能打开、用户能查到,但推荐列表一行数据都没有。

原因是协同过滤完全依赖评分共现。用户只评过一两本书,或者所有用户之间没有任何相同评分记录时,相似度矩阵就是稀疏的目录,找不出足够相似的书。

解决方法是先检查该用户的评分记录条数,然后从后端临时加一个兜底逻辑:当候选数不足时,返回评分最高的热门图书作为推荐。这也是生产系统的常规操作,在recommend_from_table函数的最后加一个if not ranked分支,从图书表按均分排序取前 N 本返回。冷启动问题不可能彻底消除,但兜底逻辑能保证界面上至少有内容可看。

5.4 数据库连接失败:配置与服务状态排查

现象是启动后访问页面报OperationalError: (2003, "Can't connect to MySQL server")或Access denied for user。

原因分两类,前者是 MySQL 服务没启动或端口不对,后者是账号密码错误。还有一个隐蔽原因:db.py和dbapis.py里各自维护了一份连接参数,改了一处忘了另一处,导致启动时用新配置、运行时用旧配置。

解决方法是先netstat -ano | findstr 3306确认 MySQL 在监听,再用命令行手动连一次数据库验证账号密码;同时全局搜索所有pymysql.connect调用点,统一收口到db.py的DB_CONFIG。从那以后我再看到项目里重复的数据库配置,就会条件反射地想合并。

5.5 Flask 5000 端口被占用:启动即崩溃

现象是执行python app.py后立刻报Address already in use。

原因很简单,5000 是 Flask 默认端口,很容易被其他开发中的服务占用。

解决方法是换端口启动,app.py末尾的app.run(debug=True, port=5001),或者命令行指定python app.py --port 5001。换端口之后浏览器访问地址同步改一下就行。这个坑虽然低级,但新手遇到时往往会误判成代码问题,折腾半天才发现只是端口打架。

6. 二次开发:把离线批量推荐改造成线上增量推荐

6.1 改造相似度更新机制:从手动跑脚本到定时重建

这套系统目前的流程是离线计算相似度、导出文本、服务启动时加载,优点是查询快,缺点是新增评分后相似度不会立即更新。真实线上系统里,我一般的做法是写一个独立的维护脚本,把“查全部评分 → 构建矩阵 → 计算相似度 → 导出文件”封装成可调用的函数,然后用系统定时任务每天凌晨跑一次。

改造点很小,只需要把itemcf.py里散落的步骤收进一个入口函数,比如rebuild_model(),脚本执行完自动覆盖两个 txt 文件,服务端也不需要重启,因为每次推荐请求都重新读取文件或者用文件更新时间做缓存淘汰。这样就能保证第二天的推荐包含了前一天的评分变化。

6.2 相似度算法替换:从余弦到皮尔逊

余弦相似度适合评分尺度一致的用户群,但如果用户打分习惯差异很大,有的人只打 4 到 5 分,有的人只在 1 到 3 分之间徘徊,余弦结果会被评分尺度带偏。皮尔逊相关系数先减掉用户均分再做计算,能有效消除打分习惯差异。

实现替换时不需要动整体架构,只需要在calc_item_similarity里把共现次数换成分数协方差计算:

# 皮尔逊相似度实现片段 def calc_pearson_sim(user_items): # 先算每本书的平均分 item_avg = {} ... # 对共同评过分的用户,计算去均值后的乘积累加 sim = numerator / math.sqrt(denom_i * denom_j)

替换后不只是公式变了,推荐结果也会跟着变:用户给某本书打了 3 分,而该用户均分是 2,那么这个 3 分代表的是正向偏好;如果用余弦,原分数直接参与加权,容易被习惯高分的人淹没。改完公式后用同一批历史数据对比新旧结果,能看到明显差异。

6.3 离线评估:用留一法验证改造效果

无论怎么改,都需要一个量化的验证手段。最实用的做法是留一法:从某个用户的评分记录里随机抽出一本书当作“测试答案”,用剩下的数据生成推荐,看看被抽出的书在不在推荐列表里,循环多次统计命中率。

# evaluate.py 留一法评估 def leave_one_out_eval(user_items, sim_table, top_k=10): hit = 0 total = 0 for uid, items in user_items.items(): for held_out in list(items.keys()): # 临时移除被抽出的书,基于剩余数据重新生成推荐 train_items = {uid: {b: s for b, s in items.items() if b != held_out}} recs = recommend_from_table(uid, train_items, sim_table, top_k) if any(book_id == held_out for book_id, _ in recs): hit += 1 total += 1 return hit / total

这个函数会告诉我每次改动是在提升还是在倒退。我拆这套项目时最大的教训就是:推荐系统的改造很容易陷入“感觉变好了”的错觉,必须有一个可复现的评估脚本把指标钉死。从那以后我每次改相似度逻辑,都强制先跑一遍留一法记录基线,再动手改代码,改完再跑一遍对比数值,肉眼观察到的推荐结果只作为定性参考,不在没有指标的情况下上线任何算法调整。这套项目的架构给了我很舒服的改造空间,因为算法层与 Web 层完全解耦,我可以不碰任何页面代码,单独把推荐模型迭代一轮。希望这个改造路径对你也有参考价值。

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

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询