AI 题解推荐系统的召回与排序怎样推荐刚好适合你的题目一、推荐你不是挑最难的或挑最简单的刷题推荐的第一直觉是把用户没做过的、同标签的题目推过去。这个策略的问题是推荐了一堆题用户点开三道做了一道其余的都静静地躺在推荐列表里。一个真正有效的推荐系统需要回答两个核心问题召回Recall从海量题库中筛选出与用户相关的候选集。排序Ranking在候选集中按最适合用户排序。二、推荐系统的整体架构flowchart TD A[用户画像] -- B[召回层] A -- C[排序层] B -- B1[标签协同过滤: 相似用户做了哪些题] B -- B2[知识图谱召回: 同一知识点的相关题] B -- B3[难度梯度召回: 逐步提升难度] B1 -- D[候选集合并去重] B2 -- D B3 -- D D -- C C -- C1[难度匹配度评分] C -- C2[知识薄弱度评分] C -- C3[新鲜度评分] C -- C4[学习效率评分] C1 -- E[排序结果 Top-K] C2 -- E C3 -- E C4 -- E三、实现from dataclasses import dataclass, field from typing import Optional from collections import defaultdict import math dataclass class UserProfile: 用户画像 user_id: str # 各标签的掌握程度 0.0-1.0 tag_mastery: dict[str, float] field(default_factorydict) # 各标签下的已做题目集合 solved_problems: dict[str, set[int]] field(default_factorydict) # 最近的错误标签高频错误 recent_errors: dict[str, int] field(default_factorydict) # 当前刷题节奏最近 7 天日均题数 daily_pace: float 0.0 # 偏好难度分布 difficulty_preference: dict[str, float] field(default_factorydict) dataclass class ProblemMeta: 题目元信息 problem_id: int title: str tags: list[str] difficulty: str # Easy / Medium / Hard acceptance_rate: float # 全站通过率 frequency_score: float # 高频题分数面试出现频率 avg_time_minutes: float # 平均解题时间 prerequisites: list[int] field(default_factorylist) # 前置题目 class ProblemRecommender: 题解推荐系统 召回 排序的两阶段架构。 # 各召回通道的权重 RECALL_WEIGHTS { tag_co_filter: 0.35, # 标签协同过滤 weakness_boost: 0.30, # 薄弱环节强化 freshness: 0.20, # 新题目探索 high_freq: 0.15, # 高频面试题 } def __init__(self, problem_pool: dict[int, ProblemMeta]): self.problems problem_pool def recommend( self, user: UserProfile, top_k: int 10 ) - list[tuple[ProblemMeta, float]]: 为用户推荐题目 # 第一阶段召回候选集 candidates self._recall(user) # 第二阶段精排打分 scored self._rank(user, candidates) # 按分数降序取 Top-K scored.sort(keylambda x: x[1], reverseTrue) return scored[:top_k] def _recall(self, user: UserProfile) - set[int]: 多路召回 candidates: dict[int, float] defaultdict(float) # 通道 1基于标签的协同过滤 # 找相同薄弱标签下相似用户做了但当前用户没做的题 for tag, mastery in user.tag_mastery.items(): if mastery 0.6: # 薄弱标签查询该标签下用户没做过的题 solved user.solved_problems.get(tag, set()) for pid, problem in self.problems.items(): if tag in problem.tags and pid not in solved: candidates[pid] self.RECALL_WEIGHTS[tag_co_filter] # 通道 2高频面试题按企业标签权重 for pid, problem in self.problems.items(): if problem.frequency_score 0.8: solved_any any( pid in s for s in user.solved_problems.values() ) if not solved_any: candidates[pid] self.RECALL_WEIGHTS[high_freq] # 通道 3错误标签的强化推荐 for tag, error_count in user.recent_errors.items(): for pid, problem in self.problems.items(): solved user.solved_problems.get(tag, set()) if tag in problem.tags and pid not in solved: # 错误越多权重越高 boost min(error_count * 0.1, 0.5) candidates[pid] self.RECALL_WEIGHTS[weakness_boost] * (1 boost) return set(candidates.keys()) def _rank( self, user: UserProfile, candidate_ids: set[int] ) - list[tuple[ProblemMeta, float]]: 精排对候选集打分 scored [] for pid in candidate_ids: problem self.problems.get(pid) if not problem: continue score self._calculate_score(user, problem) scored.append((problem, score)) return scored def _calculate_score( self, user: UserProfile, problem: ProblemMeta ) - float: 计算题目对用户的综合得分 评分因素 1. 难度匹配度核心 2. 知识薄弱度 3. 学习效率时间投入产出比 4. 前置知识满足度 # 1. 难度匹配度 diff_score self._difficulty_match(user, problem) # 2. 知识薄弱度 weakness_score self._weakness_boost(user, problem) # 3. 学习效率 efficiency_score self._efficiency_score(problem) # 4. 前置知识检查 prereq_score self._prerequisite_check(user, problem) # 加权综合 weights { difficulty: 0.35, weakness: 0.30, efficiency: 0.20, prerequisite: 0.15, } return ( diff_score * weights[difficulty] weakness_score * weights[weakness] efficiency_score * weights[efficiency] prereq_score * weights[prerequisite] ) def _difficulty_match(self, user: UserProfile, problem: ProblemMeta) - float: 难度匹配度推荐比当前水平稍难一点的题目 维果茨基的「最近发展区」理论在现有水平和潜在水平之间 的难度最适合学习。 # 找到用户在问题标签上的平均掌握度 tag_masteries [ user.tag_mastery.get(tag, 0.5) for tag in problem.tags ] avg_mastery sum(tag_masteries) / len(tag_masteries) if tag_masteries else 0.5 # 难度映射 diff_map {Easy: 0.3, Medium: 0.6, Hard: 0.9} # 最佳难度比当前水平高 0.2 左右 target_diff min(1.0, avg_mastery 0.2) actual_diff diff_map.get(problem.difficulty, 0.5) # 差距越小分数越高 gap abs(target_diff - actual_diff) return max(0.0, 1.0 - gap) def _weakness_boost(self, user: UserProfile, problem: ProblemMeta) - float: 薄弱环节加权 boost 0.0 for tag in problem.tags: if tag in user.recent_errors: boost min(user.recent_errors[tag] * 0.15, 0.6) return min(0.5, boost) 0.5 def _efficiency_score(self, problem: ProblemMeta) - float: 学习效率评分投入时间短、收获大的题目优先 # 通过率适中不要太高也不要太低的题目学习效率最高 optimal_rate 0.5 # 最佳通过率 rate_gap abs(problem.acceptance_rate - optimal_rate) return max(0.0, 1.0 - rate_gap * 2) def _prerequisite_check(self, user: UserProfile, problem: ProblemMeta) - float: 前置知识检查 if not problem.prerequisites: return 1.0 # 统计用户已完成的前置题目比例 all_solved set() for s in user.solved_problems.values(): all_solved.update(s) completed sum( 1 for p in problem.prerequisites if p in all_solved ) return completed / len(problem.prerequisites)四、边界与权衡4.1 探索与利用的平衡纯利用策略只推荐用户薄弱环节会导致知识面狭窄。需要保留一定比例的探索推荐新的标签、新的题型。通行的做法是 ε-greedy 策略90% 推荐最优10% 随机探索。4.2 冷启动新用户没有历史数据标签掌握度和错误记录都是空的。冷启动方案按知识点依赖图从最基础的标签开始推荐难度从 Easy 开始逐步提升。4.3 推荐疲劳如果连续推荐同类题目用户会疲劳。需要在排序层加入多样性约束相邻位置的题目尽量不共享相同的主标签。五、总结题解推荐系统的核心挑战不是在找对题——简单的标签匹配就能做到。真正的挑战是找到刚好适合的题——既不太难让你放弃也不太简单让你没有收获。这需要综合考虑用户的掌握程度、知识的薄弱环节、题目的难度梯度并在探索与利用之间取得平衡。