推荐系统（一）—— 入门与基础概念

1
2
3
4
5
6

def predict(user, item):
    if item.rating_count < MIN_RATINGS:
        return content_based_predict(user, item)
    if user.rating_count < MIN_USER_RATINGS:
        return content_based_predict(user, item)
    return collaborative_filtering_predict(user, item)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79

import numpy as np
from typing import List, Tuple

class MatrixFactorization:
    """带用户/物品偏置的矩阵分解，使用 SGD 训练。"""

    def __init__(
        self,
        n_users: int,
        n_items: int,
        n_factors: int = 20,
        learning_rate: float = 0.01,
        regularization: float = 0.02,
        n_epochs: int = 20,
    ):
        self.n_users = n_users
        self.n_items = n_items
        self.n_factors = n_factors
        self.lr = learning_rate
        self.reg = regularization
        self.n_epochs = n_epochs

        # 隐因子用小随机数初始化
        rng = np.random.default_rng(0)
        self.P = rng.normal(0, 0.1, (n_users, n_factors))
        self.Q = rng.normal(0, 0.1, (n_items, n_factors))

        self.user_bias = np.zeros(n_users)
        self.item_bias = np.zeros(n_items)
        self.global_mean = 0.0

    def fit(self, ratings: List[Tuple[int, int, float]]) -> "MatrixFactorization":
        self.global_mean = float(np.mean([r for _, _, r in ratings]))
        for epoch in range(self.n_epochs):
            np.random.shuffle(ratings)
            sse = 0.0
            for u, i, r in ratings:
                pred = self._predict(u, i)
                err = r - pred
                sse += err ** 2

                p_u, q_i = self.P[u].copy(), self.Q[i].copy()
                self.P[u] += self.lr * (err * q_i - self.reg * p_u)
                self.Q[i] += self.lr * (err * p_u - self.reg * q_i)
                self.user_bias[u] += self.lr * (err - self.reg * self.user_bias[u])
                self.item_bias[i] += self.lr * (err - self.reg * self.item_bias[i])

            if (epoch + 1) % 5 == 0:
                rmse = np.sqrt(sse / len(ratings))
                print(f"epoch {epoch + 1:3d}  RMSE = {rmse:.4f}")
        return self

    def _predict(self, u: int, i: int) -> float:
        return (
            self.global_mean
            + self.user_bias[u]
            + self.item_bias[i]
            + self.P[u] @ self.Q[i]
        )

    def recommend(
        self, u: int, top_n: int = 10, exclude: set | None = None
    ) -> List[Tuple[int, float]]:
        exclude = exclude or set()
        scores = [(i, self._predict(u, i)) for i in range(self.n_items) if i not in exclude]
        scores.sort(key=lambda x: x[1], reverse=True)
        return scores[:top_n]

if __name__ == "__main__":
    ratings = [
        (0, 0, 5.0), (0, 1, 3.0), (0, 2, 2.5),
        (1, 0, 4.0), (1, 2, 4.5), (1, 3, 3.0),
        (2, 1, 3.5), (2, 2, 4.0), (2, 4, 2.0),
        (3, 0, 3.0), (3, 3, 4.0), (3, 4, 4.5),
        (4, 1, 4.0), (4, 2, 3.5), (4, 3, 5.0),
    ]
    model = MatrixFactorization(n_users=5, n_items=5, n_factors=10, n_epochs=50)
    model.fit(ratings)
    print("\n用户 0 的 Top-3:", model.recommend(0, top_n=3, exclude={0, 1, 2}))

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44

import numpy as np
from typing import List

def precision_at_k(ranked: List[int], relevant: set, k: int) -> float:
    if k == 0:
        return 0.0
    hits = sum(1 for i in ranked[:k] if i in relevant)
    return hits / k

def recall_at_k(ranked: List[int], relevant: set, k: int) -> float:
    if not relevant:
        return 0.0
    hits = sum(1 for i in ranked[:k] if i in relevant)
    return hits / len(relevant)

def average_precision(ranked: List[int], relevant: set, k: int) -> float:
    """AP@K：相关物品越靠前得分越高。"""
    if not relevant:
        return 0.0
    score, hits = 0.0, 0
    for rank, item in enumerate(ranked[:k], start=1):
        if item in relevant:
            hits += 1
            score += hits / rank
    return score / min(len(relevant), k)

def ndcg_at_k(true_relevances: List[float], k: int) -> float:
    """NDCG@K：支持分级相关性（评分）。"""
    rel = np.asarray(true_relevances[:k], dtype=float)
    if rel.size == 0:
        return 0.0
    discounts = np.log2(np.arange(2, rel.size + 2))
    dcg = np.sum((2 ** rel - 1) / discounts)
    ideal = np.sort(np.asarray(true_relevances, dtype=float))[::-1][:k]
    idcg = np.sum((2 ** ideal - 1) / discounts[: ideal.size])
    return float(dcg / idcg) if idcg else 0.0

