基于用户画像的skill推荐系统实战：从算法选型到性能优化

1次阅读

没有评论

共计 2997 个字符，预计需要花费 8 分钟才能阅读完成。

在构建 skill 推荐系统时，我们遇到了几个特有的挑战：

数据稀疏性：用户 - 技能交互矩阵极度稀疏（平均每个用户仅标注 3 - 5 个技能），传统协同过滤效果受限
冷启动问题：新上线技能缺少用户行为数据，内容特征提取困难
实时性要求：用户期望在个人主页加载时（<200ms）获得实时推荐结果
技能关联性：编程语言与框架之间存在隐式关联（如 Python→Django），需要捕捉高阶关系

我们对比了三种主流推荐算法在 skill 场景的表现：

协同过滤（CF）
优势：能发现用户潜在兴趣（用户 A 会 Java→可能喜欢 Spring）
劣势：无法处理新技能，依赖密集用户行为数据
内容推荐（CB）
优势：利用技能描述文本（TF-IDF/BERT 嵌入）解决冷启动
劣势：难以捕捉 ”Python→数据分析 ” 这类跨领域关联
图神经网络（GNN）
优势：通过用户 - 技能二部图捕捉高阶关系
劣势：线上服务计算开销大，实时响应困难

最终选择 混合推荐架构：用矩阵分解处理显式反馈，注意力机制融合技能文本特征。

使用 TensorFlow 实现带注意力机制的 DMF 模型：

class DMF(tf.keras.Model):
    def __init__(self, user_dim=64, skill_dim=64):
        super().__init__()
        self.user_encoder = tf.keras.Sequential([layers.Dense(128, activation='relu'),
            layers.Dense(user_dim)  # 用户潜在向量
        ])
        self.skill_encoder = tf.keras.Sequential([layers.Dense(128, activation='relu'),
            layers.Dense(skill_dim)  # 技能潜在向量
        ])
        self.attention = layers.Attention()  # 交叉特征注意力层

    def call(self, inputs):
        user_emb = self.user_encoder(inputs["user_features"])
        skill_emb = self.skill_encoder(inputs["skill_features"])
        return tf.reduce_sum(user_emb * self.attention([user_emb, skill_emb]), axis=1)

用 PySpark 构建用户 - 技能二部图特征：

from pyspark.sql import functions as F

def build_interaction_features(df):
    # 用户行为加权（浏览 1 分，收藏 3 分，认证 5 分）weighted_df = df.withColumn(
        "weight",
        F.when(F.col("action_type") == "view", 1)
         .when(F.col("action_type") == "favorite", 3)
         .otherwise(5)
    )

    # 生成协同过滤特征
    cf_features = weighted_df.groupBy("user_id", "skill_id")
        .agg(F.sum("weight").alias("interaction_score"))
        .persist()

    # 计算技能共现矩阵（用于后续图特征）co_occurrence = cf_features.join(cf_features.alias("other"),
        on="user_id"
    ).filter("skill_id != other.skill_id")

    return {
        "interaction": cf_features,
        "co_occurrence": co_occurrence
    }

Flask API 集成 Redis 缓存的关键代码：

app = Flask(__name__)
redis = RedisCluster(startup_nodes=[{"host": "redis-node1", "port": 6379}],
    decode_responses=True
)

@app.route("/recommend", methods=["POST"])
def recommend():
    # 1. 从 Redis 获取用户最新特征
    user_id = request.json["user_id"]
    user_feature = redis.get(f"user:{user_id}:features")

    if not user_feature:
        # 2. 回退到离线预计算特征
        user_feature = get_offline_features(user_id)

    # 3. 实时推理（<50ms）recommendations = model.predict(user_feature)
    return jsonify(recommendations)

离线层：每天 0 点用 Spark 预计算用户特征，写入 HBase
近线层：用户行为事件触发 Flink 实时更新 Redis 缓存
在线层：API 优先读取 Redis，未命中时查询 HBase

通过 Locust 模拟 2000 并发请求时发现：

GC 问题：默认 JVM 参数下，GC 停顿导致 P99 延迟达 320ms
优化方案：
启用 G1 垃圾回收器：-XX:+UseG1GC
限制 Young 区大小：-Xmn512m
调整最大 GC 停顿：-XX:MaxGCPauseMillis=100
效果：P99 延迟降至 85ms，吞吐量提升 3 倍

避免特征穿越的三种策略：

时间分割：严格按事件时间划分训练 / 测试集
特征隔离：实验组 / 对照组使用独立特征管道
版本快照：模型训练时冻结特征版本

技能相似度矩阵的更新流程：

每小时跑增量 Job 处理新用户行为
更新时采用 Double-Buffer 模式：
当前版本 v1 服务线上
异步构建 v2 版本
通过原子切换完成更新

所有生产代码必须包含：

类型注解（Python 3.8+）：

def calculate_similarity(
    skill_a: str, 
    skill_b: str
) -> float:
    """计算两个技能的余弦相似度"""

单元测试（pytest）：

def test_similarity_calculation():
    # 已知 Python 和 Django 应具有高相似度
    assert calculate_similarity("Python", "Django") > 0.7
    # 不相关技能得分应接近 0
    assert calculate_similarity("Photoshop", "Kubernetes") < 0.1

建议尝试将 BERT 应用于技能语义理解：