简介这份基于 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 技术栈对照每个环节为什么用这个库环节选型常见替代选择理由Web 框架FlaskDjango、FastAPI路由少、依赖轻改接口快适合演示型系统数据采集requests BeautifulSoupScrapy、Selenium页面结构简单Scrapy 对新手偏重中文分词jiebaHanLP、pkuseg装完即用支持自定义词典和关键词抽取向量化scikit-learn TfidfVectorizerWord2Vec、BERT无需训练冷启动阶段就能出效果召回item-based 协同过滤矩阵分解、双塔实现直观行为数据少时也能跑存储MySQL / SQLiteMongoDB、PostgreSQL结构固定课程设计用 SQLite 零配置缓存Redis本地字典推荐列表跨进程共享支持过期策略定时任务APSchedulercrontab、Celery跟 Flask 同进程部署简单这张表不是让你照抄而是让你知道每个位置的可替换空间。比如你的新闻量上了十万级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 协议和访问频率限制把并发数压到 12间隔加到 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;q0.9, }) # 重试策略只对连接错误和 5xx 重试4xx 直接放弃 retry Retry(total3, backoff_factor1.5, status_forcelist[500, 502, 503, 504]) session.mount(http://, HTTPAdapter(max_retriesretry)) session.mount(https://, HTTPAdapter(max_retriesretry)) def fetch(url, timeout10): 抓单页返回 HTML 字符串失败返回 None 而不是抛异常 try: resp session.get(url, timeouttimeout) 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几个参数值得说清楚。total3是重试次数配合backoff_factor1.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(stripTrue) 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(fhttps://example.com/news?page{page})) for it in items: if save_news(conn, it): print(f新增: {it[title][:20]}) sched BackgroundScheduler(timezoneAsia/Shanghai) sched.add_job(crawl_job, interval, minutes30, idcrawl) sched.start()save_news先查后插的写法在并发下理论上会有竞态但单进程定时任务场景够用如果要多进程跑把插入改成INSERT OR IGNORE配合唯一索引让数据库来兜底。minutes30这个间隔是按新闻的更新节奏定的一天几百条的源半小时一轮完全够抓太频反而增加被封风险。调度器跟 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, encodingutf-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, topKtopk, withWeightTrue)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_features5000, # 词表上限控制向量维度 min_df2, # 至少出现在 2 篇文档里才算有效词 max_df0.85, # 出现在 85% 以上文档的词直接丢弃 token_patternr(?u)\S, # 已分词按空格切 ) matrix vec.fit_transform(corpus) return vec, matrix def build_profile(clicked_ids, matrix, behaviors, decay0.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, axis0, weightsweights) return profile # 长度 5000 的稠密向量max_features5000是个平衡点。词表太大会让向量又长又稀相似度计算慢且噪声多太小则很多新闻被映射成几乎相同的向量。min_df2过滤只出现一次的词基本等于去掉了人名和生僻词max_df0.85过滤掉“新闻”“报道”这种到处都是、区分度为零的词。画像构建里的decay0.9是时间衰减系数。假设用户点了 10 条最早的权重是 0.9^9 ≈ 0.387最近的接近 1这样画像会自然向近期兴趣漂移。这个参数调大接近 1画像更稳定调小0.7 左右对兴趣变化更敏感。资讯场景我一般用 0.850.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, topk50): behavior_matrix: shape(n_news, n_users) 的稀疏矩阵 返回每篇新闻最相似的 topk 篇用 dict 存{news_id: [(sim_id, score), ...]} sim cosine_similarity(behavior_matrix, dense_outputFalse) 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, topn200): 把用户点过的新闻的相似新闻聚合起来按累加分数排序 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(), keylambda x: -x[1])[:topn]cosine_similarity用dense_outputFalse保持稀疏输出很重要新闻上万条时稠密矩阵会直接吃光内存。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, nowNone): 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, keylambda x: -x[1]) def diversify(ranked, topn10, max_per_category3): 按分类打散防止首页全是同一类新闻 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 离线评估precisionK 和覆盖率改完算法得知道效果变好还是变差不能只靠首页看着顺眼。项目里带了一个简单的离线评估脚本。def precision_at_k(recommend_fn, test_users, k10): 用留一法把用户最后一个点击藏起来看他能不能被推荐回来 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, typeint) size min(request.args.get(size, 10, typeint), 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, topnsize) set_recommend_cache(uid, recs, ttl600) return jsonify({code: 0, data: recs, source: online}) bp.route(/feedback, methods[POST]) def feedback(): body request.get_json(silentTrue) 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(hostlocalhost, port6379, db0, socket_timeout0.1, socket_connect_timeout0.1, decode_responsesTrue) def get_recommend_cache(uid): try: raw r.get(frec:{uid}) return json.loads(raw) if raw else None except redis.RedisError: return None # 缓存挂了当作未命中走在线计算 def set_recommend_cache(uid, recs, ttl600): try: r.setex(frec:{uid}, ttl, json.dumps(recs, ensure_asciiFalse)) except redis.RedisError: pass # 写缓存失败不影响主流程 def warm_up(active_users, pipeline): 离线任务结束后批量预热活跃用户的推荐列表 pipe r.pipeline() for uid in active_users: recs pipeline.recommend_for_user(uid, topn20) pipe.setex(frec:{uid}, 900, json.dumps(recs, ensure_asciiFalse)) 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 接口返回之前把这次返回的每条新闻写入一条 action1 的曝光记录量大的话用异步队列批量写。有了曝光数据排序模型才有训练的可能哪怕先上一个逻辑回归效果也比纯规则融合稳。第三块是实验分流。想验证新算法是不是真的更好最简单的做法是在用户维度做哈希分桶bucket hash(uid) % 100049 走老策略5099 走新策略推荐结果里带上 bucket 字段前端埋点一起上报。跑一周对比两组的点击率和人均阅读数比离线指标可信得多。改造点涉及文件改动量验证方式向量召回替换 itemCFrecommender/recall.py约 60 行离线 precision10 线上 CTR曝光日志与负样本web/routes.py、models/schema.sql约 40 行检查 behaviors 表 action1 占比用户维度 A/B 分流web/routes.py、config.py约 30 行分桶后两组指标对比分类打散改 MMRrecommender/rank.py约 25 行首页分类分布统计有一件事我踩过之后每次都做改完任何算法先跑一遍离线评估脚本把 precisionK、覆盖率、平均推荐分类数三个数记下来跟改动前对比。有一回我把时效性权重从 0.15 提到 0.35主观感觉首页“新鲜多了”结果离线指标掉了 8 个点覆盖率也降了因为时效性权重一大长尾的旧新闻全被挤出候选。要不是先跑了评估这个改动就直接上线了。还有一个习惯所有跟时间、衰减、权重相关的参数全部写进 config.py 并加注释说明调大调小的效果别硬编码在函数里。推荐系统的参数是要反复试的散落在代码里三个月后自己都找不到。这份资源的价值不在算法多先进而在于它把一条完整链路都摆出来了每一层都能单独替换、单独验证。想跑通推荐系统的第一版从它开始能省掉很多搭架子的时间想在它上面做实验分层结构也接得住。希望帮到你。本文还有配套的精品资源点击获取