企业培训场景下的个性化课程推荐系统设计与实现：混合推荐架构的工程实践

一、问题拆解：企业培训推荐为什么比电商推荐更难

电商推荐和企业培训推荐看起来都是"给用户推内容"，但底层逻辑有几个关键差异，直接决定了推荐系统的设计方向。

差异一：数据稀疏到极致。 电商平台一个活跃用户一年可能产生几百条行为数据（浏览、加购、下单），但企业培训平台上，一个员工一年可能只学 3-5 门课。5000 人的公司，800 门课，学习记录矩阵的稀疏度超过 99%。传统的协同过滤在这种稀疏度下几乎失效。

差异二：冷启动不是边缘问题，是常态。 企业人员流动频繁，每个月都有新员工入职。新员工在培训平台上没有任何历史数据，但入职第一周就需要完成必修课和岗位基础课。如果推荐系统对新员工束手无策，那它覆盖了 20% 的用户空白。

差异三：推荐目标不是"点击率"，而是"能力补齐"。 电商推荐的目标是让用户多买东西，培训推荐的目标是帮员工补齐岗位能力缺口。这意味着推荐系统不能只推用户"喜欢"的课，还要推用户"需要"的课。一个做后端开发的员工可能对"项目管理"感兴趣，但他当前最紧迫的需求是补齐"分布式系统"的知识短板。

差异四：信息茧房的危害更大。 电商推荐陷入信息茧房，用户少买几样东西而已。培训推荐陷入信息茧房，员工的能力结构会变得畸形——只精通一个方向，缺乏跨领域视野。企业需要的是 T 型人才，推荐系统必须有意地引入跨领域内容。

理解了这四个差异，才能设计出真正适合企业培训场景的推荐系统。

二、整体架构：三路召回 + 交叉注意力排序

整个推荐系统采用经典的"召回 → 粗排 → 精排 → 重排"四阶段架构，但每个阶段都针对企业培训场景做了定制。

┌──────────────────────────────────────────────────────────────┐
│                       推荐请求入口                            │
│              (userId, 当前页面上下文, 设备信息)                 │
└──────────────────────────┬───────────────────────────────────┘
                           │
         ┌─────────────────┼─────────────────┐
         │                 │                 │
   ┌─────▼─────┐    ┌─────▼─────┐    ┌─────▼─────┐
   │ 协同过滤   │    │ 内容特征   │    │ 知识图谱   │
   │ 召回路     │    │ 召回路     │    │ 召回路     │
   │           │    │           │    │           │
   │ User-CF   │    │ 课程Embedding│  │ 岗位能力   │
   │ + Item-CF │    │ 相似度匹配  │    │ 缺口分析   │
   └─────┬─────┘    └─────┬─────┘    └─────┬─────┘
         │                 │                 │
         └─────────────────┼─────────────────┘
                           │ 合并去重（约 200-500 条候选）
                    ┌──────▼──────┐
                    │   粗排层     │
                    │  轻量模型    │
                    │  (双塔)     │
                    └──────┬──────┘
                           │ 保留 Top 50
                    ┌──────▼──────┐
                    │   精排层     │
                    │  交叉注意力  │
                    │  排序模型    │
                    └──────┬──────┘
                           │ 保留 Top 20
                    ┌──────▼──────┐
                    │   重排层     │
                    │  多样性控制  │
                    │  茧房打破   │
                    │  去重/过滤   │
                    └──────┬──────┘
                           │ 输出 Top 10
                    ┌──────▼──────┐
                    │   推荐结果   │
                    └─────────────┘

四阶段各自的职责：

召回层负责从 800 门课中快速筛选出 200-500 门"值得考虑"的候选课程。三路召回各有所长：协同过滤捕捉"和你学习偏好相似的人也在学"，内容特征捕捉"和你已学课程内容相似的课程"，知识图谱捕捉"你的岗位能力模型要求你但你还不会的"。

粗排层用轻量级双塔模型对候选课程做初步打分，从 500 条压缩到 50 条。双塔模型的计算量小（用户塔和课程塔可以分别预计算），适合处理中等规模的候选集。

精排层用交叉注意力模型对用户特征和课程特征做深度交互，产出更精准的排序分数。这一层的模型更重，但只需要处理 50 条候选，计算量可控。

重排层不关心"课程好不好"，只关心"这组推荐结果的整体质量"——是否足够多样、是否有跨领域内容、是否包含了用户还没看过的类型。这一层是打破信息茧房的关键。

三、数据基础：三类特征工程

推荐系统的上限由特征工程决定，模型只是逼近这个上限的工具。企业培训场景下，我们需要三类特征。

3.1 用户特征

from dataclasses import dataclass, field
from typing import Optional
import numpy as np


@dataclass
class UserFeatures:
    """用户特征向量"""
    user_id: str

    # 基础属性（冷启动时主要依赖这些）
    department: str                    # 部门编码
    job_role: str                      # 岗位编码（如 "backend_dev", "product_mgr"）
    job_level: str                     # 职级（如 "P5", "P6", "P7"）
    years_of_experience: float         # 工作年限
    education: str                     # 学历

    # 行为特征（有历史数据时可用）
    completed_courses: list[str] = field(default_factory=list)   # 已完课的课程 ID
    in_progress_courses: list[str] = field(default_factory=list) # 正在学的课程 ID
    learning_hours_total: float = 0.0  # 累计学习时长
    learning_hours_weekly: float = 0.0 # 近 4 周周均学习时长
    avg_completion_rate: float = 0.0   # 平均完课率

    # 偏好特征（从历史行为中提取）
    category_preference: dict[str, float] = field(default_factory=dict)
    # 例如 {"programming": 0.35, "management": 0.20, "communication": 0.10, ...}

    difficulty_preference: float = 0.5  # 偏好难度（0=入门, 1=高级）

    # 向量化表示（由特征编码器生成）
    embedding: Optional[np.ndarray] = None

3.2 课程特征

@dataclass
class CourseFeatures:
    """课程特征向量"""
    course_id: str

    # 基础属性
    title: str
    category: str                      # 一级分类（如 "技术", "管理", "通用素质"）
    sub_category: str                  # 二级分类（如 "后端开发", "前端开发"）
    tags: list[str]                    # 标签（如 ["Python", "Django", "REST API"]）
    difficulty_level: int              # 难度等级（1-5）
    duration_hours: float              # 课程时长
    language: str                      # 语言

    # 内容特征
    description_embedding: np.ndarray  # 课程描述的文本 Embedding（预计算）
    transcript_embedding: np.ndarray   # 课程全文转录的文本 Embedding（预计算）

    # 知识图谱关联
    skill_ids: list[str]               # 关联的技能点 ID（来自知识图谱）
    prerequisite_course_ids: list[str] # 前置课程 ID

    # 统计特征
    avg_rating: float = 0.0            # 平均评分
    rating_count: int = 0              # 评分人数
    completion_count: int = 0          # 完课人数
    completion_rate: float = 0.0       # 完课率（开始学习的人中，完课的比例）

    # 向量化表示
    embedding: Optional[np.ndarray] = None

3.3 知识图谱特征：岗位-技能-课程映射

