Django个性化文章推荐系统：从内容召回协同过滤到工程落地

简介基于Django框架开发的个性化文章推荐系统完整项目面向计算机相关专业在校学生、教师及企业开发者适用于毕业设计、课程设计或推荐系统入门进阶。系统综合运用数据库、机器学习、爬虫与网页交互技术结合经典推荐算法搭建基于网络热点的个性化文章推送平台可帮助读者掌握从数据采集、特征处理到推荐引擎构建的完整流程。资源包共407个文件以Python源码63个py、JavaScript脚本79个js、HTML页面21个html及CSS样式29个css为主另含SQL数据库文件、图片素材、配置文件与说明文档压缩包大小12.31MB目录结构清晰便于按模块学习与二次开发。目前已有209人学习下载代码经测试运行成功。项目既可直接作为毕设答辩的完整方案也可在此基础上扩展其他功能对推荐系统感兴趣的初学者与进阶者均有参考价值。1. 给 Django 项目加上个性化文章推荐的最短闭环把 Django 内容站的主页从时间倒序改成“猜你喜欢”最先出问题的不是算法而是数据太薄新用户只有几次点击协同过滤矩阵几乎全空公式再精确也排不出像样的结果。可行的做法是先把最小链路跑通再逐步加复杂度。这套基于 Django 的个性化文章推荐系统重点就在可交付的环环相扣文章画像、用户偏好、召回排序、源代码和配套文档。它解决的是中小型文章站最实际的诉求只依赖文章标签和用户行为在 Django 内部生成推荐列表不需要引入外部算法服务。适合正在做软件综合实践作品的学生也适合站点已经上线、想从“时间序”切换到“个性化排序”的工程师。默认你会建 Django 应用、能用 admin 和 ORM但不要求有机器学习背景。后文按工程主线推进先定推荐策略再落数据模型、推荐函数和接口最后把源代码与文档整理成一套可复现的交付物。2. 推荐策略选型基于内容、协同过滤还是混合召回在自建文章站里硬套公开推荐算法的常见误区是拿 MovieLens 的规模来要求自己的数据。MovieLens 有几十万评分文章站上线第一周可能只有几百条阅读点击。在这种场景下一开始就做协同过滤等价于用一个几乎全空的矩阵去算相似度结果没法看。因此第一版系统应该采用“基于内容召回为主协同过滤为辅”的策略。2.1 三种候选策略的适用场景策略需要的数据冷启动表现工程成本什么时候别用它基于内容文章标签、用户阅读记录较好低标签质量差时不建议基于用户的协同过滤大量用户行为差中行为日志数量不足时基于物品的协同过滤行为记录和物品相似度一般中文章更新快、矩阵频繁失效时我的习惯是先把基于内容的方案跑通再把基于用户的协同过滤作为补充分数叠上去。这样即使没有任何相似用户也能依靠标签相似度给出可解释的推荐。2.2 Jaccard 相似度以及它与余弦相似度的取舍文章推荐的第一版不需要 TF-IDF也不需要向量化。文章标签在落地时通常是字符串集合用 Jaccard 系数算集合重叠比例最直接两篇文章的标签交集越大它们就越相似。相比余弦相似度Jaccard 对“文章长度”和“高频词”不敏感代码也更简单。def jaccard_similarity(a: set, b: set) - float: # 空集合没有交集直接返回 0避免除零 if not a or not b: return 0.0 union a | b if not union: return 0.0 return len(a b) / len(union)这段代码的逻辑很直白a | b取并集a b取交集最终得到 0 到 1 之间的相似度。调用前要确认a和b是集合而不是列表否则运算符会报错从 Django ORM 拿到的标签列表需要先转换成set例如set(article.tags.values_list(name, flatTrue))。文章内容如果只有纯文本而没有有效标签这篇文章的标签集为空算出来的相似度永远是 0。所以数据准备阶段要把“标签为空”的文章先过滤掉或者给它们补一个默认分类否则这些文章永远不会出现在候选池里。2.3 混合推荐时的权重分配方法基于内容召回的问题在于只会推荐和过去很像的文章用户很快就会觉得重复。混合协同过滤可以缓解这个“信息茧房”问题但直接改模型结构并不划算。常见做法是打分后加权合并def hybrid_score(content_score: float, cf_score: float, alpha: float 0.7) - float: # alpha 控制内容分数权重 if cf_score 0: return content_score return alpha * content_score (1 - alpha) * cf_scorealpha是唯一需要调的参数。alpha0.7时内容分数占七成推荐结果偏保守适合新用户alpha0.5时协同过滤贡献一半活跃用户更容易发现新主题。注意cf_score可能为 0这种情况直接返回content_score保证候选列表不为空。3. Django 数据层文章标签、用户行为与画像存储推荐系统的数据建模不需要设计成数仓那样复杂。在 Django 里只需要三个核心模型文章、标签、用户点击行为。用户画像可以在每次推荐前实时计算文章量不大的时候完全没有必要落一张画像表。3.1 源码目录如何划分职责文件或目录核心职责不需要做的事情articles/models.py定义文章、标签、点击行为不放推荐算法recommender/engine.py画像、召回、排序、缓存不直接写 ORM 查询recommender/views.py接口参数解析、返回结果不塞业务逻辑config/settings.py注册应用、配置缓存不写死环境差异目录拆成这样后续调试时能很快定位问题推荐结果不对去engine.py数据没查到去models.py接口报错去views.py。3.2 三个核心模型如何支持个性化推荐先建一个articles应用和一个recommender应用模型主要在articles/models.py里。最基础的版本长这样# articles/models.py from django.conf import settings from django.db import models class Tag(models.Model): name models.CharField(标签名, max_length32, uniqueTrue) def __str__(self): return self.name class Article(models.Model): title models.CharField(标题, max_length255) content models.TextField(正文, default) tags models.ManyToManyField(Tag, blankTrue, related_namearticles) published_at models.DateTimeField(发布时间, auto_now_addTrue) class Meta: ordering [-published_at] class UserClick(models.Model): user models.ForeignKey(settings.AUTH_USER_MODEL, on_deletemodels.CASCADE) article models.ForeignKey(Article, on_deletemodels.CASCADE, related_nameclicks) action models.CharField(行为类型, max_length16, defaultread) created_at models.DateTimeField(发生时间, auto_now_addTrue) class Meta: indexes [ models.Index(fields[user, created_at]), ]逻辑说明Article.tags是多对多关系一个文章有多个标签标签也能挂在多个文章上UserClick记录用户行为action字段可以扩展成read、like、collect第一版只统计“读了”就够。on_deletemodels.CASCADE表示用户删除后行为一并清理避免脏数据。注意UserClick加了一条数据库索引Index(fields[user, created_at])因为推荐时总要先查某个用户最近的行为没有索引则数据量稍大就会慢。新用户清理测试数据时很多人会逐条delete()效率很低。只要行为数据不再需要直接执行UserClick.objects.filter(user_idtarget_user_id).delete()即可一条 SQL 就能清掉该用户的所有行为这也是 Django ORM 执行删除对象最直接的方式。3.3 用 admin 录入标签并优化查询标签质量直接影响推荐效果所以后台录入体验要跟上。注册 admin 时用filter_horizontal后台就能用双栏组件快速选择标签不用在长列表里一个个找# articles/admin.py from django.contrib import admin from .models import Article, Tag, UserClick admin.register(Article) class ArticleAdmin(admin.ModelAdmin): list_display (id, title, published_at) filter_horizontal (tags,) search_fields (title, tags__name) admin.register(Tag) class TagAdmin(admin.ModelAdmin): search_fields (name,) admin.register(UserClick) class UserClickAdmin(admin.ModelAdmin): list_display (user, article, action, created_at) list_filter (action,)输出推荐结果时最容易踩的坑是 N1 查询循环每篇文章去查 tags文章一多接口就慢。正确做法是配合prefetch_related一次取全from django.contrib import admin from .models import Article def all_articles_with_tags(): return Article.objects.prefetch_related(tags).only(id, title)prefetch_related(tags)会先把所有文章查出来再一次性查关联标签最后在 Python 内存里完成配对。4. 推荐引擎落地用户画像、相似度计算和结果 View推荐引擎我建议放在独立的recommender/engine.py中。不用 pandas也不用 NumPy用字典和Counter就能在文章量五万以内跑出可接受的结果。推荐的本质可以拆成两步给用户建画像再给候选文章打分排序。4.1 用字典组织标签画像从用户最近的阅读记录中统计标签出现次数就是最简单的用户画像。用户读过的文章中某个标签出现越频繁代表他对这个主题越感兴趣。代码可以把时间窗口当参数传进来避免一直算历史全量数据from collections import Counter from datetime import timedelta from django.utils import timezone from articles.models import UserClick def build_tag_index(): 返回 {文章ID: set(标签名)} 的映射供推荐函数复用。 index {} articles Article.objects.prefetch_related(tags).only(id) for article in articles: index[article.id] set(article.tags.values_list(name, flatTrue)) return index def build_user_profile(user_id, tag_index, recent_days30): 统计用户最近 recent_days 天读过的标签。 since timezone.now() - timedelta(daysrecent_days) clicks UserClick.objects.filter( user_iduser_id, actionread, created_at__gtesince, ).only(article_id) profile Counter() for click in clicks: for tag in tag_index.get(click.article_id, []): profile[tag] 1 return profile这段代码把文章和标签的关系收敛到了一个字典里Counter负责累加标签权重。recent_days30的意义是只取最近 30 天的点击太久远的行为会让用户画像越来越像历史画像跟不上口味变化。4.2 基于内容的候选打分函数有了用户画像之后遍历候选文章计算每篇文章的标签与画像的重合程度然后去掉已经读过的文章得到候选排序。这里每篇文章的标签数量不同用1 len(tags)做分母做一点轻微惩罚能避免超长标签文章因为标签多而占便宜def content_based_scores(user_id, tag_index, recent_days30, top_k50): profile build_user_profile(user_id, tag_index, recent_days) if not profile: return {} clicked_ids set( UserClick.objects.filter(user_iduser_id) .values_list(article_id, flatTrue) ) scores {} for article_id, tags in tag_index.items(): if article_id in clicked_ids: continue matched 0.0 for tag in tags: if tag in profile: matched profile[tag] / (1 len(tags)) if matched 0: scores[article_id] matched return dict(sorted(scores.items(), keylambda item: -item[1])[:top_k])top_k控制召回数量不需要对全站文章排序先取分数最高的 50 篇进入下一轮减少后续计算量。这个分数目前只反映“用户和文章像不像”不反映文章新旧。4.3 叠加协同过滤分数协同过滤这里选择基于用户的策略找到与当前用户点击记录重叠最多的用户把他们已读而当前用户没读的文章作为补充候选。用 Jaccard 相似度计算用户相似性代码量很小from collections import defaultdict def similar_user_scores(user_id, tag_index, top_k10): target_profile build_user_profile(user_id, tag_index) if not target_profile: return {} rows UserClick.objects.filter(actionread).values(user_id, article_id) user_articles defaultdict(set) for row in rows: user_articles[row[user_id]].add(row[article_id]) my_articles user_articles.get(user_id, set()) neighbor_scores [] for other_id, articles in user_articles.items(): if other_id user_id: continue union my_articles | articles if not union: continue score len(my_articles articles) / len(union) neighbor_scores.append((other_id, score)) result defaultdict(float) for other_id, _ in sorted(neighbor_scores, keylambda x: -x[1])[:top_k]: for aid in user_articles[other_id]: if aid not in my_articles: result[aid] 1.0 return dict(result)用户相似度分数只取前 10 个“邻居”降低噪声。计算出的result是其他用户推荐文章的加分项每被一个相似用户读过就加 1.0。当用户行为很少时这个函数返回空字典不会影响基于内容的主流程。最终在引擎入口合并两部分分数并加上时间衰减。发布时间越久的文章即使标签匹配也应该让位给新内容衰减系数用0.5 ** (days / half_life)可以理解成 30 天后权重减半import math from datetime import datetime def time_decay(created_at, now, half_life_days30): days max((now - created_at).days, 0) return 0.5 ** (days / half_life_days) def engine_recommend(user_id, top_k10, alpha0.7): tag_index build_tag_index() content content_based_scores(user_id, tag_index) cf similar_user_scores(user_id, tag_index) if alpha 1.0 else {} merged defaultdict(float) for article_id, score in content.items(): merged[article_id] alpha * score for article_id, score in cf.items(): merged[article_id] (1 - alpha) * score now datetime.now() ranked [] for article_id, score in merged.items(): article Article.objects.get(pkarticle_id) ranked.append((article_id, score * time_decay(article.published_at, now))) ranked.sort(keylambda x: -x[1]) return [article_id for article_id, _ in ranked[:top_k]]engine_recommend里先调用内容召回再按alpha决定是否补充协同过滤。注意这一段真实项目中不能用Article.objects.get(pk...)逐条取文章应该提前把Article对象放进字典代码里保留最直白的写法是为了先跑通链路。4.4 推荐结果的 API 视图推荐接口用 JsonResponse 返回方便前端和后台共同调用。参数top_k和alpha都从 URL 查询参数里读方便调参时不改代码# recommender/views.py from django.http import JsonResponse from articles.models import Article, UserClick from .engine import engine_recommend def recommend_api(request): if not request.user.is_authenticated: return JsonResponse({code: 401, msg: 请先登录}, status401) try: top_k min(int(request.GET.get(top_k, 10)), 50) alpha float(request.GET.get(alpha, 0.7)) except ValueError: return JsonResponse({code: 400, msg: 参数格式不对}, status400) rec_ids engine_recommend(request.user.id, top_ktop_k, alphaalpha) items [] for aid in rec_ids: article Article.objects.only(id, title, published_at).get(pkaid) items.append({ id: article.id, title: article.title, published_at: article.published_at.isoformat(), }) return JsonResponse({code: 0, items: items})min(int(...), 50)避免客户端传入超大top_k拖垮接口alpha的范围建议手动限制在 0.5 到 1.0 之间小于 0.5 时内容召回权重过低新用户推荐结果容易乱。实测阶段可以先用 admin 登录然后访问/api/recommend/?top_k10alpha0.7观察返回结果是否合理。调参经验可以简单总结成一张表用户状态建议 alpha原因点击少于 10 次0.9协同过滤无数据尽量靠标签匹配点击 10 到 50 次0.75内容为主协同为参考活跃用户0.5 到 0.6引入更多探索性推荐5. 源码交付时最关键的技巧离线验证系统的源代码和文档说明不仅要“能跑”还要让另一个人按文档十分钟内复现。这里最重要的是在交付前用历史数据离线验证推荐参数而不是直接上生产看反馈。5.1 源码包里必须有什么源码包的最简目录结构应该一眼能看懂article_recommend/ ├── config/ ├── articles/ ├── recommender/ ├── data/ │ └── sample_articles.json ├── scripts/ │ └── import_sample_data.py ├── docs/ │ └── 设计说明.md ├── requirements.txt └── README.mddata目录放手动构造的样例文章避免新环境里没有数据无法验证scripts专门放导入脚本和评估脚本README.md里固定写清楚 Django 版本锁定方式、启动命令、数据导入步骤、推荐接口的访问地址。建应用时用的python3 manage.py startapp命令在 README 里也要体现方便对照。5.2 用时间切分法验证命中率最常用的离线指标是hit_ratio10把用户行为按时间切分成前 80% 和后 20%用前 80% 训练画像用后 20% 里的文章去验证推荐列表是否命中。# scripts/evaluate.py from collections import defaultdict def evaluate_hit_ratio(history, recommender, top_k10): split_point int(len(history) * 0.8) train history[:split_point] test history[split_point:] train_by_user defaultdict(list) for row in train: train_by_user[row[user_id]].append(row[article_id]) test_by_user defaultdict(set) for row in test: test_by_user[row[user_id]].add(row[article_id]) total 0 hit 0 for user_id, seen in test_by_user.items(): recs recommender(user_id, top_ktop_k) hit len(set(recs) seen) total len(seen) return hit / total if total else 0.0评估脚本调用recommender(user_id, top_ktop_k)而不是直接调用某个函数就是为了能方便地替换不同算法版本做对比。跑完一次后把alpha从 0.9 调到 0.5再跑一次观察命中率如何变化比在黑盒里猜参数可靠得多。最终交付前检查这几个细节样例数据导入后每篇文章是否有标签新用户首次访问接口是否返回最新文章兜底把alpha0.7时的推荐列表和alpha0.5时的列表并排对比确认结果有明显差异。这样复制或评审项目的人不仅能复现结果也知道该动哪个参数、在哪里看效果。本文还有配套的精品资源点击获取