简介：这份基于 Python 的新闻推荐系统学习包面向希望理解推荐系统落地流程的开发者与高年级学生，围绕新闻文本整理、特征处理与推荐排序等环节，提供可拆解、可复用的工程样例。压缩包共 181 个文件，约 10.32MB，主要由 35 个 py 源码、102 个 txt 文本、32 个 pyc 编译文件构成，并配有 csv 数据样例、md 说明文档及工程配置文件，便于查看目录结构、模块依赖与运行线索。包内还包含概率发射、状态转移、字符状态表等模型参数文件，可辅助理解中文分词或文本预处理在新闻推荐中的具体作用。已有 4077 人学习，适合作为课程设计、毕业设计及推荐算法入门参考；读者可据此梳理从数据准备、文本处理到推荐逻辑实现的完整链路，并基于说明文档与源码进行调试、改造和二次开发。

1. 推荐位只按时间倒序，用户刷三条就退出了

有个做资讯后台的朋友给我看他们的首页：编辑把新闻按发布时间倒序一挂，看着挺整齐，可后台数据里点击几乎全压在头部三条，往下滑的人寥寥无几。问题不在内容量，在于每个人看到的顺序一模一样。后来他们换了一套按用户行为排的首页，同样的新闻池，停留时长涨了一截。

这份资源就是一个基于 Python 的新闻推荐系统，覆盖采集、清洗、分词、用户画像、召回、排序到 Web 接口展示的整条链路，代码结构清晰，适合想搞懂推荐系统怎么写、又不想从零搭脚手架的人。学生拿来做课程设计够用，开发者拿它当二次开发的底座也合适，尤其是想快速跑通“行为数据进、推荐列表出”这条闭环的。

2. 架构全景与技术选型：一条推荐结果要经过哪几道手

2.1 离线链路与在线链路分开跑

新闻推荐和电商推荐最大的区别是时效性。一条新闻的热度窗口可能只有几个小时，你不可能等全量模型训练完再推。所以这个项目的结构是典型的离线 + 在线双链路，两条链路靠数据库和缓存交接。

离线链路负责“重活”：定时抓取新闻列表与正文、清洗 HTML 噪声、中文分词、抽取关键词、计算新闻向量、根据历史行为构建用户画像、算物品相似度矩阵，最后把结果落到 MySQL 或 SQLite 里。这条链路慢一点没关系，通常按小时或按天跑。

在线链路负责“快活”：用户打开页面，Flask 接口拿到 user_id，先从 Redis 里查有没有预生成的推荐列表，命中就直接返回；没命中就现场从候选集里召回一批，做一次轻量排序，再写回缓存。整条链路的目标响应时间控制在几十毫秒级。

我一般会把两条链路的交接点设计得很窄，只通过两张表通信：一张存新闻向量和关键词，一张存用户画像和推荐结果。这样离线侧换算法不用动在线代码，在线侧改接口也不用重跑模型，排障的时候边界也清楚——推荐不准去看离线表，接口超时去看在线日志。

2.2 技术栈对照：每个环节为什么用这个库

这张表不是让你照抄，而是让你知道每个位置的可替换空间。比如你的新闻量上了十万级，TF-IDF 的全量矩阵就有点吃力，这时候换成预训练的句向量模型或者加一层倒排索引会更稳。但项目初期，上面这套组合的性价比是最高的，依赖少意味着踩坑少。

有一点要提醒：scikit-learn 的版本差异会直接影响 TfidfVectorizer 的分词行为，尤其是自定义 tokenizer 传入时，不同版本的参数校验规则变过。项目里如果锁了版本号，别随手升级。

2.3 目录结构逐层拆：每个文件夹到底放什么

news_recommend/ ├── app.py # Flask 入口，注册蓝图、启动定时任务 ├── config.py # 数据库、Redis、抓取参数集中配置 ├── requirements.txt ├── crawler/ │ ├── fetcher.py # 请求封装：UA、超时、重试、限速 │ ├── parser.py # 列表页与详情页解析、正文抽取 │ └── pipeline.py # 清洗、去重、入库的流水线 ├── recommender/ │ ├── segment.py # jieba 分词与关键词抽取 │ ├── vectorizer.py # TF-IDF 向量化、向量持久化 │ ├── profile.py # 用户画像构建 │ ├── recall.py # 协同过滤召回 │ ├── rank.py # 混合排序、多样性打散 │ └── evaluate.py # 离线指标计算 ├── web/ │ ├── routes.py # 推荐、详情、反馈三个接口 │ ├── cache.py # Redis 读写与降级 │ └── templates/ # 首页与详情页模板 ├── models/ │ └── schema.sql # 建表语句 ├── jobs/ │ └── scheduler.py # 定时任务注册 └── data/ └── stopwords.txt # 停用词表

