用AI做学习路径推荐：基于知识图谱的个性化学习方案落地

企业培训场景下，"学什么"比"怎么学"更难。本文记录我们如何用知识图谱 + 学习者画像 + 混合推荐算法，为数十万企业员工构建个性化学习路径的工程实践。

一、问题背景：企业培训的"千人一面"困境

企学宝服务了大量中大型企业客户，课程库规模通常在数千到数万门之间。在引入AI推荐之前，学习路径主要靠培训管理员手工编排——给所有新员工配同一套入职课程，给所有销售配同一套销售技巧课。这套模式有三个明显问题：

第一，个体差异被忽略。同样是新入职的销售，一个有3年行业经验，一个刚毕业，学习起点完全不同，但被塞进同一条路径。

第二，知识断层频发。员工跳过前置课程直接学习高阶内容，导致完课率低、考试通过率低。某客户的数据显示，未按推荐顺序学习的员工，考试首次通过率只有 41%，而按序学习的达到 78%。

第三，管理员负担重。一个大企业的培训管理员要维护几十条岗位学习路径，每次课程库更新都要手动调整，路径之间还存在大量重复。

目标很明确：给每个员工生成一条个性化的、知识结构合理的、可持续优化的学习路径。不是简单地推一门课，而是推一条"路径"——从当前位置到目标岗位能力的最优学习序列。

二、整体架构设计

系统分为四层，从下到上依次是数据层、画像层、推荐引擎层和业务应用层。

┌─────────────────────────────────────────────────────────┐
│                    业务应用层                            │
│  学习路径卡片 / 课程详情页 / 学习中心首页 / 岗位发展计划   │
└──────────────────────────┬──────────────────────────────┘
                           │ REST / gRPC
┌──────────────────────────┴──────────────────────────────┐
│                    推荐引擎层                            │
│  ┌─────────┐  ┌──────────┐  ┌──────────┐  ┌──────────┐  │
│  │ 路径规划 │  │ 协同过滤  │  │ 内容召回  │  │ 重排策略  ││
│  │ (图搜索) │  │ (CF)     │  │ (召回层)  │  │ (业务规则)││
│  └─────────┘  └──────────┘  └──────────┘  └──────────┘  │
└──────────────────────────┬──────────────────────────────┘
                           │
┌──────────────────────────┴──────────────────────────────┐
│                      画像层                             │
│  ┌──────────────┐       ┌──────────────┐                │
│  │ 学习者画像    │       │ 知识图谱服务  │               │
│  │ (实时+离线)   │       │ (Neo4j)      │               │
│  └──────────────┘       └──────────────┘                │
└──────────────────────────┬──────────────────────────────┘
                           │
┌──────────────────────────┴──────────────────────────────┐
│                    数据层                               │
│  课程库 /考试成绩 /学习行为埋点 /组织架构 /岗位胜任力模型  │
└─────────────────────────────────────────────────────────┘

数据层不用多说，就是业务库里的课程元数据、学习行为日志、考试结果等。重点讲中间三层。

三、知识图谱构建：学习路径的"骨架"

学习路径推荐的第一步，是把课程库结构化成一张知识图谱。没有图谱，推荐就只是"看了这门课的人还看了那门课"的协同过滤，无法保证知识结构的合理性。

3.1 图谱Schema设计

我们定义了四类核心实体和四类关系：

实体类型：

关系类型：

Cypher 建模示例：

// 创建课程节点
CREATE (c:Course {id: 'C001', title: 'Python数据分析实战', difficulty: 3, duration: 480})

// 创建知识点节点
CREATE (kp1:KnowledgePoint {id: 'KP001', name: 'Python基础语法', level: 'basic'})
CREATE (kp2:KnowledgePoint {id: 'KP002', name: 'Pandas数据清洗', level: 'intermediate'})
CREATE (kp3:KnowledgePoint {id: 'KP003', name: 'Matplotlib可视化', level: 'intermediate'})

// 课程包含知识点
CREATE (c)-[:CONTAINS {order: 1}]->(kp1)
CREATE (c)-[:CONTAINS {order: 2}]->(kp2)
CREATE (c)-[:CONTAINS {order: 3}]->(kp3)

