Flask+协同过滤图书推荐系统源码拆解：从评分矩阵到Top-N推荐

简介：基于Flask与协同过滤算法的图书推荐系统毕业设计项目，面向需要完成Python类毕设的在校学生，提供一套可运行、可解释的高分参考方案。项目以图书评分数据为核心，实现用户登录、图书展示、协同过滤推荐、排行榜等常见功能，前端页面与后端接口齐备，覆盖Flask路由、数据持久化及推荐算法封装等关键环节。压缩包共55个文件，大小仅3.72MB，其中包含6个Python源码文件、19个txt文档/说明、9个HTML页面、1个SQL数据库脚本，以及配套的CSS、JS、图片和项目配置文件，整体结构清晰，按模板、数据、接口分层，便于快速定位学习。目前已有216人次学习下载，资料内附完整源码、详细设计文档和数据库初始化数据，所有代码经本地编译通过，评审分达95分以上。项目难度适中，既能辅助理解协同过滤算法在真实场景下的工程落地，也可直接作为课程设计或毕业设计的开发起点，按需修改即可复用。

1. 一份能直接跑通的图书推荐系统：Flask + 协同过滤源码拆解

图书推荐系统在毕业设计里算是常青树了。不过网上流传的代码包，十有八九是 Java Spring Boot 写的，光 Maven 依赖就能把人绕晕。我这段时间拆了这套 Python Flask + 协同过滤的图书推荐系统源码，比想象中清爽：后端按功能拆分模块，推荐逻辑集中在一个 Python 文件里，SQLite 做存储，建表语句、接口说明和部署文档都收在配套资料里。它能做三件事——用户注册登录、给图书打分、系统基于评分数据生成 Top-N 推荐列表。适合要交毕设、需要快速跑通推荐链路的人，也适合第一次接触 Flask 后端但想看看完整项目长什么样的初学者。本地部署不需要额外中间件，Python 3.8 以上环境就能直接跑，这也是我推荐先拆它而不是直接上手 Java 方案的原因。

2. 协同过滤的两条路线：为什么图书推荐要选基于物品的 CF

2.1 用户-物品评分矩阵：推荐系统的地面真相

协同过滤这个名字听起来高深，起点却特别朴素：把"谁给什么书打了多少分"整理成一张二维表。行是用户 ID，列是图书 ID，交叉点是评分。用户没读过的书在表格里就是空着的，评分数据天生稀疏，这一点是后面很多坑的根源。

import pandas as pd ratings = pd.read_sql_query( "SELECT user_id, book_id, score FROM ratings", engine ) pivot = ratings.pivot_table( index='user_id', columns='book_id', values='score' ).fillna(0) print("矩阵形状:", pivot.shape) active_rows = (pivot > 0).sum(axis=1) print("每行评分数分布:") print(active_rows.describe())

这段代码在源码包里位于rec_utils.py开头。逻辑说明：read_sql_query先把评分表拉出来，pivot_table按用户和图书展开成矩阵，fillna(0)把没有评分的格子填成 0。最后两行 print 用来观察矩阵稀疏程度——active_rows.describe()输出评分条数的均值、分位数，如果中位数只有个位数，说明大多数用户只评过几本书。

参数说明：pivot_table中index是透视表的行维度，columns是列维度，values是要聚合的数值字段。这里没写aggfunc，默认取均值，也就是说同一用户对同一本书如果有多条评分记录，pandas 会自行平均。想要日志更明确，可以显式写aggfunc='mean'。fillna(0)这一步不做，后面余弦相似度算出来的就是 NaN 矩阵，排错会特别难受。

实际跑出来的矩阵通常是这样的：大部分用户只有 5 到 20 条评分，极少数活跃用户评了上百本。这种长尾分布说明热门书之间的共现关系足够多，推荐效果不会太差；反而是那些只有一两人评分的冷门书，基本注定进不了推荐列表。如果评分数据比这还稀疏——平均每个用户只有 2 条——那先别急着调算法，回去把评分功能在界面上做得更显眼，先把数据密度提上来。

2.2 基于用户的 CF 与基于物品的 CF：公式与直觉差异