这个结构的关键在于把「抓取」「算法」「服务」三层彻底隔开。crawler 只关心怎么把数据弄干净，recommender 只关心怎么算，web 只关心怎么返回。三层之间只通过数据库表和函数签名交互，不互相 import 内部实现。

我见过不少同类项目把所有逻辑塞进一个 app.py，两百行里夹着爬虫、分词、SQL 和路由，改一个地方全崩。这个项目的分层虽然朴素，但对二次开发友好得多——你想换召回算法，只需要保证 recall.py 的函数签名不变，其余代码一行不用动。

data 目录放停用词表这种静态资源是常规做法，注意停用词表要跟分词逻辑一起版本管理。我就吃过亏：换了一版停用词，历史新闻的向量没重算，新老向量不在一个空间里，推荐结果莫名其妙地偏。

2.4 环境与启动：三条命令把它跑起来

# 1. 建虚拟环境，Python 3.8 以上均可，3.9/3.10 最稳 python -m venv venv source venv/bin/activate # Windows 用 venv\Scripts\activate # 2. 装依赖，建议加国内镜像加速 pip install -r requirements.txt # 3. 初始化数据库（SQLite 会自动建文件，MySQL 需先建库） python manage.py initdb # 4. 先跑一次抓取，把新闻灌进去，否则推荐是空的 python manage.py crawl --pages 5 # 5. 跑一次离线计算，生成向量、画像和推荐结果 python manage.py offline # 6. 起服务 python app.py

这几步的顺序不能乱。很多人上来就python app.py，打开首页一片空白，以为是模板坏了，其实是新闻表里一条数据都没有，推荐接口拿不到候选集自然返回空列表。initdb 建表、crawl 灌数据、offline 算推荐，这三步是有依赖关系的流水线。

crawl --pages 5里的 pages 控制翻页数，第一次跑建议小一点，确认解析规则没问题再放开。offline命令背后是依次执行分词、向量化、画像、召回、排序五个函数，日志里会打印每一步的耗时和产出条数，哪一步产出为 0 就说明那一步有问题。

config.py 里需要你手动确认的通常只有三项：数据库连接串、Redis 地址、抓取目标站点的入口 URL。前两项本地开发用默认值就行，第三项要换成你实际要抓的资讯源。

注意：抓取前先确认目标站点的 robots 协议和访问频率限制，把并发数压到 1～2，间隔加到 1 秒以上。这不是客气，是避免 IP 被封导致后续调试全断。

3. 数据层拆解：新闻采集、清洗与入库的完整实现

3.1 采集入口：请求封装与列表页解析

推荐系统的上限由数据质量决定，而数据质量从请求那一行代码就开始分化了。项目里的 fetcher.py 做了三件容易被忽略的事：统一 UA、限制超时、失败重试。

import time import random import requests from requests.adapters import HTTPAdapter from urllib3.util.retry import Retry # 会话复用，避免每次请求都重新握手 session = requests.Session() session.headers.update({ "User-Agent": "Mozilla/5.0 (compatible; NewsBot/1.0)", "Accept-Language": "zh-CN,zh;q=0.9", }) # 重试策略：只对连接错误和 5xx 重试，4xx 直接放弃 retry = Retry(total=3, backoff_factor=1.5, status_forcelist=[500, 502, 503, 504]) session.mount("http://", HTTPAdapter(max_retries=retry)) session.mount("https://", HTTPAdapter(max_retries=retry)) def fetch(url, timeout=10): """抓单页，返回 HTML 字符串；失败返回 None 而不是抛异常""" try: resp = session.get(url, timeout=timeout) resp.encoding = resp.apparent_encoding # 中文站点关键 resp.raise_for_status() time.sleep(random.uniform(0.8, 1.5)) # 随机间隔，别用固定值 return resp.text except requests.RequestException as e: print(f"[fetch] {url} -> {e}") return None

