个性化题目推荐引擎协同过滤与知识图谱的混合推荐策略一、用户刷题卡壳了推荐什么一个算法初学者刚刚刷过两数之和但在三数之和上卡了 30 分钟。此时系统应该建议他先去刷一道双指针的基础题还是直接给他另一道哈希表的题目推荐引擎的核心是在正确的时间推荐正确的题目。这需要融合三种信息用户的能力画像已刷题、正确率、用时、题目的知识图谱关系前置知识点、相似题型、难度梯度、其他用户的协同行为相似用户还刷了什么。flowchart TB A[用户能力画像] -- D[混合推荐引擎] B[题目知识图谱] -- D C[协同过滤矩阵] -- D D -- E[协同过滤分数] D -- F[知识图谱分数] D -- G[难度适配分数] E -- H[加权融合层] F -- H G -- H H -- I[候选题目排序] I -- J[去重 多样性过滤] J -- K[Top-N 推荐列表] style D fill:#ccf style H fill:#ffc二、混合推荐的三层机制协同过滤层基于用户-题目的交互矩阵刷题次数/正确率找到相似用户喜欢的题目。核心公式是余弦相似度similarity(u, v) (R_u · R_v) / (|R_u| * |R_v|)其中 R_u 是用户 u 的题目评分向量。知识图谱层基于知识点的前置关系和难度梯度推荐跳一跳够得着的题目。图结构中的路径权重由知识点关联强度决定。难度适配层根据用户当前的能力水平正确率和平均用时筛选适当难度的题目。难度阈值 当前能力 0.1稍微挑战。三、混合推荐引擎的完整实现 个性化题目推荐引擎 三层融合协同过滤 知识图谱 难度适配 import math import numpy as np from typing import List, Dict, Tuple, Optional from dataclasses import dataclass from collections import defaultdict dataclass class Problem: 题目定义 problem_id: int title: str difficulty: float # 0-1 难度值 knowledge_points: List[str] # 知识点列表 prerequisites: List[int] # 前置题目 ID class CollaborativeFilter: 协同过滤推荐 def __init__(self): self.user_ratings: Dict[int, Dict[int, float]] {} def add_rating(self, user_id: int, problem_id: int, rating: float): 添加用户对题目的评分 if user_id not in self.user_ratings: self.user_ratings[user_id] {} self.user_ratings[user_id][problem_id] rating def find_similar_users(self, target_user: int, top_k: int 10) - List[Tuple[int, float]]: 找到最相似的 K 个用户 if target_user not in self.user_ratings: return [] target_vec self.user_ratings[target_user] similarities [] for uid, ratings in self.user_ratings.items(): if uid target_user: continue # 余弦相似度 common set(target_vec.keys()) set(ratings.keys()) if len(common) 3: continue dot sum(target_vec[p] * ratings[p] for p in common) norm_a math.sqrt(sum(v ** 2 for v in target_vec.values())) norm_b math.sqrt(sum(v ** 2 for v in ratings.values())) sim dot / (norm_a * norm_b) if norm_a and norm_b else 0 similarities.append((uid, sim)) similarities.sort(keylambda x: x[1], reverseTrue) return similarities[:top_k] def predict_rating(self, user_id: int, problem_id: int) - float: 预测用户对题目的评分 similar_users self.find_similar_users(user_id) if not similar_users: return 0.0 weighted_sum 0.0 sim_sum 0.0 for uid, sim in similar_users: if problem_id in self.user_ratings.get(uid, {}): weighted_sum sim * self.user_ratings[uid][problem_id] sim_sum abs(sim) return weighted_sum / sim_sum if sim_sum 0 else 0.0 class KnowledgeGraphRecommender: 基于知识图谱的推荐 def __init__(self): self.problems: Dict[int, Problem] {} self.kg_edges: Dict[str, float] {} # kp_a→kp_b: weight def add_problem(self, problem: Problem): 添加题目到知识图谱 self.problems[problem.problem_id] problem def add_kg_edge(self, kp_a: str, kp_b: str, weight: float 1.0): 添加知识点之间的关联边 self.kg_edges[f{kp_a}→{kp_b}] weight def get_related_problems(self, problem_id: int, max_distance: int 2, top_k: int 10) - List[Tuple[int, float]]: 获取与题目相关的其他题目 if problem_id not in self.problems: return [] src_kps set(self.problems[problem_id].knowledge_points) scores defaultdict(float) for pid, problem in self.problems.items(): if pid problem_id: continue dst_kps set(problem.knowledge_points) # 计算知识点重叠与距离 overlap len(src_kps dst_kps) if overlap 0: scores[pid] overlap * 0.5 ( 1.0 / (abs(problem.difficulty - self.problems[problem_id].difficulty) 0.1) ) * 0.3 sorted_scores sorted( scores.items(), keylambda x: x[1], reverseTrue ) return sorted_scores[:top_k] class DifficultyAdapter: 难度适配器 def __init__(self, target_success_rate: float 0.7): self.target_success_rate target_success_rate self.user_skill: Dict[int, float] {} def update_skill(self, user_id: int, correct: bool, problem_difficulty: float, learning_rate: float 0.1): 更新用户能力估计基于 Elo 评分思想 if user_id not in self.user_skill: self.user_skill[user_id] 0.3 # 初始能力 expected 1.0 / (1.0 math.exp( -(self.user_skill[user_id] - problem_difficulty) * 5 )) actual 1.0 if correct else 0.0 self.user_skill[user_id] learning_rate * (actual - expected) def score(self, user_id: int, problem_difficulty: float) - float: 根据难度适配程度打分越接近最近发展区越高 skill self.user_skill.get(user_id, 0.3) # 最优难度 当前能力 0.1稍微挑战 optimal skill 0.1 diff abs(problem_difficulty - optimal) return math.exp(-diff * 5) # 高斯核 class HybridRecommender: 混合推荐引擎 def __init__(self): self.cf CollaborativeFilter() self.kg KnowledgeGraphRecommender() self.da DifficultyAdapter() # 三路权重 self.weights {cf: 0.4, kg: 0.35, da: 0.25} def recommend(self, user_id: int, current_problem_id: Optional[int] None, top_n: int 5) - List[int]: 综合推荐 all_scores defaultdict(float) # 1. 协同过滤分数 for pid in list(self.kg.problems.keys())[:20]: cf_score self.cf.predict_rating(user_id, pid) all_scores[pid] cf_score * self.weights[cf] # 2. 知识图谱分数 if current_problem_id: kg_recs self.kg.get_related_problems(current_problem_id) for pid, kg_score in kg_recs: all_scores[pid] kg_score * self.weights[kg] # 3. 难度适配分数 for pid in list(self.kg.problems.keys())[:20]: diff self.kg.problems.get(pid) if diff: da_score self.da.score(user_id, diff.difficulty) all_scores[pid] da_score * self.weights[da] # 排序去重 sorted_items sorted( all_scores.items(), keylambda x: x[1], reverseTrue ) return [pid for pid, _ in sorted_items[:top_n]] if __name__ __main__: engine HybridRecommender() for i in range(20): engine.kg.add_problem(Problem( i, fProblem {i}, difficulty0.1 i * 0.04, knowledge_points[fkp_{i % 5}], prerequisites[max(0, i - 1)], )) engine.cf.add_rating(1, 0, 1.0) engine.cf.add_rating(1, 1, 0.8) engine.cf.add_rating(2, 0, 0.9) engine.cf.add_rating(2, 2, 1.0) recs engine.recommend(1, current_problem_id0) print(f推荐题目: {recs})四、冷启动与多样性问题新用户冷启动新用户没有任何行为数据时协同过滤失效。解决初始用知识图谱推荐按知识点拓扑排序随着交互增多逐渐增加协同过滤权重。内容多样性如果只推荐分数最高的题目可能连续推荐 5 道双指针题。加入 MMR最大边缘相关性算法平衡相关性和多样性。探索-利用平衡总是推荐最可能喜欢的题目会导致信息茧房。定期以 10-15% 的概率随机推荐探索性题目帮助用户发现新的知识点。五、总结协同过滤负责你也可能喜欢冷启动弱但长期效果好。知识图谱负责学习路径确保推荐遵循知识点前置关系。难度适配负责跳一跳够得着70% 成功率的目标对学习最有利。混合策略权重可动态调整初期知识图谱权重高中后期协同过滤权重高。本文实现了融合协同过滤、知识图谱和难度适配的三层混合推荐引擎核心的评分融合逻辑和动态权重机制可直接用于个性化学习推荐。