推荐系统召回-排序-重排架构实战

一、项目概述

1.1 项目背景

短视频平台的内容推荐系统是用户留存和变现的核心引擎。当用户在刷"下一条"时，后台需要在 100ms 以内从百万级内容池中选出最可能让用户感兴趣的几十条视频。这个过程必须在用户感知不到延迟的情况下完成——稍有卡顿，用户就会划走，而划走即流失。

项目初期，平台日活（DAU）已达到 3000 万，内容池超过 200 万条视频，日均推荐调用量超过 8 亿次。当时的推荐系统存在严重问题：基于规则的协同过滤无法捕捉用户的实时兴趣变化，热门内容霸屏导致长尾内容曝光不足，新用户冷启动依赖人工配置的标签规则，点击率长期徘徊在 11% 左右，用户平均停留时长仅为 35 分钟。

1.2 核心矛盾：候选集太大 vs 精排代价太高

推荐系统的本质是一个信息检索与排序问题。百万级内容池意味着任何时刻都有百万个候选等待被评估，而精排模型（深度神经网络）单次推理耗时约 10-20ms，如果对百万内容全部做精排，仅模型推理就需要消耗 2.7 小时——这显然不可接受。

核心矛盾在于：精排模型的精度最高，但计算代价也最高；我们既需要模型的精度来提升推荐质量，又无法承受全量精排的代价。这形成了推荐系统最经典的设计范式——漏斗架构：用计算代价低、精度较低的方式逐步缩小候选集，最后只在极小集合上运行重型模型。

1.3 漏斗架构总体设计

系统采用四级漏斗架构，每一级逐步缩小候选集规模，同时逐步提升候选质量：

1.4 量化指标

• DAU：峰值 3500 万，稳定在千万级别
• 推荐响应 P99：端到端 < 100ms（包含网络开销）
• 推荐点击率：从 11% 提升至 15.2%（提升 38%）
• 用户平均停留时长：从 35 分钟提升至 48 分钟（提升 37%）
• 系统吞吐量：日均推荐调用 12 亿次，峰值 QPS 稳定在 15 万

二、技术架构设计

2.1 全链路漏斗流程

从用户发起请求到推荐结果返回，完整链路经过六个核心模块，每层都有明确的 QPS 目标和职责边界：

图：推荐系统全链路流水线，各模块 QPS 逐级递减

2.2 召回层多路召回设计

召回层是漏斗的第一级，也是最关键的一级。召回的质量直接决定了推荐系统的上限——如果好的内容根本没被召回，后续的排序和重排都是在矮子里选将军。

我们采用多路召回策略，并行执行多路独立的召回通道，每路召回解决不同维度的问题，最终合并去重：

2.3 在线/离线分离策略

不同排序阶段对延迟和精度的要求不同，因此采用差异化的在线/离线策略：

2.4 实时特征工程架构

用户刚刚点了猫视频，下一条就想看到更多猫——这就是实时特征的价值。整个实时特征更新的链路如下：

2.5 技术选型

三、核心技术挑战与解决方案

四、关键技术实现

4.1 多路召回合并策略（分数归一化 + 加权求和）

多路召回的核心挑战是：各路召回的分数分布不同，不能直接比较。例如协同过滤召回的分数范围可能是 [0, 1]，而向量召回的内积分数范围可能是 [-10, 100]。如果不进行归化就直接加权求和，低分路的召回结果会完全被高分路淹没。

// 多路召回合并：分数归一化 + 加权求和
public class MultiWayRecallMerger {

    // 各路召回的权重配置（通过Apollo动态调整）
    private static final Map RECALL_WEIGHTS = Map.of(
        "cf_recall",     0.30,  // 协同过滤召回
        "vector_recall",  0.40,  // 向量召回（最重要）
        "hot_recall",     0.15,  // 热度召回
        "geo_recall",     0.15   // 地理召回
    );

    public List merge(List<RecallResult>[] recallResults) {
        // Step 1: 遍历各路召回结果，分别归一化
        for (List<RecallResult> results : recallResults) {
            if (results.isEmpty()) continue;

            double minScore = results.stream().mapToDouble(r -> r.score).min().orElse(0);
            double maxScore = results.stream().mapToDouble(r -> r.score).max().orElse(1);
            double range = maxScore - minScore;

            // Min-Max归一化到 [0, 1]
            for (RecallResult r : results) {
                r.normalizedScore = (range > 0) ? (r.score - minScore) / range : 0.5;
            }
        }

        // Step 2: 合并去重（同一item可能在多路召回中出现）
        Map<String, MergedRecallItem> merged = new HashMap<>();
        for (List<RecallResult> results : recallResults) {
            for (RecallResult r : results) {
                String key = r.itemId;
                if (!merged.containsKey(key)) {
                    merged.put(key, new MergedRecallItem(r.itemId, r.itemScore));
                }
                MergedRecallItem m = merged.get(key);
                m.weightedScore += r.normalizedScore * RECALL_WEIGHTS.getOrDefault(r.source, 0.25);
                m.sourceCount++;
            }
        }

        // Step 3: 综合得分 = 加权得分 * 来源多样性奖励
        List<MergedRecallItem> finalList = merged.values().stream()
            .peek(m -> {
                // 多样性奖励：来自更多路召回的item得分更高
                double diversityBonus = 1 + 0.1 * Math.min(m.sourceCount - 1, 3);
                m.finalScore = m.weightedScore * diversityBonus;
            })
            .sorted((a, b) -> Double.compare(b.finalScore, a.finalScore))
            .limit(5000)  // 召回层输出：5000条
            .collect(Collectors.toList());

        return finalList;
    }
}