// 前置依赖
CREATE (kp2)-[:PREREQUISITE]->(kp1)
CREATE (kp3)-[:PREREQUISITE]->(kp2)

// 技能聚合
CREATE (s:Skill {id: 'S001', name: '数据处理与分析'})
CREATE (kp2)-[:BELONGS_TO]->(s)
CREATE (kp3)-[:BELONGS_TO]->(s)

// 岗位技能要求
CREATE (p:Position {id: 'P001', name: '数据分析师', level: 'L3'})
CREATE (p)-[:REQUIRES {weight: 0.8}]->(s)

3.2 知识点抽取：从非结构化课程到结构化图谱

最大的工程量在于，课程库里的课程只有标题、简介和大纲文本，没有结构化的知识点标注。手动标注几万门课程不现实。

我们的方案是用LLM做知识点抽取 + 人工校验：

import json
from openai import OpenAI

client = OpenAI()

EXTRACTION_PROMPT = """你是一个企业培训课程的知识点提取专家。
请从以下课程信息中提取结构化的知识点列表。

课程标题：{title}
课程简介：{description}
课程大纲：{outline}

请按以下JSON格式输出，每个知识点包含：
- name: 知识点名称（简洁，不超过15字）
- level: 难度级别（basic/intermediate/advanced）
- description: 一句话描述

输出格式：
{
   {"knowledge_points": [{
   {"name": "...", "level": "...", "description": "..."}}]}}

注意：
1. 知识点应该是原子级别的，不可再分
2. 一门课程的的知识点数量控制在3-12个
3. 保持术语规范化
"""

def extract_knowledge_points(course_info: dict) -> list[dict]:
    prompt = EXTRACTION_PROMPT.format(
        title=course_info["title"],
        description=course_info["description"],
        outline=course_info["outline"]
    )
    response = client.chat.completions.create(
        model="qwen-max",
        messages=[{
   "role": "user", "content": prompt}],
        response_format={
   "type": "json_object"},
        temperature=0.3
    )
    result = json.loads(response.choices[0].message.content)
    return result["knowledge_points"]

抽取之后还有一个关键步骤——知识点对齐。不同课程抽取出的知识点名称可能不一致（比如"Pandas数据清洗"和"用Pandas清洗数据"），需要做实体消歧。我们用embedding相似度 + 人工规则做了一层对齐：

from sklearn.metrics.pairwise import cosine_similarity
import numpy as np

def align_knowledge_points(raw_points: list[dict], existing_points: list[dict],
                           embedding_model, threshold: float = 0.85) -> list[dict]:
    """
    将新抽取的知识点与已有知识点库对齐：
    - 相似度 > threshold：合并到已有知识点
    - 相似度 < threshold：创建新知识点
    """
    if not existing_points:
        return raw_points

    existing_embeddings = np.array([
        embedding_model.embed(p["name"] + " " + p.get("description", ""))
        for p in existing_points
    ])

    aligned = []
    for rp in raw_points:
        rp_emb = embedding_model.embed(rp["name"] + " " + rp.get("description", ""))
        sims = cosine_similarity([rp_emb], existing_embeddings)[0]
        best_idx = np.argmax(sims)

        if sims[best_idx] > threshold:
            # 合并到已有知识点，保留已有ID
            rp["aligned_id"] = existing_points[best_idx]["id"]
            rp["similarity"] = float(sims[best_idx])
        aligned.append(rp)

    return aligned

前置依赖关系的建立更复杂，我们用了两种策略：一是从课程大纲的章节顺序中提取隐式依赖（前章节是后章节的前置）；二是用LLM判断两个知识点之间是否存在前置关系。两种策略的结果做交叉验证，不一致的交人工审核。

3.3 图谱存储与查询

选择 Neo4j 作为图数据库，主要原因是学习路径规划本质是图搜索问题，Neo4j 的 Cypher 查询在处理多跳路径时性能远优于关系型数据库。

一个典型的查询场景——"找出从员工当前掌握的知识点到目标岗位所需技能的最短学习路径"：

// 输入：员工已掌握的知识点ID列表 [KP001]，目标岗位 P001
// 输出：需要补充学习的知识点序列