协同过滤有两条最常见的实现路线。基于用户的 CF（User-based CF）先找"跟我品味相似的用户"，再看这些人还读过什么书，把那些书推荐给我。基于物品的 CF（Item-based CF）反过来，先找我评分高的书有哪些"兄弟书"，把兄弟书推荐给我。两条路线都离不开相似度计算，最常用的是余弦相似度：

similarity(A, B) = (A · B) / (|A| × |B|)

A 和 B 是两个评分向量，分子是点积，分母是模长乘积。向量方向越一致，值越接近 1。举个例子：用户 A 给《三体》打 5 分、《球状闪电》打 4 分，用户 B 给《三体》打 4 分、《球状闪电》打 5 分，这两条向量的余弦相似度会超过 0.9。系统会认为 A 和 B 品味接近，于是把 B 读过的《流浪地球》推荐给 A。

在图书推荐这个场景里，我建议优先实现基于物品的 CF。核心原因是图书间的关联关系比用户间的关系稳定得多——某两本书反复出现在同一批用户的评分列表里，这个模式不会因为个别用户口味变化就崩塌。而基于用户的 CF 要求每个用户都有足够多的评分记录，否则冷启动阶段连"相似用户"都找不出来。这套源码默认走的就是 Item-based 路线，核心代码只有几十行，比想象中好拆。

2.3 相似度度量怎么选：余弦、皮尔逊、杰卡德的边界

源码默认用余弦相似度，并且在计算前对向量做了归一化——每个评分除以向量模长再做点积。这样评分的绝对大小影响变弱，算法更看重"你喜欢的方向"。如果想把皮尔逊换上去，代码改动不大，但一定要处理除零：评分矩阵里有一行只有一个非零值时，皮尔逊分母直接变 0，返回 NaN 会污染后续所有推荐结果。

可能有人会问，既然矩阵都建好了，为什么不直接上 SVD 矩阵分解？这里有一个实际答辩的原因：这套项目的数据规模很小，通常几百个用户、几千本书，SVD 分解出来的隐因子很难解释，"第 4 个隐因子代表什么"这一关就很难过。协同过滤至少能直接回答"这本书和哪本书像、依据是什么"，解释成本低很多。这也是源码选 CF 而不是矩阵分解的合理之处。

3. Flask 后端与数据库：三张表如何撑起评分和推荐接口

3.1 建表 SQL：用户、图书、评分三张核心表的结构

拿到源码先把数据库脚本打开看一遍。整个项目只有三张业务表，表结构设计得规矩，外键关系和唯一约束都考虑到了：

-- 用户表 CREATE TABLE users ( user_id INTEGER PRIMARY KEY AUTOINCREMENT, username TEXT NOT NULL UNIQUE, password_hash TEXT NOT NULL, created_at DATETIME DEFAULT CURRENT_TIMESTAMP ); -- 图书表 CREATE TABLE books ( book_id INTEGER PRIMARY KEY AUTOINCREMENT, title TEXT NOT NULL, author TEXT, category TEXT, pub_year INTEGER, created_at DATETIME DEFAULT CURRENT_TIMESTAMP ); -- 评分表 CREATE TABLE ratings ( rating_id INTEGER PRIMARY KEY AUTOINCREMENT, user_id INTEGER NOT NULL, book_id INTEGER NOT NULL, score INTEGER NOT NULL CHECK(score BETWEEN 1 AND 5), created_at DATETIME DEFAULT CURRENT_TIMESTAMP, FOREIGN KEY (user_id) REFERENCES users(user_id), FOREIGN KEY (book_id) REFERENCES books(book_id), UNIQUE (user_id, book_id) );

逻辑说明：users表存账号，密码存哈希而不是明文；books表是图书元数据，category字段可以给后面做"同类推荐"留余地；ratings表是核心，UNIQUE(user_id, book_id)保证一个用户对同一本书只能有一条评分记录，再评分就是更新，不会累积出重复数据。

参数说明：score字段用CHECK约束限制在 1 到 5 之间，这比在后端写 if 判断更可靠——数据库层面直接拒绝非法值。pub_year字段选INTEGER而不是TEXT，方便后面做年份过滤。如果自己的数据里有多个出版社版本的书，可以在books表加一个isbn字段，建唯一索引防重。