这是企业培训推荐区别于电商推荐的核心特征。我们需要构建一个"岗位能力模型"知识图谱，把岗位、技能点、课程三者关联起来。

"""
岗位知识图谱的数据模型

图谱结构：
    JobRole (岗位) --[requires]--> Skill (技能点) --[taught_by]--> Course (课程)
         |                              |
         |--[nice_to_have]--> Skill     |--[difficulty]--> Level (1-5)
                                      |
                                      |--[related_to]--> Skill (相关技能)

示例：
    后端开发工程师 --[requires]--> Python编程 --[taught_by]--> 《Python高级编程》
         |                              |
         |--[requires]--> 分布式系统     |--[related_to]--> 微服务架构
         |                              |
         |--[nice_to_have]--> 前端基础   |--[taught_by]--> 《Django实战》
"""

from dataclasses import dataclass, field


@dataclass
class Skill:
    """技能点"""
    skill_id: str
    name: str                          # 如 "Python编程", "分布式系统"
    category: str                      # 技能领域
    difficulty_level: int              # 技能难度（1-5）
    related_skill_ids: list[str] = field(default_factory=list)


@dataclass
class JobRoleModel:
    """岗位能力模型"""
    job_role: str                      # 岗位编码
    required_skills: list[SkillRequirement] = field(default_factory=list)
    nice_to_have_skills: list[SkillRequirement] = field(default_factory=list)


@dataclass
class SkillRequirement:
    """岗位对技能的要求"""
    skill_id: str
    required_level: int                # 要求达到的水平（1-5）
    priority: float                    # 优先级（0-1）


@dataclass
class SkillCourseMapping:
    """技能点与课程的映射"""
    skill_id: str
    course_id: str
    coverage_score: float              # 该课程对该技能点的覆盖度（0-1）
    is_prerequisite: bool              # 是否为前置课程

知识图谱的构建方式：

技能点体系由培训部门和业务专家共同定义。一个典型的技能体系有 3 层：技能领域（如"技术"、"管理"）→ 技能分类（如"后端开发"、"项目管理"）→ 具体技能点（如"Python编程"、"敏捷开发"）。通常 50-200 个技能点就能覆盖一个中型企业的培训需求。

岗位-技能映射由 HR 和各业务线负责人共同维护。每个岗位定义"必备技能"（required）和"加分技能"（nice_to_have），每个技能标注要求达到的水平等级。

课程-技能映射可以通过两种方式建立：人工标注（课程创建时由讲师选择关联的技能点）和自动提取（用 NLP 模型从课程描述和转录文本中提取技能关键词，与技能点体系做匹配）。实际中建议两者结合——自动提取做初筛，人工做校验。

class KnowledgeGraphService:
    """岗位知识图谱服务"""

    def __init__(self, job_models: dict[str, JobRoleModel],
                 skill_course_map: list[SkillCourseMapping],
                 user_skill_levels: dict[str, dict[str, int]]):
        """
        Args:
            job_models: 岗位 → 能力模型映射
            skill_course_map: 技能点 → 课程映射列表
            user_skill_levels: 用户ID → {技能点ID → 当前水平}
        """
        self.job_models = job_models
        self.skill_course_map = skill_course_map
        self.user_skill_levels = user_skill_levels

    def get_skill_gaps(self, user_id: str, job_role: str) -> list[SkillGap]:
        """
        计算用户的技能缺口：岗位要求水平 - 用户当前水平

        返回按缺口大小排序的技能点列表
        """
        model = self.job_models.get(job_role)
        if not model:
            return []

        current_levels = self.user_skill_levels.get(user_id, {
   })
        gaps = []

        for req in model.required_skills:
            current = current_levels.get(req.skill_id, 0)
            gap = req.required_level - current
            if gap > 0:
                gaps.append(SkillGap(
                    skill_id=req.skill_id,
                    current_level=current,
                    required_level=req.required_level,
                    gap=gap,
                    priority=req.priority,
                    is_required=True
                ))

        for req in model.nice_to_have_skills:
            current = current_levels.get(req.skill_id, 0)
            gap = req.required_level - current
            if gap > 0:
                gaps.append(SkillGap(
                    skill_id=req.skill_id,
                    current_level=current,
                    required_level=req.required_level,
                    gap=gap,
                    priority=req.priority * 0.5,  # 加分技能优先级打折
                    is_required=False
                ))

        # 按 gap × priority 排序
        gaps.sort(key=lambda g: g.gap * g.priority, reverse=True)
        return gaps

    def get_courses_for_skill_gap(self, gap: SkillGap) -> list[str]:
        """根据技能缺口推荐课程"""
        candidates = [
            m for m in self.skill_course_map
            if m.skill_id == gap.skill_id and m.coverage_score > 0.5
        ]
        candidates.sort(key=lambda c: c.coverage_score, reverse=True)
        return [c.course_id for c in candidates]


@dataclass
class SkillGap:
    skill_id: str
    current_level: int
    required_level: int
    gap: int
    priority: float
    is_required: bool

四、召回层：三路并行召回

召回层的核心要求是"快"和"全"——在 50ms 内从 800 门课中筛选出 200-500 门候选课程，且三路召回要尽可能互补，减少重叠。

4.1 协同过滤召回

企业培训场景的数据稀疏度极高，纯 User-CF 或 Item-CF 效果都不好。我们的做法是结合两种 CF，并加入时间衰减和完课率加权。

import numpy as np
from scipy.sparse import csr_matrix
from sklearn.metrics.pairwise import cosine_similarity