MATCH (p:Position {id: 'P001'})-[:REQUIRES]->(s:Skill)<-[:BELONGS_TO]-(kp:KnowledgePoint)
WHERE NOT kp.id IN ['KP001']  // 排除已掌握的
MATCH path = shortestPath((kp)-[:PREREQUISITE*0..5]->(target:KnowledgePoint))
WHERE target.id IN ['KP001'] OR target.id IN ['KP002']
RETURN kp.name AS need_to_learn, kp.level AS difficulty,
       [n in nodes(path) | n.name] AS dependency_chain,
       length(path) AS distance
ORDER BY distance ASC

这个查询会返回员工需要学习的知识点及其前置依赖链。但实际生产中，纯粹的 shortestPath 不够用——还要考虑课程载体（知识点要通过课程来学）、课程难度递进、学习时长约束等。所以我们在应用层做了更复杂的路径规划，后面会讲。

四、学习者画像：知道员工"在哪、会什么、想去哪"

学习者画像解决三个问题：员工当前的知识掌握水平（在哪）、学习偏好（怎么学）、发展目标（想去哪）。

4.1 知识掌握度评估

掌握度是画像的核心。我们没有简单地把"学过=掌握"，而是设计了一个综合评分模型：

 mastery = w1 * 考试成绩得分
        + w2 * 练习正确率
        + w3 * 学习行为完整度
        + w4 * 知识点覆盖率

各项权重通过历史数据回归得到，当前线上值为 w1=0.45, w2=0.25, w3=0.15, w4=0.15。

其中"知识点覆盖率"是指员工学过的课程覆盖了该知识点关联的多少门课程：

def calculate_mastery(employee_id: str, knowledge_point_id: str,
                      db_session) -> float:
    """计算某员工对某知识点的掌握度（0-1）"""
    # 1. 考试成绩：与该知识点关联的考试得分
    exam_score = db_session.query_exam_avg_score(
        employee_id, knowledge_point_id
    )  # 0-1

    # 2. 练习正确率
    exercise_accuracy = db_session.query_exercise_accuracy(
        employee_id, knowledge_point_id
    )  # 0-1

    # 3. 学习行为完整度：视频完播率、课件阅读完成率
    behavior_completeness = db_session.query_behavior_completeness(
        employee_id, knowledge_point_id
    )  # 0-1

    # 4. 知识点覆盖率：关联课程的学习覆盖比例
    related_courses = db_session.query_courses_by_kp(knowledge_point_id)
    learned_courses = db_session.query_learned_courses(employee_id)
    coverage = len(set(related_courses) & set(learned_courses)) / max(len(related_courses), 1)

    # 贝叶斯平滑：数据量少时向0.5收敛，避免"做对1道题=100%掌握"
    exam_count = db_session.query_exam_count(employee_id, knowledge_point_id)
    exam_score = bayesian_smooth(exam_score, exam_count, prior=0.5, strength=5)

    mastery = (
        0.45 * exam_score +
        0.25 * exercise_accuracy +
        0.15 * behavior_completeness +
        0.15 * coverage
    )
    return min(max(mastery, 0.0), 1.0)

def bayesian_smooth(observed: float, n: int, prior: float = 0.5,
                    strength: int = 5) -> float:
    """贝叶斯平滑：数据量不足时向先验收敛"""
    return (observed * n + prior * strength) / (n + strength)

贝叶斯平滑这一步很关键。早期版本没做平滑，结果一个员工只做对了一道题，掌握度就显示100%，推荐系统认为他已经掌握了这个知识点，不再推荐相关课程——这显然不合理。加上平滑后，数据量少时掌握度会向0.5收敛，随着练习/考试次数增加，逐渐逼近真实水平。

4.2 学习偏好建模

偏好信号包括：偏好的内容形式（视频/文档/直播）、偏好的学习时段、单次学习时长、难度偏好（喜欢挑战高难度还是稳扎稳打）。这些信号从行为埋点中提取，用简单的统计特征即可，不需要复杂模型。

4.3 画像存储与更新

画像数据分冷热两层：

• 冷数据（知识掌握度、岗位信息）：更新频率低（每次考试/课程完成后更新），存 MySQL + Redis 缓存
• 热数据（实时学习行为、偏好信号）：更新频率高（每次学习行为都产生），走 Kafka 实时计算，写入 Redis

实时画像更新流程：

# Kafka consumer：监听学习行为事件，实时更新画像
from kafka import KafkaConsumer
import json