几个参数值得说清楚。total=3是重试次数，配合backoff_factor=1.5，三次重试的等待时间大约是 0、1.5、3 秒，不会把服务端打穿。resp.apparent_encoding是根据内容猜测编码，比直接用resp.encoding靠谱，因为很多资讯站返回头里写的编码和实际字节并不一致，不设这一行，中文全变问号。

sleep用random.uniform而不是固定值，是因为固定间隔的请求序列特征太明显。这个细节在抓取量小的时候看不出差别，量一上去就是封 IP 和正常跑的区别。

返回值设计成失败返回 None 而不是抛异常，是为了让上层的流水线可以跳过失败项继续跑，一条抓不到不该让整批任务崩掉。上层拿到 None 就记一条失败日志，继续下一条。

3.2 正文抽取与字段标准化：把 HTML 洗成干净文本

列表页解析相对简单，按 CSS 选择器拿到标题、链接、发布时间就够了。麻烦的是详情页正文，导航栏、推荐位、版权声明、广告都会混进来，直接丢给 jieba 分词，出来的关键词全是“点击查看”“版权所有”。

import re from bs4 import BeautifulSoup # 正文容器候选选择器，按优先级排列，命中第一个就用 CONTENT_SELECTORS = ["article", ".article-content", "#content", ".post-body"] NOISE_TAGS = ["script", "style", "nav", "footer", "aside", "form"] NOISE_PATTERNS = [r"责任编辑.*", r"来源[:：].*", r"原标题[:：].*"] def extract_content(html): soup = BeautifulSoup(html, "html.parser") # 第一步：删掉整块的非正文标签 for tag in soup(NOISE_TAGS): tag.decompose() node = None for sel in CONTENT_SELECTORS: node = soup.select_one(sel) if node: break if node is None: node = soup.body or soup # 第二步：按段落取文本，保留换行 paragraphs = [p.get_text(strip=True) for p in node.find_all("p")] text = "\n".join(p for p in paragraphs if len(p) > 15) # 第三步：正则清掉残留的版式信息 for pat in NOISE_PATTERNS: text = re.sub(pat, "", text) return re.sub(r"\n{2,}", "\n", text).strip()

逻辑是「先粗删、再定位、后精修」。decompose()把脚本样式这些整块删除，能省掉后面大量误判；选择器列表按优先级尝试，是因为不同站点的正文容器类名不一样，硬编码一个选择器换个源就失效；段落过滤条件len(p) > 15很关键，短于 15 字的段落基本都是“上一篇”“分享到”这类噪音。

正则清洗放在最后一步，因为前两步之后残留的通常只有格式化的来源行和编辑署名，用正则收拾最省事。如果你抓的源比较多，建议把 NOISE_PATTERNS 做成可配置的列表，按域名挂不同的规则，别全塞在一个函数里。

3.3 三张核心表：新闻、用户、行为

推荐系统的数据模型不需要很复杂，三张表就能撑起完整闭环。表结构设计得好的话，后面加算法几乎不用改表。

-- 新闻表：存内容本身和离线算好的特征 CREATE TABLE news ( id INTEGER PRIMARY KEY AUTOINCREMENT, url_hash CHAR(32) NOT NULL UNIQUE, -- URL 的 md5，用于去重 title VARCHAR(255) NOT NULL, content TEXT, category VARCHAR(32), -- 频道/分类，排序打散时要用 keywords VARCHAR(255), -- 分词后 top 关键词，逗号分隔 vector BLOB, -- TF-IDF 向量，pickle 后存二进制 publish_at DATETIME, created_at DATETIME DEFAULT CURRENT_TIMESTAMP ); CREATE INDEX idx_news_publish ON news(publish_at DESC); -- 用户表：基础信息 + 画像缓存 CREATE TABLE users ( id INTEGER PRIMARY KEY AUTOINCREMENT, username VARCHAR(64) UNIQUE NOT NULL, profile TEXT, -- 用户画像 JSON：{关键词: 权重} updated_at DATETIME ); -- 行为表：曝光、点击、停留，推荐系统的燃料 CREATE TABLE behaviors ( id INTEGER PRIMARY KEY AUTOINCREMENT, user_id INTEGER NOT NULL, news_id INTEGER NOT NULL, action TINYINT NOT NULL, -- 1 曝光 2 点击 3 收藏 4 负反馈 duration INTEGER DEFAULT 0, -- 停留秒数 created_at DATETIME DEFAULT CURRENT_TIMESTAMP ); CREATE INDEX idx_behavior_user ON behaviors(user_id, created_at DESC); CREATE UNIQUE INDEX uk_behavior ON behaviors(user_id, news_id, action);

