企业官网建设流程全解析

AI 题解推荐系统的召回与排序：怎样推荐"刚好适合你"的题目

一、推荐你不是"挑最难的"或"挑最简单的"

刷题推荐的第一直觉是：把用户没做过的、同标签的题目推过去。这个策略的问题是：推荐了一堆题，用户点开三道，做了一道，其余的都静静地躺在推荐列表里。

一个真正有效的推荐系统需要回答两个核心问题：

1. 召回（Recall）：从海量题库中，筛选出与用户相关的候选集。
2. 排序（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_factory=dict) # 各标签下的已做题目集合 solved_problems: dict[str, set[int]] = field(default_factory=dict) # 最近的错误标签（高频错误） recent_errors: dict[str, int] = field(default_factory=dict) # 当前刷题节奏（最近 7 天日均题数） daily_pace: float = 0.0 # 偏好难度分布 difficulty_preference: dict[str, float] = field(default_factory=dict) @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_factory=list) # 前置题目 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(key=lambda x: x[1], reverse=True) 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 推荐疲劳

如果连续推荐同类题目，用户会疲劳。需要在排序层加入多样性约束，相邻位置的题目尽量不共享相同的主标签。

五、总结

题解推荐系统的核心挑战不是在"找对题"——简单的标签匹配就能做到。真正的挑战是找到"刚好适合"的题——既不太难让你放弃，也不太简单让你没有收获。这需要综合考虑用户的掌握程度、知识的薄弱环节、题目的难度梯度，并在探索与利用之间取得平衡。