浏览这套源码时我注意到一个小细节：ratings表没有单独给user_id建索引。数据量小的时候没问题，但如果想把项目扩展成能支撑几千用户的演示，CREATE INDEX idx_ratings_user ON ratings(user_id);这条语句值得补上，推荐接口的查询速度会明显改善。

3.2 Flask 应用初始化与模块划分

Flask 项目的入口文件代码量不大，但有几个配置值得注意：

from flask import Flask from flask_sqlalchemy import SQLAlchemy app = Flask(__name__) app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///books.db' app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False app.config['JSON_AS_ASCII'] = False db = SQLAlchemy(app) # 拆分蓝图：页面路由与推荐接口分开 from views import main_bp from recommend_api import rec_bp app.register_blueprint(main_bp) app.register_blueprint(rec_bp) if __name__ == '__main__': app.run(debug=False, host='0.0.0.0', port=5000)

逻辑说明：SQLALCHEMY_DATABASE_URI指定 SQLite 数据库文件路径，books.db会在首次运行后自动在项目根目录生成。JSON_AS_ASCII = False这个配置很关键——不关掉的话，Flask 返回中文 JSON 时会被转成\uXXXX的形式，前端拿到还得自己转码。路由拆成两个蓝图，main_bp管页面渲染和登录注册，rec_bp管评分提交和推荐接口，这样推荐逻辑不会和页面代码混在一起。

参数说明：app.run里的debug=False是给正式运行用的，本地调试可以改成debug=True，改完代码自动重载，不用手动重启。host='0.0.0.0'允许局域网内其它设备访问，写毕业设计答辩时，评委如果连的是同一个 WiFi，可以直接用你电脑的局域网 IP 打开页面。

3.3 评分与推荐接口：请求/响应协议说明

接口协议在源码文档里专门占了两页，这部分对答辩特别重要。先把协议梳理清楚，再看实现会轻松很多：

响应示例：

{ "code": 0, "msg": "ok", "data": [ {"book_id": 3, "title": "三体", "score": 4.8}, {"book_id": 27, "title": "球状闪电", "score": 4.5} ] }

逻辑说明：评分接口接收一个 JSON 对象，三个字段缺一不可，返回的code字段为 0 表示成功。推荐接口路径参数是用户 ID，top_n作为查询参数表示要几条推荐，data数组里的score是系统算出来的推荐得分，不是用户评分，这两个概念在文档里做了区分，接口对齐的时候容易踩混。

源码里对错误码的定义很明确：1 表示参数缺失，2 表示用户不存在，3 表示图书不存在。答辩时被问到"怎么保证接口健壮性"，直接拿这套错误码回答，比较有说服力。

4. 推荐引擎实现与调参：把相似度计算写进路由里的细节

4.1 构建相似度矩阵的核心代码

推荐引擎的全套逻辑集中在一个模块里，核心是先把评分矩阵转成"图书 x 图书"的相似度矩阵：

import numpy as np def build_item_similarity(pivot): # pivot: user x book 的评分矩阵 # 按列做归一化，每个图书向量缩放到单位长度 norm = pivot / np.sqrt((pivot ** 2).sum(axis=0)) # 相似度矩阵 = 归一化向量的两两点积 sim = norm.T.dot(norm) return sim

逻辑说明：这一步做的是基于物品的协同过滤。pivot是用户 x 图书的矩阵，对每一列（即每一本书）做归一化；norm.T.dot(norm)得到的矩阵，第 i 行第 j 列就是图书 i 和图书 j 的余弦相似度。为什么先归一化再点积？因为余弦相似度的数学表达式就是归一化后的内积，把除以模长的步骤提前做掉，后面的计算更快，可读性也更好。

参数说明：sum(axis=0)表示按列求和。用 pandas DataFrame 存储时，要确认pivot的索引是用户 ID、列是图书 ID，顺序反了的话相似度矩阵就变成"用户 x 用户"了。每次数据更新后这个矩阵要重建，构建成本是 O(books²)，几百本图书时毫秒级完成，等攒到上万本书再考虑用近似算法。

我拆过几个推荐项目，发现一个共性：后面调参调得晕头转向的时候，回头看基本都是这一步的归一化没做对。推荐得分算出来全在 0.5 上下浮动，那多半是向量没有缩放到单位长度，直接拿原始评分做了点积。调参有时候接近玄学，但先把数据管干净比什么都重要。