三张表各自承担一个角色。news 表里存 vector 这个二进制字段是刻意为之：在线排序时要现场算新闻和用户的相似度，如果每次从 content 重新分词再向量化，一次请求就要几百毫秒，存下来直接反序列化能压到个位数毫秒。

behaviors 表的action用整数编码而非字符串，是为了聚合查询快；uk_behavior这个唯一索引防止同一用户对同一新闻重复点击时插出多条记录，否则画像会被刷偏。duration字段很多同类项目会漏掉，但它是判断“点了但不喜欢”的唯一依据——点击只说明标题吸引人，停留三秒就退出说明内容不匹配。

注意：vector 字段用 BLOB 存 pickle 数据在单机版没问题，但如果要做多语言互通或者跨版本升级，建议改成 JSON 数组存浮点数，牺牲一点体积换兼容性。

3.4 增量抓取与定时调度：别每次全量重跑

全量重跑的问题不是慢，是会把已有数据覆盖掉，用户行为和新计算的推荐结果之间的关联会断。正确的做法是按 url_hash 做增量，已存在的直接跳过。

import hashlib from datetime import datetime, timedelta from apscheduler.schedulers.background import BackgroundScheduler def url_to_hash(url: str) -> str: return hashlib.md5(url.encode("utf-8")).hexdigest() def save_news(conn, item): """存在则跳过，不存在则插入；返回 True 表示新入库""" h = url_to_hash(item["url"]) cur = conn.execute("SELECT 1 FROM news WHERE url_hash = ?", (h,)) if cur.fetchone(): return False conn.execute( "INSERT INTO news(url_hash, title, content, category, publish_at) " "VALUES (?, ?, ?, ?, ?)", (h, item["title"], item["content"], item["category"], item["publish_at"]) ) return True def crawl_job(): # 只抓最近 24 小时的列表，翻页上限 3 页 for page in range(1, 4): items = parse_list(fetch(f"https://example.com/news?page={page}")) for it in items: if save_news(conn, it): print(f"新增: {it['title'][:20]}") sched = BackgroundScheduler(timezone="Asia/Shanghai") sched.add_job(crawl_job, "interval", minutes=30, id="crawl") sched.start()

save_news先查后插的写法在并发下理论上会有竞态，但单进程定时任务场景够用；如果要多进程跑，把插入改成INSERT OR IGNORE配合唯一索引，让数据库来兜底。minutes=30这个间隔是按新闻的更新节奏定的，一天几百条的源，半小时一轮完全够，抓太频反而增加被封风险。

调度器跟 Flask 放同一个进程里启动，好处是部署简单，坏处是 Flask 用多 worker 模式时每个 worker 都会起一份调度器，任务会重复执行。生产环境要么把调度器拆成独立进程，要么用文件锁保证只有一个实例在跑。这个坑在后面会细说。

4. 推荐引擎核心：从分词画像到召回排序的代码实现

4.1 中文分词与关键词抽取：jieba 的三种用法

分词这一步看着简单，实际上决定了后面特征空间的质量。项目里主要用 jieba 的三个能力：精确模式切词、自定义词典补领域词、TF-IDF 关键词抽取。

import jieba import jieba.analyse # 加载领域词典，避免"人工智能"被切成"人工/智能" jieba.load_userdict("data/userdict.txt") # 加载停用词，过滤"的、了、是"这类高频无信息词 STOPWORDS = set(open("data/stopwords.txt", encoding="utf-8").read().split()) def cut(text: str) -> list: """精确模式分词 + 停用词过滤 + 长度过滤""" words = jieba.lcut(text) return [w for w in words if w not in STOPWORDS and len(w) > 1 and not w.isdigit()] def extract_keywords(text: str, topk: int = 8) -> list: """基于 TF-IDF 的关键词抽取，返回 [(词, 权重), ...]""" return jieba.analyse.extract_tags(text, topK=topk, withWeight=True)

load_userdict必须放在所有分词调用之前，否则领域词表不生效。词典格式是「词 词频 词性」，词性可以省略。过滤条件里len(w) > 1排除单字词，not w.isdigit()排除纯数字——新闻里的时间、金额数字很多，留着会污染特征。

