【每日一题】推荐系统 Top-K 推荐结果的离线评估(附参考答案)

题目:推荐/搜索算法迭代时常用离线指标评估模型效果。给定 N 个用户的真实点击列表(ground truth)和模型给出的推荐列表,实现函数 evaluate,返回 (HitRate@K, MRR)。

规则:

  • 每个用户只评估模型输出的前 K 个推荐位,不足 K 个按实际长度算;
  • HitRate@K = 至少命中一个真实点击项的用户数 / 总用户数;
  • MRR = 各用户「第一个命中位置倒数」之和 / 总用户数,位置从 1 开始,未命中记 0;
  • 真实点击为空的用户按未命中处理;用户数为 0 时返回 (0.0, 0.0)。
def evaluate(topk: list, truth: list, k: int) -> tuple:
    """
    推荐系统离线评估。
    :param topk:  N 个用户的推荐 item 列表,如 [["a","b","c"], ["d","e","f"]]
    :param truth: N 个用户的真实点击 item 列表,如 [["c","x"], ["d"]]
    :param k:     只评估每个用户前 k 个推荐位
    :return:      (HitRate@k, MRR),均为 float;无用户时返回 (0.0, 0.0)
    """
    pass


if __name__ == "__main__":
    topk = [["a", "b", "c"], ["d", "e", "f"], ["x", "y", "z"]]
    truth = [["c", "q"], ["d"], ["m", "n"]]
    hr, mrr = evaluate(topk, truth, 3)
    assert round(hr, 4) == 0.6667   # 2/3 用户命中
    assert round(mrr, 4) == 0.4444  # (1/3 + 1 + 0) / 3

    assert evaluate([["a", "z"]], [["z"]], 1) == (0.0, 0.0)  # 命中位被 k 截断

    hr, mrr = evaluate([["a", "b", "c"]], [["b"]], 3)
    assert round(hr, 4) == 1.0 and round(mrr, 4) == 0.5      # 唯一命中在第 2 位

    hr, mrr = evaluate([["p", "q", "r"]], [["q", "p"]], 3)
    assert round(hr, 4) == 1.0 and round(mrr, 4) == 1.0      # 多命中取最靠前位置

    hr, mrr = evaluate([["a"], ["b"]], [[], ["b"]], 1)
    assert round(hr, 4) == 0.5 and round(mrr, 4) == 0.5      # 空真实点击按未命中算

    assert evaluate([], [], 5) == (0.0, 0.0)

难度:中级
题型:笔试(编程题)

:point_right:参考答案/答题要点:

思路一:暴力解(显式双层循环,O(N×K×L))

def evaluate_brute(topk, truth, k):
    n = len(topk)
    if n == 0:
        return (0.0, 0.0)
    hit_users = 0
    rr_sum = 0.0
    for i in range(n):
        cand = topk[i][:k]          # 只看前 k 个推荐位
        rank = 0
        for j in range(len(cand)):  # 找最靠前的命中位置
            if cand[j] in truth[i]:
                rank = j + 1
                break
        if rank:
            hit_users += 1
            rr_sum += 1.0 / rank
    return (hit_users / n, rr_sum / n)

思路二:优雅解(set + 生成器 next,O(N×K),推荐)

def evaluate(topk, truth, k):
    n = len(topk)
    if n == 0:
        return (0.0, 0.0)
    hits = 0
    rr = 0.0
    for rec, gt in zip(topk, truth):
        gt = set(gt)                # 真实点击转集合,命中判断降为 O(1)
        first = next((i + 1 for i, item in enumerate(rec[:k]) if item in gt), 0)
        if first:                   # 未命中为 0,命中位置从 1 起
            hits += 1
            rr += 1.0 / first
    return (hits / n, rr / n)

思路三:加分项·numpy 向量化(把逐用户循环换成矩阵布尔索引)

import numpy as np

def evaluate_numpy(topk, truth, k):
    n = len(topk)
    if n == 0:
        return (0.0, 0.0)
    items = {it for row in topk for it in row} | {it for row in truth for it in row}
    idx = {it: i for i, it in enumerate(sorted(items))}
    gt = np.zeros((n, max(len(idx), 1)), dtype=bool)
    for i, row in enumerate(truth):
        for it in row:
            gt[i, idx[it]] = True
    rec = np.full((n, k), -1, dtype=np.int64)   # -1 作空位哨兵
    for i, row in enumerate(topk):
        codes = [idx[it] for it in row[:k]]
        if codes:
            rec[i, :len(codes)] = codes
    hit = np.where(rec >= 0, gt[np.arange(n)[:, None], np.maximum(rec, 0)], False)
    any_hit = hit.any(axis=1)
    first = np.where(any_hit, hit.argmax(axis=1) + 1, 0)  # argmax 取首个 True 下标
    rr = np.where(any_hit, 1.0 / np.maximum(first, 1), 0.0)
    return (float(any_hit.mean()), float(rr.mean()))

以上三种实现均在 Python 3 下跑通全部 assert,并用 500 组随机数据交叉对拍结果一致(case1 输出 (0.6666666666666666, 0.4444444444444444))。

采分点:

  1. 分母用「用户总数」而不是「命中数」,未命中用户也要参与 HitRate 和 MRR 的平均;
  2. 先按 [:k] 截断再判断命中,漏掉截断属于理解错指标;
  3. 位置从 1 开始计数,取的是最靠前的命中位置,不是最后一个也不是全部;
  4. 空输入返回 (0.0, 0.0),别写 ZeroDivisionError;空真实点击按未命中计;
  5. 真实点击转 set 降低复杂度,能主动说出 O(N×K) 与推导过程;
  6. 浮点比较用 round/误差容忍断言,而不是 == 硬比;
  7. 加分:主动补「同一用户推荐位有重复 item 要不要去重」「曝光为 0 的用户是否剔除」这类口径澄清,以及指标之外还要看覆盖率、多样性、命中率变化的版本对比。