class CollaborativeFilteringRecall:
    """
    协同过滤召回路

    策略：
    1. 构建用户-课程交互矩阵（完课=1.0, 学到50%=0.5, 开始学习=0.2）
    2. User-CF：找到与目标用户最相似的 K 个用户，推荐他们学过但目标用户没学过的课
    3. Item-CF：找到与目标用户已学课程最相似的 N 门课
    4. 两路结果合并去重，取 Top 100
    """

    def __init__(self, user_course_matrix: csr_matrix,
                 course_embeddings: np.ndarray):
        """
        Args:
            user_course_matrix: 用户-课程交互矩阵 (n_users × n_courses)
                                值为交互强度（0-1）
            course_embeddings: 课程 Embedding 矩阵 (n_courses × dim)
        """
        self.user_course_matrix = user_course_matrix
        self.course_embeddings = course_embeddings

        # 预计算用户相似度矩阵（离线更新，每天一次）
        self.user_similarity = cosine_similarity(user_course_matrix)

        # 预计算课程相似度矩阵
        self.course_similarity = cosine_similarity(course_embeddings)

    def recall(self, user_id: int, top_k: int = 100) -> list[tuple[int, float]]:
        """
        协同过滤召回

        Returns:
            [(course_id, score), ...] 按分数降序排列
        """
        # User-CF 召回
        user_cf_scores = self._user_cf_recall(user_id, top_k=50)

        # Item-CF 召回
        item_cf_scores = self._item_cf_recall(user_id, top_k=50)

        # 合并两路结果（取并集，分数加权合并）
        merged = {
   }
        for course_id, score in user_cf_scores:
            merged[course_id] = merged.get(course_id, 0) + score * 0.4
        for course_id, score in item_cf_scores:
            merged[course_id] = merged.get(course_id, 0) + score * 0.6

        # 排除用户已学完的课程
        user_learned = set(
            self.user_course_matrix[user_id].nonzero()[1].tolist()
        )
        merged = {
   k: v for k, v in merged.items() if k not in user_learned}

        # 排序取 Top K
        sorted_items = sorted(merged.items(), key=lambda x: x[1], reverse=True)
        return sorted_items[:top_k]

    def _user_cf_recall(self, user_id: int, top_k: int) -> list[tuple[int, float]]:
        """User-CF：相似用户喜欢但我没学过的课程"""
        # 找到最相似的 K 个用户（排除自己）
        similarities = self.user_similarity[user_id].copy()
        similarities[user_id] = 0  # 排除自己

        # 取 Top 20 相似用户
        similar_user_ids = np.argsort(similarities)[-20:][::-1]

        # 统计这些相似用户学过但目标用户没学过的课程
        target_learned = set(
            self.user_course_matrix[user_id].nonzero()[1].tolist()
        )

        course_scores = {
   }
        for sim_uid in similar_user_ids:
            sim_weight = similarities[sim_uid]
            if sim_weight <= 0:
                continue

            learned_courses = self.user_course_matrix[sim_uid].nonzero()[1]
            for cid in learned_courses:
                cid = int(cid)
                if cid in target_learned:
                    continue
                course_scores[cid] = course_scores.get(cid, 0) + sim_weight

        sorted_items = sorted(course_scores.items(), key=lambda x: x[1], reverse=True)
        return sorted_items[:top_k]

    def _item_cf_recall(self, user_id: int, top_k: int) -> list[tuple[int, float]]:
        """Item-CF：与我已学课程相似但我没学过的课程"""
        # 获取用户已学过的课程及其交互强度
        learned = self.user_course_matrix[user_id].nonzero()[1]
        if len(learned) == 0:
            return []

        learned_weights = {
   }
        for cid in learned:
            learned_weights[int(cid)] = float(self.user_course_matrix[user_id, cid])

        # 对每门已学课程，找到最相似的课程
        course_scores = {
   }
        learned_set = set(learned_weights.keys())

        for cid, weight in learned_weights.items():
            similarities = self.course_similarity[cid]

            # 取 Top 10 相似课程
            top_similar = np.argsort(similarities)[-11:][::-1]
            for sim_cid in top_similar:
                sim_cid = int(sim_cid)
                if sim_cid in learned_set:
                    continue  # 排除已学过的
                if sim_cid == cid:
                    continue

                score = similarities[sim_cid] * weight
                course_scores[sim_cid] = course_scores.get(sim_cid, 0) + score

        sorted_items = sorted(course_scores.items(), key=lambda x: x[1], reverse=True)
        return sorted_items[:top_k]

4.2 内容特征召回

内容特征召回的核心思路是：用户学过的课程可以用文本 Embedding 表示，在课程 Embedding 空间中找到与用户"学习画像"最接近的课程。

class ContentFeatureRecall:
    """
    内容特征召回路

    策略：
    1. 用用户已学课程的 Embedding 加权平均，得到用户的"学习画像向量"
    2. 在课程 Embedding 空间中找到与画像向量最接近的课程
    3. 引入"探索因子"：以一定概率推荐与画像向量较远但质量高的课程
    """

    def __init__(self, course_embeddings: np.ndarray,
                 course_features: dict[int, CourseFeatures]):
        self.course_embeddings = course_embeddings  # (n_courses, dim)
        self.course_features = course_features

        # 预计算课程质量的先验分数（评分 × 完课率的加权）
        self.course_quality = np.array([
            self._compute_quality(cf) for cf in course_features.values()
        ])

        # L2 归一化 Embedding，方便计算余弦相似度
        norms = np.linalg.norm(course_embeddings, axis=1, keepdims=True)
        self.normalized_embeddings = course_embeddings / (norms + 1e-8)

    def recall(self, user_features: UserFeatures,
               top_k: int = 100,
               exploration_ratio: float = 0.2) -> list[tuple[int, float]]:
        """
        内容特征召回

        Args:
            user_features: 用户特征
            top_k: 召回数量
            exploration_ratio: 探索比例（20% 的推荐位给"非相似但高质量"的课程）
        """
        # 1. 构建用户学习画像向量
        user_vector = self._build_user_profile(user_features)

        if user_vector is None:
            # 冷启动用户：没有已学课程，用岗位特征构建画像
            user_vector = self._build_cold_start_profile(user_features)

        # 2. 计算与所有课程的相似度
        user_vector_norm = user_vector / (np.linalg.norm(user_vector) + 1e-8)
        similarities = self.normalized_embeddings @ user_vector_norm

        # 3. 分离"利用"和"探索"两部分
        exploit_k = int(top_k * (1 - exploration_ratio))
        explore_k = top_k - exploit_k

        # 利用部分：相似度最高的课程
        exploit_indices = np.argsort(similarities)[-exploit_k * 2:][::-1]

        # 探索部分：相似度中等但质量高的课程
        # 策略：取相似度在 40-70 分位之间、质量分数 Top 的课程
        median_sim = np.median(similarities)
        q40 = np.percentile(similarities, 40)
        q70 = np.percentile(similarities, 70)

        explore_candidates = []
        for i in range(len(similarities)):
            if q40 <= similarities[i] <= q70:
                explore_candidates.append((i, self.course_quality[i]))

        explore_candidates.sort(key=lambda x: x[1], reverse=True)
        explore_indices = [idx for idx, _ in explore_candidates[:explore_k]]

        # 4. 合并结果
        results = {
   }
        for idx in exploit_indices:
            idx = int(idx)
            results[idx] = float(similarities[idx]) * 0.7 + self.course_quality[idx] * 0.3

        for idx in explore_indices:
            idx = int(idx)
            # 探索课程的分数中，质量分权重更高
            results[idx] = float(similarities[idx]) * 0.3 + self.course_quality[idx] * 0.7

        # 5. 排除已学课程
        learned = set(user_features.completed_courses + user_features.in_progress_courses)
        results = {
   k: v for k, v in results.items()
                   if str(k) not in learned}

        sorted_items = sorted(results.items(), key=lambda x: x[1], reverse=True)
        return sorted_items[:top_k]

    def _build_user_profile(self, user: UserFeatures) -> np.ndarray | None:
        """用已学课程的 Embedding 加权平均构建用户画像"""
        if not user.completed_courses:
            return None

        embeddings = []
        weights = []

        for cid_str in user.completed_courses:
            cid = int(cid_str)
            if cid < len(self.course_embeddings):
                embeddings.append(self.course_embeddings[cid])
                # 完课的课程权重为 1.0
                weights.append(1.0)

        for cid_str in user.in_progress_courses:
            cid = int(cid_str)
            if cid < len(self.course_embeddings):
                embeddings.append(self.course_embeddings[cid])
                # 在学的课程权重为 0.5
                weights.append(0.5)

        if not embeddings:
            return None

        embeddings = np.array(embeddings)
        weights = np.array(weights).reshape(-1, 1)

        # 加权平均
        profile = (embeddings * weights).sum(axis=0) / weights.sum()
        return profile

    def _build_cold_start_profile(self, user: UserFeatures) -> np.ndarray:
        """冷启动用户画像：基于岗位和部门特征"""
        # 用岗位关联的课程 Embedding 平均作为初始画像
        # 这里简化处理：用部门内其他用户的平均画像
        # 实际实现中可以从岗位能力模型中提取关联课程
        return np.zeros(self.course_embeddings.shape[1])

    def _compute_quality(self, cf: CourseFeatures) -> float:
        """计算课程的先验质量分"""
        rating_score = cf.avg_rating / 5.0 if cf.rating_count > 0 else 0.5
        completion_score = cf.completion_rate
        popularity_score = min(cf.completion_count / 100.0, 1.0)  # 归一化

        return rating_score * 0.4 + completion_score * 0.4 + popularity_score * 0.2