4.2 生成推荐列表：排除已读书和 Top-N 截断

相似度矩阵到手后，推荐的生成过程其实就三步：

def recommend_for_user(pivot, sim_matrix, user_id, top_n=10): user_ratings = pivot.loc[user_id] # 相似度矩阵与评分向量相乘，得到每个图书的加权得分 scores = sim_matrix.dot(user_ratings) # 过滤已读图书，避免重复推荐 rated_books = user_ratings[user_ratings > 0].index scores = scores.drop(index=rated_books) top_items = scores.sort_values(ascending=False).head(top_n) return top_items.index.tolist(), top_items.values.tolist()

逻辑说明：sim_matrix.dot(user_ratings)是这一步的核心——相似度矩阵与用户评分向量相乘，相当于把"这本书与用户看过的书的相似度"按评分加权求和。比如用户给《三体》打了 5 分，那跟《三体》相似度 0.8 的《球状闪电》会拿到 4 分的加权贡献。drop操作把用户已经读过的书排除掉，避免推荐重复内容。最后用sort_values取 Top-N。

参数说明：top_n的值直接影响推荐列表节奏，源码默认 10。实际调参时可以设 15 或 20，前端再展示前 10 条，这样即使后几条质量不高也有缓冲。如果评分数据特别稀疏，建议把top_n降到 5——与其推 10 本里有 7 本是凑数的，不如只推 5 本有把握的。

4.3 冷启动：新用户和新图书的兜底策略

冷启动是推荐系统绕不开的话题。这套源码的处理方式比较务实：

def get_recommendations(user_id, pivot, rating_stats): # 用户不存在，直接给热门榜 if user_id not in pivot.index: return get_hot_books(rating_stats, limit=10) # 评分太少，给热门榜兜底 user_rated = (pivot.loc[user_id] > 0).sum() if user_rated < 3: return get_hot_books(rating_stats, limit=10) # 正常用户走协同过滤 sim = load_similarity_matrix() return recommend_for_user(pivot, sim, user_id, top_n=10) def get_hot_books(rating_stats, limit=10): return rating_stats.sort_values('avg_score', ascending=False).head(limit).to_dict('records')

逻辑说明：get_recommendations入口做了两个判断——用户不存在时直接返回热门榜；用户评分少于 3 条时也返回热门榜。为什么是 3 条？因为少于 3 条评分时，相似度计算的结果置信度太低，还不如老老实实给热门书。rating_stats是预先聚合好的图书平均分和评分人数，热门榜在这里按平均分排序。

参数说明：评分阈值的 3 不是拍脑袋定的。在实际数据里，评分条数少于 3 的用户，其评分向量和任何一本书的相似度都可能恰好算出高分，但那纯属偶然。想更稳妥，可以把阈值提到 5。热门榜的分母建议加一个评分人数过滤，比如评分人数少于 5 的图书不参与热门排序，不然一本只有一个人打 5 分的冷门书会常年霸榜——这个逻辑源码里没有，我建议自己补上，改动只有三行。

5. 避坑排查：5 个最容易翻车的地方

5.1 现象：推荐接口返回的数据全是 NaN

调试时打出来的推荐结果里，book_id正常，但推荐得分全是NaN。原因多半是评分矩阵在构建时没有处理空值，或者pivot.fillna(0)被注释掉了。矩阵里只要有一个 NaN，后续所有 dot 运算都会把 NaN 传播下去。解决方法是确认pivot在进入build_item_similarity之前已经fillna(0)，而且全项目只做这一处填充：

pivot = pivot.fillna(0) # 全局只做一次，别在相似度函数里再做

一条血泪经验：不要图省事在相似度计算函数内部fillna，因为全局填充和局部填充的效果不一样。全局填过以后再统计每个用户的评分条数，结果才稳定；局部填充会因为函数多次调用导致数据状态不一致，排查起来非常痛苦。

5.2 现象：Windows 下路径拼接报错

同样的代码在 Windows 上运行，favicon、模板文件、静态图片加载 404。原因是项目里用了硬编码的/拼接路径，Windows 文件系统用的是反斜杠。Flask 的url_for还好，但open('template/' + name)这种写法在 Windows 上必翻车。解决方法是统一用app.root_path拼路径，或者直接用pathlib：