if __name__ == "__main__":
    ranked = [5, 2, 8, 1, 9, 3, 7, 4, 6, 10]
    relevant = {1, 2, 5, 7}
    print(f"Precision@5  = {precision_at_k(ranked, relevant, 5):.3f}")
    print(f"Recall@5     = {recall_at_k(ranked, relevant, 5):.3f}")
    print(f"AP@10        = {average_precision(ranked, relevant, 10):.3f}")
    print(f"NDCG@10      = {ndcg_at_k([5,5,2,5,1,3,4,2,1,1], 10):.3f}")

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17

class CandidateGenerator:
    def __init__(self):
        self.user_cf = UserCFRecall()
        self.item_cf = ItemCFRecall()
        self.mf      = MatrixFactorizationRecall()
        self.popular = PopularItemsRecall()
        self.content = ContentBasedRecall()

    def generate(self, user_id: int, n: int = 2000) -> list[int]:
        candidates: set[int] = set()
        candidates.update(self.user_cf.recall(user_id, n=500))
        candidates.update(self.item_cf.recall(user_id, n=500))
        candidates.update(self.mf.recall(user_id, n=500))
        candidates.update(self.popular.recall(user_id, n=200))
        candidates.update(self.content.recall(user_id, n=300))
        candidates -= self.history(user_id)
        return list(candidates)[:n]

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33

import torch
import torch.nn as nn

class WideAndDeep(nn.Module):
    """Wide & Deep 排序模型——线性记忆 + MLP 泛化。"""

    def __init__(
        self,
        n_wide_features: int,
        embedding_sizes: list[int],
        embedding_dim: int = 16,
        deep_layers: tuple[int, ...] = (128, 64, 32),
    ):
        super().__init__()
        self.wide = nn.Linear(n_wide_features, 1)
        self.embeddings = nn.ModuleList(
            [nn.Embedding(n, embedding_dim) for n in embedding_sizes]
        )

        layers, in_dim = [], len(embedding_sizes) * embedding_dim
        for hidden in deep_layers:
            layers += [nn.Linear(in_dim, hidden), nn.ReLU(), nn.Dropout(0.3)]
            in_dim = hidden
        self.deep = nn.Sequential(*layers)
        self.out = nn.Linear(deep_layers[-1] + 1, 1)

    def forward(self, wide_x, emb_idx):
        wide = self.wide(wide_x)
        deep_in = torch.cat(
            [emb(emb_idx[:, i]) for i, emb in enumerate(self.embeddings)], dim=1
        )
        deep = self.deep(deep_in)
        return torch.sigmoid(self.out(torch.cat([wide, deep], dim=1))).squeeze()

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21

import numpy as np

def maximal_marginal_relevance(
    candidates: list,
    scores: np.ndarray,
    similarity: np.ndarray,
    top_n: int,
    lam: float = 0.5,
) -> list:
    """MMR：平衡相关性与多样性。lam=1 表示纯相关性。"""
    selected = [int(np.argmax(scores))]
    remaining = [i for i in range(len(candidates)) if i != selected[0]]
    while len(selected) < top_n and remaining:
        mmr = [
            lam * scores[i] - (1 - lam) * max(similarity[i, j] for j in selected)
            for i in remaining
        ]
        pick = remaining[int(np.argmax(mmr))]
        selected.append(pick)
        remaining.remove(pick)
    return [candidates[i] for i in selected]

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22

import numpy as np

def epsilon_greedy_recommend(
    candidates, scores, item_counts, *, epsilon=0.1, top_n=10, threshold=10
):
    """ε-贪心：Top-N 大部分用于利用，小部分用于探索。"""
    candidates = np.asarray(candidates)
    scores = np.asarray(scores)
    cold = np.array([item_counts.get(c, 0) < threshold for c in candidates])

    n_explore = int(round(top_n * epsilon))
    n_exploit = top_n - n_explore

    exploit = candidates[~cold][np.argsort(-scores[~cold])][:n_exploit]
    explore_pool = candidates[cold]
    explore = np.random.choice(
        explore_pool, size=min(n_explore, len(explore_pool)), replace=False
    ) if len(explore_pool) else np.array([], dtype=candidates.dtype)

    out = np.concatenate([exploit, explore])
    np.random.shuffle(out)
    return out[:top_n].tolist()

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66

import numpy as np
from collections import defaultdict
from typing import Dict, List, Tuple