4.3 知识图谱召回：岗位能力缺口驱动

这是企业培训推荐最独特的召回路——不是基于"用户喜欢什么"，而是基于"用户需要什么"。

class KnowledgeGraphRecall:
    """
    知识图谱召回路

    策略：
    1. 查询用户的岗位能力模型，计算技能缺口
    2. 按缺口大小排序，为每个缺口匹配最佳课程
    3. 引入"相邻技能"扩展：不仅推当前缺口的课，还推相关技能的课
       （培养 T 型人才的跨领域视野）
    """

    def __init__(self, kg_service: KnowledgeGraphService,
                 course_features: dict[int, CourseFeatures]):
        self.kg_service = kg_service
        self.course_features = course_features

    def recall(self, user_features: UserFeatures,
               top_k: int = 100,
               adjacent_expansion: bool = True) -> list[tuple[int, float]]:
        """
        知识图谱召回

        Args:
            user_features: 用户特征
            top_k: 召回数量
            adjacent_expansion: 是否扩展到相邻技能
        """
        user_id = user_features.user_id
        job_role = user_features.job_role

        # 1. 计算技能缺口
        gaps = self.kg_service.get_skill_gaps(user_id, job_role)

        if not gaps:
            # 没有技能缺口（用户已达到岗位要求水平）
            # 退化为推荐"加分技能"或"进阶课程"
            return self._recommend_advanced(user_features, top_k)

        # 2. 为每个技能缺口匹配课程
        course_scores = {
   }

        for i, gap in enumerate(gaps):
            # 缺口权重：gap 越大、priority 越高，权重越大
            gap_weight = (gap.gap / 5.0) * gap.priority

            # 技能缺口排名越靠前，额外加权
            rank_bonus = 1.0 / (1.0 + i * 0.1)

            candidate_courses = self.kg_service.get_courses_for_skill_gap(gap)

            for j, cid_str in enumerate(candidate_courses[:5]):  # 每个技能点最多取 5 门课
                cid = int(cid_str)

                # 课程排名越靠前，coverage_score 越高
                course_bonus = 1.0 / (1.0 + j * 0.2)

                score = gap_weight * rank_bonus * course_bonus
                course_scores[cid] = max(course_scores.get(cid, 0), score)

        # 3. 相邻技能扩展（打破信息茧房的关键机制）
        if adjacent_expansion and gaps:
            expansion_k = int(top_k * 0.3)  # 30% 的推荐位给相邻技能
            adjacent_courses = self._get_adjacent_skill_courses(
                gaps[:5], user_features  # 只对 Top 5 缺口做扩展
            )
            for cid, score in adjacent_courses[:expansion_k]:
                if cid not in course_scores:
                    course_scores[cid] = score * 0.6  # 相邻技能分数打折

        # 4. 排除已学课程
        learned = set(user_features.completed_courses + user_features.in_progress_courses)
        course_scores = {
   k: v for k, v in course_scores.items()
                        if str(k) not in learned}

        sorted_items = sorted(course_scores.items(), key=lambda x: x[1], reverse=True)
        return sorted_items[:top_k]

    def _get_adjacent_skill_courses(
        self, top_gaps: list, user_features: UserFeatures
    ) -> list[tuple[int, float]]:
        """
        获取相邻技能的课程

        相邻技能定义：与当前缺口技能在知识图谱中有 related_to 关系的技能
        例如：缺口是"分布式系统"，相邻技能可能是"微服务架构"、"消息队列"
        """
        adjacent_courses = {
   }

        for gap in top_gaps:
            skill = self.kg_service.get_skill(gap.skill_id)
            if not skill:
                continue

            for related_sid in skill.related_skill_ids:
                # 检查用户是否也缺这个相邻技能
                related_gap = SkillGap(
                    skill_id=related_sid,
                    current_level=self.kg_service.get_user_skill_level(
                        user_features.user_id, related_sid
                    ),
                    required_level=3,  # 相邻技能默认要求 Level 3
                    gap=0,
                    priority=0.3,
                    is_required=False
                )

                courses = self.kg_service.get_courses_for_skill_gap(related_gap)
                for cid_str in courses[:3]:
                    cid = int(cid_str)
                    if cid not in adjacent_courses:
                        adjacent_courses[cid] = 0.5  # 相邻技能课程的基础分

        return sorted(adjacent_courses.items(), key=lambda x: x[1], reverse=True)

    def _recommend_advanced(self, user: UserFeatures,
                            top_k: int) -> list[tuple[int, float]]:
        """
        用户已达到岗位要求水平时的降级策略：
        推荐进阶课程或跨岗位课程
        """
        # 推荐难度高于用户已学课程的高质量课程
        avg_difficulty = np.mean([
            self.course_features[int(cid)].difficulty_level
            for cid in user.completed_courses
            if int(cid) in self.course_features
        ]) if user.completed_courses else 3.0

        candidates = []
        for cid, cf in self.course_features.items():
            if cf.difficulty_level >= avg_difficulty:
                score = cf.avg_rating / 5.0 * cf.completion_rate
                candidates.append((cid, score))

        candidates.sort(key=lambda x: x[1], reverse=True)
        return candidates[:top_k]

五、排序层：从候选到精排

三路召回合并后大约有 200-500 条候选课程，排序层的任务是给出更精准的分数。

5.1 粗排：双塔模型

粗排用双塔模型（用户塔 + 课程塔），分别编码用户特征和课程特征，通过内积计算匹配分数。双塔模型的优势是用户塔和课程塔可以分别预计算，在线推理只需一次向量内积。

import torch
import torch.nn as nn


class DualTowerModel(nn.Module):
    """
    双塔粗排模型

    用户塔：输入用户特征 → 输出 64 维用户向量
    课程塔：输入课程特征 → 输出 64 维课程向量
    匹配分数：用户向量 · 课程向量
    """

    def __init__(self, user_feature_dim: int, course_feature_dim: int,
                 embedding_dim: int = 64):
        super().__init__()

        # 用户塔
        self.user_tower = nn.Sequential(
            nn.Linear(user_feature_dim, 256),
            nn.ReLU(),
            nn.Dropout(0.2),
            nn.Linear(256, 128),
            nn.ReLU(),
            nn.Linear(128, embedding_dim),
            nn.functional.normalize,  # L2 归一化
        )

        # 课程塔
        self.course_tower = nn.Sequential(
            nn.Linear(course_feature_dim, 256),
            nn.ReLU(),
            nn.Dropout(0.2),
            nn.Linear(256, 128),
            nn.ReLU(),
            nn.Linear(128, embedding_dim),
            nn.functional.normalize,
        )

    def forward(self, user_features, course_features):
        user_emb = self.user_tower(user_features)
        course_emb = self.course_tower(course_features)
        # 余弦相似度作为匹配分数
        score = (user_emb * course_emb).sum(dim=-1)
        return score

    def encode_user(self, user_features) -> torch.Tensor:
        """预计算用户向量（每次请求只需算一次）"""
        with torch.no_grad():
            return self.user_tower(user_features)

    def encode_course(self, course_features) -> torch.Tensor:
        """预计算课程向量（离线批量计算）"""
        with torch.no_grad():
            return self.course_tower(course_features)