from pathlib import Path BASE_DIR = Path(__file__).resolve().parent

后来我把项目里所有涉及文件路径的地方都改成了基于BASE_DIR的相对路径，再也没在路径上出过幺蛾子。

5.3 现象：SQLite 里存的中文出现乱码

通过页面表单提交中文书名后，从 SQLite 查出来变成乱码。这里有两个可能：数据在写入时就乱了，或者只是终端输出编码不对。SQLite 本身存储 UTF-8 没问题，问题通常出在连接字符串没指定编码，或者命令行终端用 GBK 去看数据。先确认数据落库时是否正常，再考虑输出环节：

app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///books.db'

flask_sqlalchemy处理 sqlite 时其实不需要加charset=utf8，加了反而报错。真正要验证的是数据写入那一步——在 route 里打印request.get_json()拿到的原始内容，看中文是否正常；如果原始内容没问题，那乱码只发生在终端显示，用chcp 65001切到 UTF-8 就能解决。

5.4 现象：评分接口偶尔报"score 参数错误"

前端明明传了{"score": 4}，后端却报参数错误。原因是request.get_json()拿到的 JSON 里，数字本来应该是 int，但前端如果用了 form-data 提交，或者做了JSON.stringify，score会变成字符串"4"。后端不能假设前端传什么类型，统一做一次转换：

try: score = int(data['score']) except (KeyError, TypeError, ValueError): return jsonify({'code': 1, 'msg': 'score参数错误'})

5.5 现象：推荐接口响应越来越慢

本地测试时第一次请求 1 秒返回，多跑几次变成 3 秒以上，而且随着评分增多越来越慢。原因是每次请求都在重建相似度矩阵。评分数据几百条时无所谓，但图书数量过千后，这个计算会变成平方级开销。解决方法很简单，把相似度矩阵缓存到内存里，等评分数据有变化再失效：

_sim_cache = None _cache_user_count = 0 def get_similarity_matrix(pivot): global _sim_cache, _cache_user_count if _sim_cache is None or _cache_user_count != len(pivot): _sim_cache = build_item_similarity(pivot) _cache_user_count = len(pivot) return _sim_cache

这个缓存的思路是拿用户数当版本号——新增用户必然导致矩阵行数变化，此时重建；只更新评分但用户数不变时，矩阵数据已经变了，这里严格来说应该用评分总条数做版本号。我后来改成_cache_rating_count = int(pivot.sum().sum())，每次请求对比总评分条数，比len(pivot)更准确。

6. 离线验证推荐效果：从历史评分里切出训练集和测试集

6.1 按时间切分而不是随机切分

推荐系统的验证和普通分类模型不一样，不能随意 shuffle。用户的行为有先后顺序，拿未来的评分去预测过去的行为没有意义。正确做法是按时间切数据：把每个用户的评分记录按时间排序，前 80% 当训练集，后 20% 当测试集。

train_list, test_list = [], [] for user_id, group in ratings.groupby('user_id'): group = group.sort_values('created_at') split_idx = int(len(group) * 0.8) train_list.append(group.iloc[:split_idx]) test_list.append(group.iloc[split_idx:]) train_df = pd.concat(train_list) test_df = pd.concat(test_list)

6.2 一个能直接用的召回率计算脚本

召回率的定义很直接：测试集里有，而推荐列表也覆盖到的图书数量，除以测试集图书总数。代码量不大，但答辩时拿出来是加分项：

def calculate_recall(train_df, test_df, pivot_train, sim_matrix): hits = 0 total = 0 for user_id in test_df['user_id'].unique(): rec_books = set(recommend_for_user( pivot_train, sim_matrix, user_id, top_n=10 )) actual_books = set( test_df[test_df['user_id'] == user_id]['book_id'] ) hits += len(rec_books & actual_books) total += len(actual_books) return hits / total if total else 0

在图书推荐场景里，召回率我一般要求 0.15 以上就算能接受。评分数据越稀疏，这个值越难看，但至少能验证算法有没有跑偏——如果召回率是 0，那一定是相似度矩阵或者数据预处理出了问题，而不是调参的问题。

从那次以后，我每拿到一个推荐类项目，先切训练测试集、跑一次召回率，再谈调参。这个习惯救了我好几次。希望帮到你。

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