consumer = KafkaConsumer(
    'learning-behavior-events',
    bootstrap_servers=['kafka:9092'],
    group_id='profile-updater',
    value_deserializer=lambda x: json.loads(x.decode('utf-8'))
)

for message in consumer:
    event = message.value
    emp_id = event["employee_id"]
    event_type = event["type"]  # video_complete / exercise_submit / exam_submit

    if event_type == "video_complete":
        # 更新行为完整度
        update_behavior_completeness(emp_id, event)
    elif event_type == "exam_submit":
        # 重新计算掌握度（考试权重最高）
        recalculate_mastery(emp_id, event["knowledge_point_id"])

    # 将更新后的画像推送到推荐引擎的缓存
    publish_profile_update(emp_id)

五、推荐算法：三路混合策略

单一推荐算法很难同时满足"路径合理""内容相关""个性化"三个目标。我们采用了三路混合策略，每一路解决一个问题，最后做融合重排。

5.1 第一路：基于知识图谱的路径规划

这一路解决"路径合理性"——保证推荐的课程序列符合知识前置依赖。

核心思路是把学习路径规划建模成一个带约束的最短路径问题：从员工当前的知识状态出发，到达目标岗位的技能要求，在满足前置依赖约束下，找到一条学习成本最低的路径。

from collections import defaultdict
import heapq

class KnowledgePathPlanner:
    """基于知识图谱的最短学习路径规划"""

    def __init__(self, graph_client):
        self.graph = graph_client  # Neo4j client

    def plan_path(self, employee_id: str, target_position_id: str,
                  max_courses: int = 10, max_duration_min: int = 600) -> list[dict]:
        # 1. 获取员工已掌握的知识点
        mastered_kps = self.graph.get_mastered_knowledge_points(employee_id)

        # 2. 获取目标岗位需要的知识点
        required_kps = self.graph.get_required_knowledge_points(target_position_id)

        # 3. 计算需要补充的知识点（目标 - 已掌握）
        gap_kps = set(required_kps) - set(mastered_kps)

        if not gap_kps:
            return []  # 岗位技能已满足，无需推荐

        # 4. 获取每个gap知识点的所有前置依赖链
        dependency_graph = self._build_dependency_subgraph(gap_kps, mastered_kps)

        # 5. 拓扑排序：确定知识点学习顺序
        learning_order = self._topological_sort(dependency_graph, mastered_kps)

        # 6. 为每个知识点匹配最优课程（考虑难度适配、时长约束）
        recommended_courses = []
        for kp in learning_order:
            course = self._match_best_course(kp, employee_id, max_duration_min)
            if course:
                recommended_courses.append({
   
                    "knowledge_point": kp,
                    "course": course,
                    "reason": "knowledge_gap"
                })
            if len(recommended_courses) >= max_courses:
                break

        return recommended_courses

    def _topological_sort(self, dep_graph: dict, mastered: set) -> list:
        """拓扑排序，保证前置知识点排在前面"""
        in_degree = defaultdict(int)
        for kp, deps in dep_graph.items():
            for dep in deps:
                if dep not in mastered:
                    in_degree[kp] += 1

        # 入度为0的知识点先学
        queue = [kp for kp in dep_graph if in_degree[kp] == 0]
        result = []
        while queue:
            kp = queue.pop(0)
            result.append(kp)
            for dependent in dep_graph:
                if kp in dep_graph[dependent]:
                    in_degree[dependent] -= 1
                    if in_degree[dependent] == 0 and dependent not in result:
                        queue.append(dependent)
        return result

    def _match_best_course(self, kp_id: str, employee_id: str,
                           max_duration: int) -> dict | None:
        """为知识点匹配最优课程：考虑难度适配、时长、员工偏好"""
        candidate_courses = self.graph.get_courses_by_kp(kp_id)
        if not candidate_courses:
            return None

        employee_level = self.graph.get_employee_level(employee_id)

        # 难度适配分：课程难度与员工水平越接近越好
        scored = []
        for c in candidate_courses:
            difficulty_score = 1.0 - abs(c["difficulty"] - employee_level) / 5.0
            duration_score = 1.0 if c["duration"] <= max_duration else 0.3
            rating_score = c.get("rating", 3.5) / 5.0  # 课程评分
            total = 0.4 * difficulty_score + 0.3 * duration_score + 0.3 * rating_score
            scored.append((total, c))

        scored.sort(key=lambda x: x[0], reverse=True)
        return scored[0][1] if scored else None