class CoarseRanker:
    """粗排器"""

    def __init__(self, model: DualTowerModel, device: str = "cpu"):
        self.model = model.to(device)
        self.model.eval()
        self.device = device

        # 预计算所有课程的向量（离线）
        self.course_vectors = None  # 在初始化时加载

    def rank(self, user_vector: torch.Tensor,
             candidate_ids: list[int],
             top_k: int = 50) -> list[tuple[int, float]]:
        """
        粗排：从候选中选出 Top K

        Args:
            user_vector: 预计算的用户向量 (1, dim)
            candidate_ids: 召回层输出的候选课程 ID 列表
            top_k: 保留数量
        """
        # 取出候选课程的预计算向量
        candidate_vectors = self.course_vectors[candidate_ids]  # (n_candidates, dim)

        # 批量计算内积分数
        with torch.no_grad():
            scores = (user_vector @ candidate_vectors.T).squeeze(0)  # (n_candidates,)

        # 取 Top K
        top_indices = scores.topk(top_k).indices
        results = [
            (candidate_ids[i], float(scores[i]))
            for i in top_indices
        ]

        return results

5.2 精排：交叉注意力模型

粗排的 50 条候选进入精排层。精排模型允许用户特征和课程特征做充分的交叉交互，比双塔模型的表达能力更强。

class CrossAttentionRanker(nn.Module):
    """
    精排模型：基于交叉注意力的排序

    核心思路：
    - 用户已学课程序列作为"查询上下文"
    - 候选课程作为"被查询对象"
    - 通过交叉注意力机制，让模型学习"候选课程与用户历史学习模式的匹配度"

    输入特征：
    - 用户侧：已学课程序列 Embedding + 用户属性 Embedding
    - 课程侧：课程 Embedding + 课程统计特征
    - 交叉特征：用户偏好与课程属性的匹配度
    """

    def __init__(self, embedding_dim: int = 128, n_heads: int = 4,
                 n_layers: int = 2):
        super().__init__()

        # 用户已学课程序列编码器
        self.history_encoder = nn.TransformerEncoder(
            nn.TransformerEncoderLayer(
                d_model=embedding_dim,
                nhead=n_heads,
                dim_feedforward=256,
                batch_first=True
            ),
            num_layers=n_layers
        )

        # 交叉注意力层：候选课程 attend to 用户历史
        self.cross_attention = nn.MultiheadAttention(
            embed_dim=embedding_dim,
            num_heads=n_heads,
            batch_first=True
        )

        # 特征融合层
        self.feature_fusion = nn.Sequential(
            nn.Linear(embedding_dim * 3, 256),  # 3 个 embedding 拼接
            nn.ReLU(),
            nn.Dropout(0.3),
            nn.Linear(256, 128),
            nn.ReLU(),
            nn.Linear(128, 1),
            nn.Sigmoid()
        )

    def forward(self, user_history_emb: torch.Tensor,
                user_attr_emb: torch.Tensor,
                candidate_emb: torch.Tensor,
                candidate_stats: torch.Tensor) -> torch.Tensor:
        """
        Args:
            user_history_emb: 用户已学课程序列 (batch, seq_len, dim)
            user_attr_emb: 用户属性向量 (batch, dim)
            candidate_emb: 候选课程 Embedding (batch, dim)
            candidate_stats: 候选课程统计特征 (batch, stats_dim)

        Returns:
            scores: 排序分数 (batch,)
        """
        # 1. 编码用户历史序列
        history_output = self.history_encoder(user_history_emb)  # (batch, seq_len, dim)

        # 2. 交叉注意力：候选课程 attend to 用户历史
        #    让模型学习"候选课程与用户哪些历史学习经历最相关"
        candidate_query = candidate_emb.unsqueeze(1)  # (batch, 1, dim)
        cross_output, attention_weights = self.cross_attention(
            query=candidate_query,
            key=history_output,
            value=history_output
        )  # (batch, 1, dim)
        cross_output = cross_output.squeeze(1)  # (batch, dim)

        # 3. 特征融合：交叉注意力输出 + 用户属性 + 课程统计特征
        fused = torch.cat([cross_output, user_attr_emb, candidate_stats], dim=-1)
        score = self.feature_fusion(fused).squeeze(-1)  # (batch,)

        return score


class FineRanker:
    """精排器"""

    def __init__(self, model: CrossAttentionRanker,
                 feature_builder, device: str = "cpu"):
        self.model = model.to(device)
        self.model.eval()
        self.device = device
        self.feature_builder = feature_builder

    def rank(self, user_features: UserFeatures,
             candidates: list[tuple[int, float]],
             top_k: int = 20) -> list[tuple[int, float]]:
        """
        精排

        Args:
            user_features: 用户特征
            candidates: 粗排输出的候选列表 [(course_id, coarse_score), ...]
            top_k: 保留数量
        """
        if not candidates:
            return []

        # 构建 batch 数据
        candidate_ids = [cid for cid, _ in candidates]
        batch = self.feature_builder.build_ranking_batch(
            user_features, candidate_ids
        )

        # 模型推理
        with torch.no_grad():
            scores = self.model(
                user_history_emb=batch['history_emb'].to(self.device),
                user_attr_emb=batch['user_attr_emb'].to(self.device),
                candidate_emb=batch['candidate_emb'].to(self.device),
                candidate_stats=batch['candidate_stats'].to(self.device),
            )

        # 融合粗排分数和精排分数
        final_scores = {
   }
        for i, (cid, coarse_score) in enumerate(candidates):
            fine_score = float(scores[i])
            # 粗排分数权重 0.3，精排分数权重 0.7
            final_scores[cid] = coarse_score * 0.3 + fine_score * 0.7

        sorted_items = sorted(final_scores.items(), key=lambda x: x[1], reverse=True)
        return sorted_items[:top_k]

六、重排层：多样性控制与信息茧房打破

重排层是推荐系统中最容易被忽略但对企业培训场景最重要的环节。它的目标不是"选最好的课程"，而是"选一组最好的课程"——这组课程需要有足够的多样性，覆盖用户能力的多个维度。

6.1 多样性重排算法