4.2 Faiss向量检索Java实现

使用Faiss的IVFFlat索引实现高效的近似最近邻检索：

// Faiss向量检索Java实现（使用JNA调用native库）
public class FaissVectorEngine {

    static { System.loadLibrary("faiss_jni"); }  // 加载Faiss JNI库

    private long indexPtr;  // Faiss Index指针

    /**
     * 构建IVFFlat索引（倒排文件索引）
     * nlist: 聚类中心数量（影响召回率和内存占用）
     */
    public void buildIndex(float[][] vectors, int nlist) throws Exception {
        int dimension = vectors[0].length;
        int totalVectors = vectors.length;

        // Step 1: 计算数据归一化（L2归一化，使余弦相似度=内积）
        for (float[] v : vectors) {
            float norm = 0;
            for (float f : v) norm += f * f;
            norm = (float) Math.sqrt(norm);
            if (norm > 0) {
                for (int i = 0; i < v.length; i++) v[i] /= norm;
            }
        }

        // Step 2: 训练量化器（k-means聚类，确定倒排列表）
        long quantizerPtr = Native.faiss_knn_center(dvectors, nlist);
        
        // Step 3: 构建IVFFlat索引
        this.indexPtr = Native.faiss_ivfflat_create(dimension, nlist, faiss_flag);
        Native.faiss_ivfflat_train(indexPtr, vectors, totalVectors);
        Native.faiss_ivfflat_add(indexPtr, vectors, totalVectors);
        
        // Step 4: 设置nprobe（搜索的聚类中心数，影响召回率和延迟）
        Native.faiss_ivfflat_set_nprobe(indexPtr, 64);
    }

    /**
     * 查询Top-K近邻
     * @param queryVector 查询向量（embedding）
     * @param topK 返回的近邻数量
     * @return 近邻ID列表和距离
     */
    public SearchResult search(float[] queryVector, int topK) {
        // 归一化查询向量
        float[] normalizedQuery = normalize(queryVector);

        // 分配结果数组
        long[] ids = new long[topK];
        float[] distances = new float[topK];

        // 执行搜索
        Native.faiss_ivfflat_search(indexPtr, normalizedQuery, topK, ids, distances);

        List<SearchResult.Item> items = new ArrayList<>();
        for (int i = 0; i < topK; i++) {
            if (ids[i] >= 0) {  // -1表示无效结果
                items.add(new SearchResult.Item((int)ids[i], distances[i]));
            }
        }
        return new SearchResult(items);
    }

    private float[] normalize(float[] v) {
        float norm = 0;
        for (float f : v) norm += f * f;
        norm = (float) Math.sqrt(norm);
        if (norm == 0) return v;
        float[] r = new float[v.length];
        for (int i = 0; i < v.length; i++) r[i] = v[i] / norm;
        return r;
    }
}

4.3 Flink实时特征更新

用户行为事件通过Flink流处理，秒级更新Redis中的特征：

// Flink实时特征更新（消费Kafka行为事件 → 更新Redis）
public class FlinkFeatureEngine {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(32);
        env.enableCheckpointing(5000, CheckpointingMode.EXACTLY_ONCE);

        // 消费Kafka行为事件
        DataStream<UserBehavior> events = env
            .addSource(new FlinkKafkaConsumer<>(
                "user_behavior_events",
                new UserBehaviorDeserializer(),
                kafkaProps
            ))
            .uid("kafka-source")
            .name("Kafka行为事件源");

        // 实时特征更新
        events
            .keyBy(e -> e.userId)  // 按用户ID分桶
            .process(new FeatureUpdateProcessFunction())
            .uid("feature-update")
            .name("实时特征更新")
            .addSink(new RedisSink<>(redisConfig));
    }
}

public class FeatureUpdateProcessFunction 
    extends KeyedProcessFunction<String, UserBehavior, FeatureUpdate> {

    private transient RedisCommands<String, String> redis;

    @Override
    public void open(Configuration params) {
        JedisPoolConfig poolConfig = new JedisPoolConfig();
        poolConfig.setMinIdle(50);
        poolConfig.setMaxTotal(200);
        JedisPooled<String> jedis = new JedisPooled<>(poolConfig);
        this.redis = jedis.connect(new RedisOptions()).sync();
    }

    @Override
    public void processElement(UserBehavior event, Context ctx, Collector<FeatureUpdate> out) {
        String featureKey = "feature:user:" + event.userId + ":" + event.featureType;

        // 根据行为类型更新对应特征
        switch (event.behaviorType) {
            case "CLICK":
                // 累计点击次数
                redis.hincrBy(featureKey, "click_count", 1);
                redis.expire(featureKey, 86400);  // 24h过期
                // 记录最近点击时间
                redis.hset(featureKey, "last_click_time", String.valueOf(event.timestamp));
                break;

            case "VIEW_DETAIL":
                // 累计浏览次数（同一商品重复浏览也计数）
                redis.hincrBy(featureKey, "detail_view_count:" + event.itemId, 1);
                break;

            case "ADD_CART":
                // 标记加购意图
                redis.sadd("intent:cart:" + event.userId, event.itemId);
                break;
        }

        // 输出特征更新事件（供下游使用）
        out.collect(new FeatureUpdate(
            event.userId,
            event.featureType,
            event.itemId,
            event.timestamp
        ));
    }
}

五、性能指标与成果

5.1 核心业务指标

5.2 各层性能数据