5.2 第二路：协同过滤——相似员工在学什么

这一路解决"内容相关性"——找到与当前员工岗位、职级、学习历史相似的其他员工，推荐他们学过且效果好的课程。

用物品协同过滤（ItemCF）而非用户协同过滤（UserCF），原因是企业培训场景下课程数远小于员工数，ItemCF 的相似度矩阵更稳定。

import numpy as np
from collections import defaultdict

class ItemCFRecommender:
    """基于物品的协同过滤：相似课程推荐"""

    def __init__(self):
        self.course_sim = None  # 课程相似度矩阵
        self.course_user_map = None

    def fit(self, interaction_data: list[dict]):
        """
        interaction_data: [{"employee_id": "...", "course_id": "...", "score": 0.8}, ...]
        """
        # 构建课程-用户倒排表
        course_user_map = defaultdict(dict)
        for item in interaction_data:
            course_user_map[item["course_id"]][item["employee_id"]] = item["score"]

        self.course_user_map = course_user_map
        course_ids = list(course_user_map.keys())
        n = len(course_ids)

        # 计算课程间相似度
        sim_matrix = np.zeros((n, n))
        for i in range(n):
            for j in range(i + 1, n):
                # 共同学习过两门课程的员工
                users_i = set(course_user_map[course_ids[i]].keys())
                users_j = set(course_user_map[course_ids[j]].keys())
                common_users = users_i & users_j

                if not common_users:
                    continue

                # 余弦相似度，加入流行度惩罚（热门课程惩罚）
                numerator = sum(
                    course_user_map[course_ids[i]][u] * course_user_map[course_ids[j]][u]
                    for u in common_users
                )
                norm_i = np.sqrt(sum(v ** 2 for v in course_user_map[course_ids[i]].values()))
                norm_j = np.sqrt(sum(v ** 2 for v in course_user_map[course_ids[j]].values()))

                similarity = numerator / (norm_i * norm_j + 1e-8)
                # 流行度惩罚
                popularity_penalty = 1.0 / np.log1p(len(common_users))
                sim_matrix[i][j] = similarity * popularity_penalty
                sim_matrix[j][i] = sim_matrix[i][j]

        self.course_sim = sim_matrix
        self.course_ids = course_ids
        self.course_id_to_idx = {
   cid: idx for idx, cid in enumerate(course_ids)}

    def recommend(self, employee_id: str, learned_courses: list[str],
                  top_k: int = 10) -> list[dict]:
        """基于已学课程推荐相似课程"""
        scores = defaultdict(float)
        for learned_cid in learned_courses:
            if learned_cid not in self.course_id_to_idx:
                continue
            idx = self.course_id_to_idx[learned_cid]
            # 找到与已学课程最相似的候选课程
            sim_row = self.course_sim[idx]
            for j, sim in enumerate(sim_row):
                candidate_cid = self.course_ids[j]
                if candidate_cid in learned_courses:
                    continue  # 过滤已学
                scores[candidate_cid] += sim

        ranked = sorted(scores.items(), key=lambda x: x[1], reverse=True)
        return [{
   "course_id": cid, "score": score, "reason": "similar_learners"}
                for cid, score in ranked[:top_k]]

5.3 第三路：内容召回——LLM语义匹配

协同过滤有冷启动问题——新课程没有交互数据，无法被推荐。我们加了一路基于embedding的内容召回，用课程内容向量与员工学习兴趣向量做相似度匹配：