class DiversityReranker:
    """
    多样性重排器

    目标：在保持推荐质量的前提下，最大化推荐列表的多样性

    策略：
    1. 类别多样性：推荐列表中不应有超过 40% 的课程来自同一分类
    2. 难度多样性：推荐列表中应包含不同难度等级的课程
    3. 茧房打破：至少包含 1-2 门用户从未接触过的领域的课程
    4. 必修兜底：如果用户有未完成的必修课程，优先插入推荐列表
    """

    def __init__(self, course_features: dict[int, CourseFeatures]):
        self.course_features = course_features

    def rerank(self, user_features: UserFeatures,
               ranked_courses: list[tuple[int, float]],
               output_size: int = 10,
               mandatory_courses: list[int] = None) -> list[int]:
        """
        重排

        Args:
            user_features: 用户特征
            ranked_courses: 精排输出的排序列表 [(course_id, score), ...]
            output_size: 最终输出数量
            mandatory_courses: 必修课程 ID 列表

        Returns:
            重排后的课程 ID 列表
        """
        result = []
        used_categories = {
   }
        used_difficulties = set()
        user_known_categories = set(user_features.category_preference.keys())

        # Step 1: 插入必修课程（最多 2 门）
        if mandatory_courses:
            for mcid in mandatory_courses[:2]:
                result.append(mcid)
                cf = self.course_features.get(mcid)
                if cf:
                    used_categories[cf.category] = used_categories.get(cf.category, 0) + 1
                    used_difficulties.add(cf.difficulty_level)

        # Step 2: 贪心选择，每一步选择"分数 + 多样性增益"最大的课程
        remaining = [
            (cid, score) for cid, score in ranked_courses
            if cid not in result
        ]

        # 计算每门课程的"探索价值"
        for cid, score in remaining:
            cf = self.course_features.get(cid)
            if not cf:
                continue

            # 类别惩罚：如果该类别已经推荐了很多，降低分数
            cat_count = used_categories.get(cf.category, 0)
            category_penalty = 0.8 ** cat_count  # 每多一门同类别，分数打 8 折

            # 难度多样性奖励：如果推荐列表中还没有这个难度等级，加分
            difficulty_bonus = 1.2 if cf.difficulty_level not in used_difficulties else 1.0

            # 跨领域奖励：如果这是用户从未接触过的领域，加分
            exploration_bonus = 1.3 if cf.category not in user_known_categories else 1.0

            # 综合分数
            adjusted_score = score * category_penalty * difficulty_bonus * exploration_bonus

            # 临时存储
            cf._adjusted_score = adjusted_score

        # 按调整后的分数排序
        remaining.sort(key=lambda x: getattr(
            self.course_features.get(x[0]), '_adjusted_score', x[1]
        ), reverse=True)

        # Step 3: 贪心选择，确保多样性约束
        for cid, score in remaining:
            if len(result) >= output_size:
                break

            cf = self.course_features.get(cid)
            if not cf:
                continue

            # 类别多样性约束：同一类别不超过 40%
            max_per_category = max(1, int(output_size * 0.4))
            if used_categories.get(cf.category, 0) >= max_per_category:
                continue

            result.append(cid)
            used_categories[cf.category] = used_categories.get(cf.category, 0) + 1
            used_difficulties.add(cf.difficulty_level)

        # Step 4: 茧房打破检查
        # 确保至少有 1 门课程来自用户未接触过的领域
        has_exploration = any(
            self.course_features.get(cid, CourseFeatures("", "", "", [], 1, 0, "", None, None, [], [], 0, 0, 0, 0, None)).category
            not in user_known_categories
            for cid in result
        )

        if not has_exploration and len(result) < output_size:
            # 从剩余候选中找一门跨领域课程插入
            for cid, score in remaining:
                if cid in result:
                    continue
                cf = self.course_features.get(cid)
                if cf and cf.category not in user_known_categories:
                    # 替换最后一门同领域课程
                    result[-1] = cid
                    break

        return result

6.2 信息茧房打破的量化指标

怎么知道推荐系统有没有打破信息茧房？需要定义量化指标并持续监控。

class FilterBubbleMonitor:
    """
    信息茧房监控器

    持续跟踪每个用户的推荐多样性，如果多样性持续下降，触发告警
    """

    def __init__(self):
        self.user_diversity_history = {
   }  # user_id → [diversity_score_per_week]

    def compute_diversity(self, recommended_courses: list[int],
                          course_features: dict[int, CourseFeatures]) -> dict:
        """
        计算一次推荐结果的多样性指标
        """
        categories = []
        difficulties = []
        tags = set()

        for cid in recommended_courses:
            cf = course_features.get(cid)
            if cf:
                categories.append(cf.category)
                difficulties.append(cf.difficulty_level)
                tags.update(cf.tags)

        # 指标 1: 类别熵（Shannon Entropy）
        # 熵越高，类别分布越均匀，多样性越好
        from collections import Counter
        cat_counts = Counter(categories)
        total = len(categories)
        entropy = -sum(
            (count / total) * np.log2(count / total)
            for count in cat_counts.values()
        ) if total > 0 else 0

        # 归一化到 0-1（最大熵 = log2(类别总数)）
        max_entropy = np.log2(len(set(cf.category for cf in course_features.values())))
        normalized_entropy = entropy / max_entropy if max_entropy > 0 else 0

        # 指标 2: 难度覆盖范围
        difficulty_range = max(difficulties) - min(difficulties) if difficulties else 0

        # 指标 3: 标签丰富度（推荐列表中不同标签的数量）
        tag_richness = len(tags)

        # 指标 4: 新颖度（用户从未接触过的类别占比）
        # 需要与用户的历史学习类别对比
        novelty = 0.0  # 在外部计算

        return {
   
            "category_entropy": normalized_entropy,
            "difficulty_range": difficulty_range,
            "tag_richness": tag_richness,
            "novelty": novelty,
            "n_courses": len(recommended_courses),
        }

    def track(self, user_id: str, diversity_metrics: dict):
        """记录用户的多样性历史"""
        if user_id not in self.user_diversity_history:
            self.user_diversity_history[user_id] = []

        self.user_diversity_history[user_id].append(diversity_metrics)

    def check_bubble_risk(self, user_id: str) -> dict:
        """
        检查用户是否有陷入信息茧房的风险

        判定标准：
        - 连续 4 周类别熵低于 0.5
        - 连续 4 周新颖度低于 0.1
        """
        history = self.user_diversity_history.get(user_id, [])
        if len(history) < 4:
            return {
   "risk": "LOW", "reason": "数据不足"}

        recent_4 = history[-4:]
        avg_entropy = np.mean([h["category_entropy"] for h in recent_4])
        avg_novelty = np.mean([h.get("novelty", 0) for h in recent_4])

        if avg_entropy < 0.5 and avg_novelty < 0.1:
            return {
   
                "risk": "HIGH",
                "reason": f"连续 4 周推荐多样性过低（熵={avg_entropy:.2f}, 新颖度={avg_novelty:.2f}）",
                "action": "触发跨领域推荐加强"
            }
        elif avg_entropy < 0.6:
            return {
   
                "risk": "MEDIUM",
                "reason": f"推荐多样性偏低（熵={avg_entropy:.2f}）",
                "action": "适当增加探索比例"
            }
        else:
            return {
   "risk": "LOW", "reason": "推荐多样性正常"}

七、冷启动方案：新员工的第一个推荐列表

冷启动是企业培训推荐的核心挑战。新员工入职第一天就需要学习，但没有任何历史数据。我们的方案是"三层兜底"。

7.1 冷启动三层策略