extract_tags返回的是带权重的词表，权重来自 jieba 内置的 IDF 语料。如果你的新闻领域比较垂直，内置 IDF 会偏，常见做法是拿自己的新闻库重新算一份 IDF 覆盖掉，jieba.analyse.set_idf_path()一行就能换。

这里有个反直觉的点：停用词表不是越长越好。我一开始把“今天”“记者”这类词也扔进去，后来发现某些新闻的核心信息就在这些词所在的句子里，过滤太狠反而丢了信号。合理做法是先跑一遍统计词频，把 Top 200 高频词里明显无信息量的挑出来，剩下的保留。

4.2 新闻向量化与用户画像：TF-IDF 加权

分词完是向量化。项目用 TF-IDF 把每篇新闻映射成向量，用户画像则是他点过的新闻向量的加权平均。

import numpy as np from sklearn.feature_extraction.text import TfidfVectorizer def build_tfidf(corpus: list): """corpus: 已分好词、用空格连接的文本列表""" vec = TfidfVectorizer( max_features=5000, # 词表上限，控制向量维度 min_df=2, # 至少出现在 2 篇文档里才算有效词 max_df=0.85, # 出现在 85% 以上文档的词直接丢弃 token_pattern=r"(?u)\S+", # 已分词，按空格切 ) matrix = vec.fit_transform(corpus) return vec, matrix def build_profile(clicked_ids, matrix, behaviors, decay=0.9): """按时间衰减给历史点击加权，越近的行为权重越高""" vecs, weights = [], [] for i, nid in enumerate(clicked_ids): idx = id_to_row[nid] # 点击得 1 分，停留超过 60 秒额外加 0.5 w = 1.0 + (0.5 if behaviors[nid]["duration"] > 60 else 0.0) w *= decay ** (len(clicked_ids) - i - 1) # 越早的行为衰减越多 vecs.append(matrix[idx].toarray()[0]) weights.append(w) profile = np.average(vecs, axis=0, weights=weights) return profile # 长度 = 5000 的稠密向量

max_features=5000是个平衡点。词表太大会让向量又长又稀，相似度计算慢且噪声多；太小则很多新闻被映射成几乎相同的向量。min_df=2过滤只出现一次的词，基本等于去掉了人名和生僻词；max_df=0.85过滤掉“新闻”“报道”这种到处都是、区分度为零的词。

画像构建里的decay=0.9是时间衰减系数。假设用户点了 10 条，最早的权重是 0.9^9 ≈ 0.387，最近的接近 1，这样画像会自然向近期兴趣漂移。这个参数调大（接近 1）画像更稳定，调小（0.7 左右）对兴趣变化更敏感。资讯场景我一般用 0.85～0.9，因为用户的长期偏好还是比单次点击更有参考价值。

4.3 协同过滤召回：item-based 相似度矩阵

召回阶段负责从几万条新闻里快速筛出几百条候选，协同过滤是这个项目的主召回。选 item-based 而不是 user-based，是因为新闻的数量远小于用户数量，物品相似度矩阵更小也更新更频繁。

import numpy as np from sklearn.metrics.pairwise import cosine_similarity def build_item_sim(behavior_matrix, topk=50): """ behavior_matrix: shape=(n_news, n_users) 的稀疏矩阵 返回每篇新闻最相似的 topk 篇，用 dict 存：{news_id: [(sim_id, score), ...]} """ sim = cosine_similarity(behavior_matrix, dense_output=False) item_sim = {} for i in range(sim.shape[0]): row = sim.getrow(i).toarray().ravel() row[i] = 0 # 自己和自己的相似度置零 top_idx = np.argpartition(row, -topk)[-topk:] top_idx = top_idx[np.argsort(row[top_idx])[::-1]] item_sim[i] = [(int(j), float(row[j])) for j in top_idx if row[j] > 0.01] return item_sim def recall_by_item(user_clicked, item_sim, topn=200): """把用户点过的新闻的相似新闻聚合起来，按累加分数排序""" scores = {} for nid in user_clicked: for sim_id, sim_score in item_sim.get(nid, []): if sim_id in user_clicked: # 已经看过的不再推 continue scores[sim_id] = scores.get(sim_id, 0.0) + sim_score return sorted(scores.items(), key=lambda x: -x[1])[:topn]

cosine_similarity用dense_output=False保持稀疏输出很重要，新闻上万条时稠密矩阵会直接吃光内存。np.argpartition取 Top-K 比全排序快一个量级，它只保证前 K 个是最大的，不保证内部有序，所以后面要再argsort一次。

row[i] = 0这行别漏。自己和自己相似度恒为 1，如果不清零，召回结果里会出现用户刚看过的那篇，体验很差。sim_score > 0.01的阈值过滤掉弱关联，避免凑数。recall_by_item里累加分数而不是取最大，是因为用户看过三篇科技新闻，那篇跟三篇都相似的新闻显然比只跟一篇相似的相关。

item_sim 这个字典我一般会序列化到磁盘（pickle 或者 json），在线服务直接加载，避免每次启动都重算。更新频率按行为数据的增长速度定，新闻站一天更新一次足够。

4.4 混合排序与冷启动兜底

召回给的是候选集，排序要把内容相似度、协同过滤分数、时效性揉成一个最终分数。

import math from datetime import datetime def rank(candidates, profile, news_matrix, now=None): """candidates: [(news_id, cf_score), ...]""" now = now or datetime.now() scored = [] for nid, cf in candidates: # 内容相似度：用户画像和新闻向量的余弦 content_sim = float(np.dot(profile, news_matrix[id_to_row[nid]].toarray()[0])) # 时效性：24 小时内线性衰减到 0.3，超过 72 小时归零 hours = (now - news_publish[nid]).total_seconds() / 3600 freshness = max(0.0, 1.0 - hours / 72) # 三路加权 score = 0.5 * content_sim + 0.35 * cf + 0.15 * freshness scored.append((nid, score)) return sorted(scored, key=lambda x: -x[1]) def diversify(ranked, topn=10, max_per_category=3): """按分类打散，防止首页全是同一类新闻""" out, cat_count = [], {} for nid, score in ranked: cat = news_category[nid] if cat_count.get(cat, 0) >= max_per_category: continue out.append((nid, score)) cat_count[cat] = cat_count.get(cat, 0) + 1 if len(out) >= topn: break return out

三个权重 0.5 / 0.35 / 0.15 是我调试出来的经验值，含义是内容匹配占主导，协同过滤做补充，时效性只做微调。你可以根据自己数据的特点调，比如做财经新闻，时效性权重可以提到 0.3。

freshness用 72 小时归零而不用平台常见的 24 小时，是给非实时新闻留空间——深度报道的价值周期更长。这个参数跟业务强相关，别照抄。

diversify解决的是信息茧房问题。纯按分数排，一个爱看体育的用户首页会全是体育，三天就看腻了。每个分类最多 3 条的硬约束简单粗暴但有效。更讲究的做法是 MMR（最大边际相关性），在相关性和多样性之间做权衡，代码要多二十行，效果提升有限，视需求选。

4.5 离线评估：precision@K 和覆盖率

改完算法得知道效果变好还是变差，不能只靠首页看着顺眼。项目里带了一个简单的离线评估脚本。

def precision_at_k(recommend_fn, test_users, k=10): """用留一法：把用户最后一个点击藏起来，看他能不能被推荐回来""" hit, total = 0, 0 for uid, (history, target) in test_users.items(): recs = [nid for nid, _ in recommend_fn(uid, history)[:k]] hit += 1 if target in recs else 0 total += 1 return hit / total if total else 0.0

留一法适合行为数据稀疏的场景，实现简单，解释性也强。但要注意用它评估协同过滤会有偏乐观的问题，因为物品相似度矩阵里可能已经包含了这个目标点击的信息。严谨一点的做法是按时间切分，用前 80% 的行为训练、后 20% 测试，代码量不大，值得改。

除了准确率，覆盖率一定要看。如果推荐系统只推头部那 5% 的新闻，precision 可能很好看，但长尾内容完全曝光不到，编辑那边也会有意见。计算方式是推荐结果里出现过的不同新闻数除以新闻总数。

5. 服务层实现与问题排查：接口、缓存和五个高频坑

5.1 三个核心路由：推荐、详情、反馈

在线层就三个接口，接口设计遵循一个原则：前端不需要知道推荐是怎么算的，只拿列表。

from flask import Blueprint, request, jsonify from web.cache import get_recommend_cache, set_recommend_cache bp = Blueprint("api", __name__, url_prefix="/api") @bp.route("/recommend") def recommend(): uid = request.args.get("user_id", type=int) size = min(request.args.get("size", 10, type=int), 50) # 硬上限防刷 if not uid: return jsonify({"code": 400, "msg": "missing user_id"}), 400 cached = get_recommend_cache(uid) if cached: return jsonify({"code": 0, "data": cached[:size], "source": "cache"}) # 缓存未命中：现场召回 + 排序，然后回填 recs = pipeline.recommend_for_user(uid, topn=size) set_recommend_cache(uid, recs, ttl=600) return jsonify({"code": 0, "data": recs, "source": "online"}) @bp.route("/feedback", methods=["POST"]) def feedback(): body = request.get_json(silent=True) or {} uid, nid = body.get("user_id"), body.get("news_id") action = body.get("action") # 2 点击 3 收藏 4 负反馈 if not all([uid, nid, action]): return jsonify({"code": 400, "msg": "bad params"}), 400 db.insert_behavior(uid, nid, action, body.get("duration", 0)) # 行为变化，缓存立即失效 invalidate_recommend_cache(uid) return jsonify({"code": 0})

size做了min(..., 50)的上限，防止有人传个 10000 把数据库拖垮。source字段返回cache还是online，是排查性能问题最有用的一行——如果线上日志里 online 占比过高，说明缓存命中率有问题，或者 TTL 设太短。

feedback 接口里缓存失效这步很关键。如果不失效，用户点了收藏，推荐列表还是十分钟前的，会觉得系统没反应。代价是频繁点击会让缓存不断重建，所以实际做法一般是延迟失效或者只对收藏、负反馈这类强信号做失效，普通点击不动缓存。

参数校验在接口层做，不要指望下层函数兜底。项目里统一返回{"code": 0, "data": ...}的结构，前端按 code 判断，比裸返回列表好维护。

5.2 缓存预热与降级

推荐列表是典型的读多写少，缓存是必选项。项目用 Redis 存每个用户的推荐列表，key 是rec:{user_id}。

import json import redis r = redis.Redis(host="localhost", port=6379, db=0, socket_timeout=0.1, socket_connect_timeout=0.1, decode_responses=True) def get_recommend_cache(uid): try: raw = r.get(f"rec:{uid}") return json.loads(raw) if raw else None except redis.RedisError: return None # 缓存挂了当作未命中，走在线计算 def set_recommend_cache(uid, recs, ttl=600): try: r.setex(f"rec:{uid}", ttl, json.dumps(recs, ensure_ascii=False)) except redis.RedisError: pass # 写缓存失败不影响主流程 def warm_up(active_users, pipeline): """离线任务结束后批量预热活跃用户的推荐列表""" pipe = r.pipeline() for uid in active_users: recs = pipeline.recommend_for_user(uid, topn=20) pipe.setex(f"rec:{uid}", 900, json.dumps(recs, ensure_ascii=False)) pipe.execute()

两个超时参数都设成 0.1 秒，意思是 Redis 稍微有点抖动就直接放弃，走在线计算。这比让请求卡住两秒强得多。所有 Redis 操作都包在 try 里，缓存不可用时系统降级但不崩溃——这是缓存使用的基本纪律，很多人图省事不写，Redis 一重启全站 500。

warm_up在离线任务跑完后调用，把活跃用户的推荐列表提前算好。这样用户第一次访问就是缓存命中，不用现场算。活跃用户的判定一般是最近 7 天有行为的用户，数量控制在万级以内。

TTL 设 600 秒是折中：太短缓存命中率低，太长用户兴趣变化跟不上。新闻场景里，用户连续刷两次的间隔通常大于 10 分钟，600 秒基本够。

5.3 五条踩坑记录：现象、原因和解决办法

第一条，推荐结果里出现用户刚看过的新闻。

现象是用户点了某篇，刷新首页它还在第一位。原因有两个叠加：item_sim 矩阵里没做对角线清零，以及召回结果没跟已点击列表做差集。解决办法是row[i] = 0和if sim_id in user_clicked: continue两处都要写，缺一个都堵不住。我一开始只写了后者，但缓存里存的是点击之前的推荐列表，所以还要在 feedback 接口里让缓存失效。

第二条，中文全部显示成乱码或者问号。

现象是抓下来的新闻存进数据库后标题变成æ–°é—»。原因是 requests 根据响应头里的 charset 解码，而很多站点声明的编码和实际不符。解决办法是统一用resp.apparent_encoding覆盖resp.encoding，并在建库时确认表和连接都是 utf8mb4。另外命令行跑脚本时要确认终端的编码，Windows 下的 cp936 终端打印 utf-8 字符串也会乱，这种情况改输出不改数据。

第三条，离线任务跑完，在线推荐没变化。

现象是手动执行了 offline 命令，日志显示生成成功，但前端刷新还是老结果。原因是推荐结果双写了两处：数据库表和 Redis 缓存，offline 只更新了前者。解决办法是把缓存失效逻辑挂在离线任务的末尾，批量delete相关 key，或者干脆给 key 加版本号，换版本等于全体失效。这个坑最典型的表现是“我明明改了权重”，排查半天发现改的东西根本没上线。

第四条，多 worker 部署后定时任务重复执行。

现象是日志里同一批新闻被处理了三四遍，抓取请求量翻倍。原因是 Flask 用 gunicorn 起多个 worker，每个进程都初始化了一次 APScheduler。解决办法有三个层次：最简单的用 PID 文件锁，进程启动时尝试获取文件锁，拿不到就不启动调度器；正规点的是把调度器拆成独立进程，用 supervisor 或 systemd 管理；再复杂就上分布式任务队列。单机演示项目用文件锁就够了，十几行代码。

第五条，用户量大了以后推荐接口越来越慢。

现象是初期几十毫秒，后来涨到一两秒。原因不是算法变复杂了，是 item_sim 矩阵和新闻向量的加载方式有问题——每次请求都从数据库读 news 表然后重新构建 id_to_row 映射。解决办法是把这些只读的中间结果在服务启动时一次性加载进内存，用全局变量持有；数据更新时通过一个版本号或者信号触发重新加载。另外news_matrix保持 scipy 稀疏矩阵格式，不要随手.toarray()转稠密，一万条新闻乘五千维就是 400MB。

注意：这五条里有三条跟缓存和中间状态有关。推荐系统出问题，第一反应应该去看“数据是不是旧了”，而不是去怀疑算法。

6. 二次开发：换召回策略与加反馈闭环的改造路径

拿到这套代码之后，最容易上手的改造是替换召回层。项目里 recall.py 暴露的是一个函数recall_for_user(uid, history, topn)，你只要保证输入输出格式不变，内部换成任何算法都行。比如接入向量召回：把新闻向量灌进 FAISS 建索引，用户画像作为查询向量，近似最近邻一查就是候选集，代码量不到五十行，召回速度和语义匹配效果都会好一截。我一般习惯先保留原来的 itemCF 作为兜底，用配置开关切换主召回，出问题能一键回退。

第二块值得改的是反馈闭环。原项目的 behaviors 表只记录点击，没有曝光日志，这意味着你没法算 CTR，也没法做负样本采样。改造方式是在 recommend 接口返回之前，把这次返回的每条新闻写入一条 action=1 的曝光记录，量大的话用异步队列批量写。有了曝光数据，排序模型才有训练的可能，哪怕先上一个逻辑回归，效果也比纯规则融合稳。

第三块是实验分流。想验证新算法是不是真的更好，最简单的做法是在用户维度做哈希分桶，bucket = hash(uid) % 100，0～49 走老策略，50～99 走新策略，推荐结果里带上 bucket 字段，前端埋点一起上报。跑一周对比两组的点击率和人均阅读数，比离线指标可信得多。

有一件事我踩过之后每次都做：改完任何算法，先跑一遍离线评估脚本，把 precision@K、覆盖率、平均推荐分类数三个数记下来，跟改动前对比。有一回我把时效性权重从 0.15 提到 0.35，主观感觉首页“新鲜多了”，结果离线指标掉了 8 个点，覆盖率也降了，因为时效性权重一大，长尾的旧新闻全被挤出候选。要不是先跑了评估，这个改动就直接上线了。

还有一个习惯：所有跟时间、衰减、权重相关的参数，全部写进 config.py 并加注释说明调大调小的效果，别硬编码在函数里。推荐系统的参数是要反复试的，散落在代码里，三个月后自己都找不到。

这份资源的价值不在算法多先进，而在于它把一条完整链路都摆出来了，每一层都能单独替换、单独验证。想跑通推荐系统的第一版，从它开始能省掉很多搭架子的时间；想在它上面做实验，分层结构也接得住。希望帮到你。

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