class ContentRecommender:
    """基于embedding的课程召回，解决冷启动"""

    def __init__(self, embedding_model):
        self.model = embedding_model
        self.course_embeddings = None
        self.course_ids = None

    def build_index(self, courses: list[dict]):
        """构建课程向量索引"""
        texts = [
            f"{c['title']} {c.get('description', '')} {' '.join(c.get('tags', []))}"
            for c in courses
        ]
        self.course_embeddings = np.array([self.model.embed(t) for t in texts])
        self.course_ids = [c["id"] for c in courses]

    def recommend(self, employee_interest_text: str, learned_courses: set,
                  top_k: int = 10) -> list[dict]:
        """基于员工兴趣文本召回相关课程"""
        interest_emb = self.model.embed(employee_interest_text)
        sims = np.dot(self.course_embeddings, interest_emb) / (
            np.linalg.norm(self.course_embeddings, axis=1) * np.linalg.norm(interest_emb) + 1e-8
        )

        ranked = sorted(enumerate(sims), key=lambda x: x[1], reverse=True)
        results = []
        for idx, score in ranked:
            cid = self.course_ids[idx]
            if cid in learned_courses:
                continue
            results.append({
   "course_id": cid, "score": float(score),
                           "reason": "content_match"})
            if len(results) >= top_k:
                break
        return results

5.4 融合重排

三路结果汇入重排层，按业务规则做最终排序：

class FusionRanker:
    """三路推荐结果融合重排"""

    def rank(self, path_results: list, cf_results: list,
             content_results: list, context: dict) -> list[dict]:
        all_candidates = {
   }

        # 路径规划结果权重最高（知识结构合理性优先）
        for item in path_results:
            cid = item["course"]["id"]
            all_candidates[cid] = {
   
                **item,
                "final_score": 0.5 * 1.0  # 路径权重0.5
            }

        # 协同过滤结果
        for item in cf_results:
            cid = item["course_id"]
            if cid in all_candidates:
                all_candidates[cid]["final_score"] += 0.3 * item["score"]
            else:
                all_candidates[cid] = {
   
                    **item, "final_score": 0.3 * item["score"]
                }

        # 内容召回结果
        for item in content_results:
            cid = item["course_id"]
            if cid in all_candidates:
                all_candidates[cid]["final_score"] += 0.2 * item["score"]
            else:
                all_candidates[cid] = {
   
                    **item, "final_score": 0.2 * item["score"]
                }

        ranked = sorted(all_candidates.values(),
                        key=lambda x: x["final_score"], reverse=True)

        # 业务规则后处理
        ranked = self._apply_business_rules(ranked, context)
        return ranked

    def _apply_business_rules(self, ranked: list[dict], ctx: dict) -> list[dict]:
        """业务规则：去重、强制必修课置顶、难度阶梯、多样性"""
        result = []
        seen_skills = set()

        # 必修课强制置顶
        mandatory = [r for r in ranked if r.get("is_mandatory")]
        optional = [r for r in ranked if not r.get("is_mandatory")]

        for item in mandatory + optional:
            # 多样性控制：同一技能方向不超过2门
            skill = item.get("skill", "unknown")
            if skill in seen_skills:
                # 降权但不完全过滤
                item["final_score"] *= 0.7
            else:
                seen_skills.add(skill)

            result.append(item)

        return result

六、工程落地中的几个关键决策

6.1 实时 vs 离线的权衡

推荐结果不需要毫秒级实时，但也不能T+1太慢。我们的方案是：

• 知识图谱查询：实时（Neo4j 查询延迟在10ms以内）
• 协同过滤相似度矩阵：离线计算，每天凌晨全量更新，存Redis
• 学习者画像：实时更新掌握度，但推荐时只读Redis缓存（5分钟刷新一次）
• 推荐结果：请求时实时融合计算，不预计算存储

这样做的权衡是——放弃预计算带来的极致性能，换取推荐结果的实时性。员工刚考完试，下次刷新学习中心就能看到推荐变化，这种"即时反馈"对用户体验的提升远大于几十毫秒的延迟优化。

6.2 冷启动策略

新员工没有学习历史，三路推荐里协同过滤和内容召回都受限。冷启动方案：

1. 基于岗位模板：新员工入职时根据岗位自动匹配预设的学习路径模板（管理员预先配置），作为初始推荐
2. 入职测评：新员工完成一次技能测评，根据测评结果初始化知识掌握度画像
3. 渐进式替换：随着学习行为积累，模板推荐的比例从100%逐步降低到30%，AI推荐占比逐步上升

