当前位置：首页 > news >正文

个性化题目推荐引擎：协同过滤与知识图谱的混合推荐策略

news 2026/7/15 22:57:58

个性化题目推荐引擎：协同过滤与知识图谱的混合推荐策略

一、用户刷题卡壳了，推荐什么

一个算法初学者刚刚刷过"两数之和"，但在"三数之和"上卡了 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(key=lambda x: x[1], reverse=True) 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(), key=lambda x: x[1], reverse=True ) 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(), key=lambda x: x[1], reverse=True ) 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, f"Problem {i}", difficulty=0.1 + i * 0.04, knowledge_points=[f"kp_{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_id=0) print(f"推荐题目: {recs}")