class ColdStartRecommender:
    """
    冷启动推荐器

    三层策略（按优先级）：

    Layer 1: 岗位必修课程
    - 每个岗位定义了"入职必修"课程列表
    - 新员工无条件推荐这些课程
    - 这不是"推荐"，是"要求"，但通过推荐列表展示

    Layer 2: 岗位能力模型驱动
    - 根据岗位能力模型，推荐填补最大技能缺口的课程
    - 即使没有个人历史数据，岗位信息已经提供了强信号

    Layer 3: 部门/团队热门课程
    - 推荐同部门、同岗位的其他员工最常学的课程
    - 本质上是"群体画像"代替"个人画像"
    """

    def __init__(self,
                 mandatory_courses: dict[str, list[str]],
                 kg_service: KnowledgeGraphService,
                 department_popular: dict[str, list[int]]):
        """
        Args:
            mandatory_courses: 岗位 → 必修课程 ID 列表
            kg_service: 知识图谱服务
            department_popular: 部门 → 热门课程 ID 列表
        """
        self.mandatory_courses = mandatory_courses
        self.kg_service = kg_service
        self.department_popular = department_popular

    def recommend(self, user_features: UserFeatures,
                  output_size: int = 10) -> list[int]:
        """
        冷启动推荐

        Args:
            user_features: 用户特征（只有基础属性，没有行为特征）
            output_size: 推荐数量
        """
        result = []

        # Layer 1: 岗位必修课程（最多占 30% 的推荐位）
        mandatory = self.mandatory_courses.get(user_features.job_role, [])
        mandatory_slots = max(1, int(output_size * 0.3))
        result.extend([int(cid) for cid in mandatory[:mandatory_slots]])

        # Layer 2: 岗位能力模型驱动（最多占 50% 的推荐位）
        kg_slots = max(1, int(output_size * 0.5))
        gaps = self.kg_service.get_skill_gaps(user_features.user_id, user_features.job_role)

        for gap in gaps:
            if len(result) >= mandatory_slots + kg_slots:
                break
            courses = self.kg_service.get_courses_for_skill_gap(gap)
            for cid_str in courses:
                cid = int(cid_str)
                if cid not in result:
                    result.append(cid)
                    break

        # Layer 3: 部门热门课程（填充剩余位置）
        remaining_slots = output_size - len(result)
        if remaining_slots > 0:
            popular = self.department_popular.get(user_features.department, [])
            for cid_str in popular:
                cid = int(cid_str)
                if cid not in result:
                    result.append(cid)
                    if len(result) >= output_size:
                        break

        return result

    def should_exit_cold_start(self, user_features: UserFeatures) -> bool:
        """
        判断用户是否应该退出冷启动模式

        条件：
        - 已完成 3 门以上课程
        - 或有 10 小时以上学习记录
        """
        completed = len(user_features.completed_courses)
        hours = user_features.learning_hours_total

        return completed >= 3 or hours >= 10

7.2 冷启动到个性化推荐的平滑过渡

用户从冷启动退出后，不应该突然切换到完全个性化的推荐——数据量还不够支撑可靠的个性化。需要一个过渡期。

class RecommendationBlender:
    """
    推荐混合器：在冷启动和个性化推荐之间平滑过渡

    策略：
    - 完课 3-5 门：70% 岗位推荐 + 30% 个性化推荐
    - 完课 6-10 门：40% 岗位推荐 + 60% 个性化推荐
    - 完课 10 门以上：10% 岗位推荐 + 90% 个性化推荐
    - 岗位推荐永远不会降到 0（确保能力模型始终在引导）
    """

    def blend(self,
              cold_start_courses: list[int],
              personalized_courses: list[int],
              user_features: UserFeatures,
              output_size: int = 10) -> list[int]:

        completed = len(user_features.completed_courses)

        # 计算混合比例
        if completed < 3:
            kg_ratio = 0.7
        elif completed < 6:
            kg_ratio = 0.7 - (completed - 3) * 0.1  # 3→0.7, 4→0.6, 5→0.5
        elif completed < 10:
            kg_ratio = 0.4 - (completed - 6) * 0.075  # 6→0.4, 10→0.1
        else:
            kg_ratio = 0.1  # 最低保留 10%

        kg_slots = max(1, int(output_size * kg_ratio))
        personal_slots = output_size - kg_slots

        result = []

        # 岗位推荐部分
        for cid in cold_start_courses[:kg_slots]:
            if cid not in result:
                result.append(cid)

        # 个性化推荐部分
        for cid in personalized_courses[:personal_slots]:
            if cid not in result:
                result.append(cid)

        # 如果个性化推荐不够，用岗位推荐填充
        if len(result) < output_size:
            for cid in cold_start_courses:
                if cid not in result:
                    result.append(cid)
                    if len(result) >= output_size:
                        break

        return result

八、工程落地：离线与在线的分工

推荐系统不能每次请求都从头计算。工程上需要把计算拆成离线和在线两部分。

8.1 离线任务（每天凌晨执行）

class OfflineRecommendationPipeline:
    """
    离线推荐流水线（每日执行）

    任务清单：
    1. 更新用户-课程交互矩阵
    2. 重新计算用户相似度和课程相似度矩阵
    3. 重新训练/更新粗排双塔模型
    4. 预计算所有课程的 Embedding 和粗排向量
    5. 更新部门/岗位热门课程排行
    6. 更新知识图谱中的技能缺口数据
    7. 计算每个用户的多样性指标，更新茧房监控
    """

    def run_daily(self):
        print("[Offline] Starting daily recommendation pipeline...")

        # Step 1: 更新交互矩阵
        print("[Offline] Step 1: Updating user-course interaction matrix...")
        self.update_interaction_matrix()

        # Step 2: 重算相似度矩阵
        print("[Offline] Step 2: Recomputing similarity matrices...")
        self.recompute_similarities()

        # Step 3: 预计算课程向量
        print("[Offline] Step 3: Pre-computing course embeddings...")
        self.precompute_course_embeddings()

        # Step 4: 更新热门排行
        print("[Offline] Step 4: Updating popularity rankings...")
        self.update_popularity_rankings()

        # Step 5: 更新技能缺口
        print("[Offline] Step 5: Updating skill gap data...")
        self.update_skill_gaps()

        # Step 6: 多样性监控
        print("[Offline] Step 6: Computing diversity metrics...")
        self.compute_diversity_metrics()

        # Step 7: 为活跃用户预计算推荐列表
        print("[Offline] Step 7: Pre-computing recommendations for active users...")
        self.precompute_recommendations()

        print("[Offline] Daily pipeline completed.")

    def precompute_recommendations(self):
        """
        为活跃用户预计算推荐列表
        结果缓存到 Redis，在线请求直接读取
        """
        active_users = self.get_active_users(days=7)  # 近 7 天活跃的用户

        for user_features in active_users:
            # 三路召回
            cf_candidates = self.cf_recall.recall(user_features, top_k=100)
            content_candidates = self.content_recall.recall(user_features, top_k=100)
            kg_candidates = self.kg_recall.recall(user_features, top_k=100)

            # 合并去重
            all_candidates = self.merge_candidates(
                cf_candidates, content_candidates, kg_candidates
            )

            # 粗排
            coarse_ranked = self.coarse_ranker.rank(
                user_features.embedding, all_candidates, top_k=50
            )

            # 精排
            fine_ranked = self.fine_ranker.rank(
                user_features, coarse_ranked, top_k=20
            )

            # 重排
            final = self.reranker.rerank(
                user_features, fine_ranked, output_size=10,
                mandatory_courses=self.get_mandatory(user_features)
            )

            # 写入 Redis 缓存
            cache_key = f"recommend:{user_features.user_id}"
            self.redis.set(cache_key, json.dumps(final), ex=86400 * 2)  # 缓存 2 天