def get_recommendation_with_cold_start(employee: dict, context: dict) -> list[dict]:
    """带冷启动策略的推荐主入口"""
    history_count = employee.get("learning_history_count", 0)

    if history_count == 0:
        # 完全冷启动：返回岗位模板路径
        return get_position_template_path(employee["position_id"])

    elif history_count < 5:
        # 冷启动阶段：70%模板 + 30%AI推荐
        template = get_position_template_path(employee["position_id"])
        ai_recs = get_ai_recommendations(employee, context)
        return merge_with_ratio(template, ai_recs, 0.7, 0.3)

    else:
        # 正常阶段：100% AI推荐
        return get_ai_recommendations(employee, context)

6.3 AB测试框架

推荐系统的迭代离不开AB测试。我们在业务层实现了一个轻量AB测试框架：

import hashlib

class ABTestFramework:
    """基于hash的AB测试分流"""

    @staticmethod
    def get_variant(employee_id: str, experiment_name: str,
                   variants: list[str]) -> str:
        """稳定分流：同一员工在同一实验中始终分到同一组"""
        key = f"{employee_id}:{experiment_name}"
        hash_val = int(hashlib.md5(key.encode()).hexdigest(), 16)
        idx = hash_val % len(variants)
        return variants[idx]

    @staticmethod
    def get_recommendations(employee: dict, context: dict) -> list[dict]:
        experiment = "learning_path_v2"
        variant = ABTestFramework.get_variant(
            employee["id"], experiment, ["control", "treatment"]
        )

        if variant == "control":
            # 对照组：旧版推荐（纯协同过滤）
            return legacy_recommend(employee, context)
        else:
            # 实验组：新版推荐（知识图谱 + 三路融合）
            return new_recommend(employee, context)

关键指标定义如下：

七、上线效果与踩过的坑

7.1 效果数据

在某客户（制造业，约3万员工）灰度上线3个月后的数据：

• 完课率从48%提升到67%，提升39%
• 考试首次通过率从59%提升到78%，提升32%
• 管理员路径维护工作量下降约70%（从手动编排改为自动生成+审核确认）
• 课程库长尾课程曝光率提升：原来Top 100课程占学习量85%，上线后降到63%，长尾课程得到更多曝光

7.2 踩过的几个坑

坑1：知识图谱的"过度依赖"导致推荐路径过长。

第一版完全依赖图谱的拓扑排序生成路径，结果一个岗位的路径推荐了28门课程——员工看到这么长的路径直接放弃了。后来加了路径长度约束（默认10门以内）和"精简模式"，优先推荐掌握度gap最大的知识点对应的课程。

坑2：协同过滤的"马太效应"。

热门课程越推越多，新课程推不出来。解决方案是在ItemCF里加了流行度惩罚（代码里的 1.0 / np.log1p(len(common_users))），同时在融合层给新课程一个冷启动加权。

坑3：掌握度评估的"考试依赖"。

早期掌握度80%权重在考试成绩上，但很多课程没有配套考试，导致这些课程的掌握度永远偏低。后来引入了"练习正确率"和"学习行为完整度"作为补充信号，降低了对考试的绝对依赖。

坑4：组织架构变动导致画像失效。

有员工调岗后，原来的岗位目标变了，但画像还带着旧岗位的学习历史，推荐结果与新岗位完全不匹配。加了组织架构变更监听事件，岗位变动时触发画像重算 + 推荐刷新。

八、总结与后续方向

这套方案的核心思路是：知识图谱保证路径的"骨架"合理性，协同过滤和内容召回填充"血肉"实现个性化，融合重排叠加业务规则做最后一道把关。 三者缺一不可——只有图谱没有个性化是"千人一面"的模板，只有个性化没有图谱是"推什么学什么"的盲目投喂。

后续我们在探索两个方向：

一是引入强化学习做长期路径优化。当前的路径规划是贪心的——每一步都选当前最优，但不一定是长期最优。如果用强化学习把"3个月后的岗位胜任力提升"作为reward，可能找到更好的路径策略。

二是LLM直接生成学习计划。当前是规则驱动的路径规划，未来想尝试用LLM结合知识图谱做自然语言层面的学习计划生成——不是推荐一个课程列表，而是生成一段"你的学习计划"自然语言描述，更像一个真人导师在给你规划。这背后需要GraphRAG把知识图谱信息注入LLM上下文，是我们在探索的另一个方向。

作者注：本文涉及的具体数据均为示意，不代表线上真实数据。架构方案为企学宝技术团队的实践总结，欢迎交流探讨。