class UserBasedCF:
    """基于皮尔逊相似度和均值中心化预测的 User-CF。"""

    def __init__(self, k_neighbors: int = 50, min_common_items: int = 3,
                 similarity: str = "pearson"):
        self.k = k_neighbors
        self.min_common = min_common_items
        self.similarity = similarity

        self.user_ratings: Dict[int, Dict[int, float]] = defaultdict(dict)
        self.item_users:   Dict[int, set]              = defaultdict(set)
        self.user_means:   Dict[int, float]            = {}
        self.global_mean = 0.0

    def fit(self, ratings: List[Tuple[int, int, float]]) -> "UserBasedCF":
        all_r = []
        for u, i, r in ratings:
            self.user_ratings[u][i] = r
            self.item_users[i].add(u)
            all_r.append(r)
        self.global_mean = float(np.mean(all_r))
        for u, items in self.user_ratings.items():
            self.user_means[u] = float(np.mean(list(items.values())))
        return self

    def _sim(self, u: int, v: int) -> float:
        common = set(self.user_ratings[u]) & set(self.user_ratings[v])
        if len(common) < self.min_common:
            return 0.0
        r1 = np.array([self.user_ratings[u][i] for i in common])
        r2 = np.array([self.user_ratings[v][i] for i in common])
        if self.similarity == "pearson":
            r1 = r1 - self.user_means[u]
            r2 = r2 - self.user_means[v]
        n1, n2 = np.linalg.norm(r1), np.linalg.norm(r2)
        return float(r1 @ r2 / (n1 * n2)) if n1 and n2 else 0.0

    def predict(self, u: int, i: int) -> float:
        if u not in self.user_ratings:
            return self.global_mean
        if i not in self.item_users:
            return self.user_means.get(u, self.global_mean)
        if i in self.user_ratings[u]:
            return self.user_ratings[u][i]

        sims = [(v, self._sim(u, v)) for v in self.item_users[i] if v != u]
        sims = sorted((s for s in sims if s[1] > 0), key=lambda x: -x[1])[: self.k]
        if not sims:
            return self.user_means[u]

        weighted = sum(s * (self.user_ratings[v][i] - self.user_means[v]) for v, s in sims)
        denom    = sum(abs(s) for _, s in sims)
        return self.user_means[u] + weighted / denom if denom else self.user_means[u]

    def recommend(self, u: int, n: int = 10, exclude_rated: bool = True
                  ) -> List[Tuple[int, float]]:
        if u not in self.user_ratings:
            popular = sorted(self.item_users.items(), key=lambda x: -len(x[1]))
            return [(i, self.global_mean) for i, _ in popular[:n]]
        rated = set(self.user_ratings[u]) if exclude_rated else set()
        scores = [(i, self.predict(u, i)) for i in self.item_users if i not in rated]
        return sorted(scores, key=lambda x: -x[1])[:n]

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55

class ItemBasedCF:
    """带相似度缓存的、使用调整余弦相似度的 Item-CF。"""

    def __init__(self, k_neighbors: int = 50, min_common_users: int = 3):
        self.k = k_neighbors
        self.min_common = min_common_users
        self.item_ratings: Dict[int, Dict[int, float]] = defaultdict(dict)
        self.user_items:   Dict[int, set]              = defaultdict(set)
        self.user_means:   Dict[int, float]            = {}
        self.sim_cache:    Dict[Tuple[int, int], float] = {}
        self.global_mean = 0.0

    def fit(self, ratings: List[Tuple[int, int, float]]) -> "ItemBasedCF":
        all_r = []
        for u, i, r in ratings:
            self.item_ratings[i][u] = r
            self.user_items[u].add(i)
            all_r.append(r)
        self.global_mean = float(np.mean(all_r))
        for u, items in self.user_items.items():
            self.user_means[u] = float(np.mean([self.item_ratings[i][u] for i in items]))
        return self

    def _sim(self, a: int, b: int) -> float:
        key = (min(a, b), max(a, b))
        if key in self.sim_cache:
            return self.sim_cache[key]
        common = set(self.item_ratings[a]) & set(self.item_ratings[b])
        if len(common) < self.min_common:
            self.sim_cache[key] = 0.0
            return 0.0
        r1 = np.array([self.item_ratings[a][u] - self.user_means[u] for u in common])
        r2 = np.array([self.item_ratings[b][u] - self.user_means[u] for u in common])
        n1, n2 = np.linalg.norm(r1), np.linalg.norm(r2)
        s = float(r1 @ r2 / (n1 * n2)) if n1 and n2 else 0.0
        self.sim_cache[key] = s
        return s

    def predict(self, u: int, i: int) -> float:
        if u not in self.user_items:
            return self.global_mean
        if i not in self.item_ratings:
            return self.user_means.get(u, self.global_mean)
        sims = [(j, self._sim(i, j)) for j in self.user_items[u] if j != i]
        sims = sorted((s for s in sims if s[1] > 0), key=lambda x: -x[1])[: self.k]
        if not sims:
            return self.user_means.get(u, self.global_mean)
        num = sum(s * self.item_ratings[j][u] for j, s in sims)
        den = sum(abs(s) for _, s in sims)
        return num / den if den else self.user_means.get(u, self.global_mean)

    def similar_items(self, i: int, n: int = 10) -> List[Tuple[int, float]]:
        others = [j for j in self.item_ratings if j != i]
        sims = [(j, self._sim(i, j)) for j in others]
        return sorted(sims, key=lambda x: -x[1])[:n]