8.2 在线服务（实时请求）

class OnlineRecommendationService:
    """
    在线推荐服务

    延迟目标：P99 < 100ms

    策略：
    1. 优先读缓存（预计算的推荐列表）
    2. 缓存未命中时，实时计算（只用简化的召回 + 排序）
    3. 冷启动用户走专门的冷启动推荐器
    """

    def __init__(self, redis_client, offline_pipeline, cold_start_recommender):
        self.redis = redis_client
        self.offline = offline_pipeline
        self.cold_start = cold_start_recommender

    def get_recommendations(self, user_features: UserFeatures,
                            context: dict = None) -> list[int]:
        """
        获取推荐列表

        Args:
            user_features: 用户特征
            context: 请求上下文（当前页面、来源等）
        """
        # 1. 冷启动检查
        if not self.offline.should_exit_cold_start(user_features):
            return self.cold_start.recommend(user_features)

        # 2. 读缓存
        cache_key = f"recommend:{user_features.user_id}"
        cached = self.redis.get(cache_key)

        if cached:
            recommended = json.loads(cached)

            # 过滤掉用户刚刚看过的课程
            just_viewed = set(context.get("just_viewed", []))
            recommended = [cid for cid in recommended if cid not in just_viewed]

            # 上下文调整：如果用户当前在某个课程分类页面，
            # 适当提升该分类课程的排名
            if context and context.get("current_category"):
                recommended = self._boost_category(
                    recommended, context["current_category"]
                )

            return recommended[:10]

        # 3. 缓存未命中：实时计算（简化版）
        return self._realtime_recommend(user_features)

    def _realtime_recommend(self, user_features: UserFeatures) -> list[int]:
        """实时计算推荐（简化版，只用内容特征召回 + 质量排序）"""
        # 简化版：只用内容特征召回，跳过协同过滤和知识图谱（太慢）
        candidates = self.offline.content_recall.recall(
            user_features, top_k=30
        )

        # 按质量分排序
        candidates.sort(key=lambda x: x[1], reverse=True)

        return [cid for cid, _ in candidates[:10]]

    def _boost_category(self, recommended: list[int],
                        category: str) -> list[int]:
        """提升特定分类课程的排名"""
        boosted = []
        normal = []

        for cid in recommended:
            cf = self.offline.course_features.get(cid)
            if cf and cf.category == category:
                boosted.append(cid)
            else:
                normal.append(cid)

        # 提升的课程放在前面，但不超过总推荐数的 50%
        max_boosted = max(1, len(recommended) // 2)
        return boosted[:max_boosted] + normal

九、效果度量：怎么知道推荐系统有用

推荐系统上线后，需要持续监控以下指标。

@dataclass
class RecommendationMetrics:
    """推荐系统核心指标"""

    # === 效果指标 ===
    click_through_rate: float        # 推荐点击率（点击推荐课程 / 展示推荐课程）
    enrollment_rate: float           # 推荐转化率（通过推荐开始学习 / 点击推荐课程）
    completion_rate_lift: float      # 完课率提升（推荐课程的完课率 vs 自然浏览的完课率）

    # === 覆盖指标 ===
    catalog_coverage: float          # 目录覆盖率（被推荐过的课程 / 全部课程）
    user_coverage: float             # 用户覆盖率（获得过推荐的用户 / 全部用户）

    # === 多样性指标 ===
    category_entropy: float          # 类别熵（推荐列表的类别分布均匀度）
    novelty_score: float             # 新颖度（用户首次接触的分类占比）

    # === 冷启动指标 ===
    cold_start_completion: float     # 冷启动用户完课率
    cold_start_to_active: float      # 冷启动用户转化为活跃用户的比例

    # === 业务指标 ===
    learning_hours_per_user: float   # 人均学习时长（推荐系统上线前后对比）
    skill_gap_closure_rate: float    # 技能缺口补齐率（推荐课程覆盖的技能缺口比例）


class RecommendationMetricsCollector:
    """指标采集器"""

    def compute_online_metrics(self, start_date: str, end_date: str) -> RecommendationMetrics:
        """
        从埋点数据计算在线指标

        关键埋点事件：
        - recommend_impression: 推荐列表展示
        - recommend_click: 点击推荐课程
        - course_start: 开始学习课程（带来源标记：recommend / browse / search）
        - course_complete: 完课
        """

        # 推荐点击率
        impressions = self.get_event_count("recommend_impression", start_date, end_date)
        clicks = self.get_event_count("recommend_click", start_date, end_date)
        ctr = clicks / impressions if impressions > 0 else 0

        # 推荐转化率
        recommend_starts = self.get_event_count(
            "course_start", start_date, end_date,
            filter={
   "source": "recommend"}
        )
        enrollment_rate = recommend_starts / clicks if clicks > 0 else 0

        # 目录覆盖率
        recommended_courses = self.get_distinct_courses(
            "recommend_impression", start_date, end_date
        )
        total_courses = self.get_total_courses()
        catalog_coverage = len(recommended_courses) / total_courses

        # 新颖度
        # 统计用户通过推荐学习的课程中，有多少是他们从未接触过的分类
        ...

        return RecommendationMetrics(
            click_through_rate=ctr,
            enrollment_rate=enrollment_rate,
            catalog_coverage=catalog_coverage,
            ...
        )

十、总结与关键数据

这套混合推荐系统在一个 5000 人、800 门课程的企业培训平台上运行 6 个月后的关键数据：

几个核心经验：

三路召回的互补性比单路精度更重要。 协同过滤召回的课程有 40% 与内容特征召回重叠，知识图谱召回有 30% 与其他两路重叠。重叠不是浪费——重叠意味着多路信号一致，这些课程在排序层会获得更高的分数。

冷启动的关键不是算法，是数据。 岗位信息、部门信息、职级信息——这些 HR 系统里已有的数据，就是冷启动阶段最强的推荐信号。不需要用户有任何学习行为，岗位能力模型就能生成一份"合理"的推荐列表。

信息茧房不是靠一次重排就能解决的。 需要持续监控每个用户的推荐多样性，在多样性持续下降时主动干预。我们设置了每周自动检查的茧房监控器，对高风险用户自动提升探索比例。

推荐系统的目标函数要和业务目标对齐。 电商推荐优化点击率，企业培训推荐应该优化"技能缺口补齐率"。一个用户点击了很多推荐课程但没有补齐任何技能缺口，这个推荐是失败的。

如果你的培训平台也在考虑做推荐系统，建议按"知识图谱召回 → 冷启动方案 → 内容特征召回 → 协同过滤召回 → 精排模型 → 多样性重排"的顺序逐步落地。前两步做完就能覆盖 80% 的场景，后面的模块可以迭代优化。