推荐系统（六）—— 序列推荐与会话建模

 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

import numpy as np
from collections import defaultdict
from typing import List

class MarkovChainRecommender:
    """一阶马尔可夫链：简单、可解释，是稳定的强基线。"""

    def __init__(self, order: int = 1):
        self.order = order
        self.transition_counts = defaultdict(lambda: defaultdict(int))
        self.context_counts = defaultdict(int)
        self.items = set()

    def fit(self, sequences: List[List[int]]):
        for seq in sequences:
            self.items.update(seq)
            for i in range(len(seq) - self.order):
                context = tuple(seq[i:i + self.order])
                next_item = seq[i + self.order]
                self.transition_counts[context][next_item] += 1
                self.context_counts[context] += 1

    def predict_next(self, sequence: List[int], top_k: int = 10):
        if len(sequence) < self.order:
            return []
        context = tuple(sequence[-self.order:])
        if context not in self.transition_counts:
            return []
        total = self.context_counts[context]
        probs = [(item, count / total)
                 for item, count in self.transition_counts[context].items()]
        probs.sort(key=lambda x: x[1], reverse=True)
        return probs[:top_k]

# --- 使用示例 ---
sequences = [
    [1, 2, 3, 4],   # 笔记本 -> 鼠标 -> 键盘 -> 显示器
    [1, 2, 5],      # 笔记本 -> 鼠标 -> 鼠标垫
    [2, 3, 4],      # 鼠标   -> 键盘 -> 显示器
    [1, 6, 7],      # 笔记本 -> 充电器 -> 电脑包
]
model = MarkovChainRecommender(order=1)
model.fit(sequences)
for item, prob in model.predict_next([1, 2], top_k=3):
    print(f"  item {item}: {prob:.3f}")
# 鼠标后面跟"键盘"出现了 2 次，所以 keyboard 排第一

 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

import torch
import torch.nn as nn
import torch.nn.functional as F
from torch.utils.data import Dataset, DataLoader

class GRU4Rec(nn.Module):
    """基于 GRU 的会话推荐模型。"""

    def __init__(self, num_items: int, embedding_dim: int = 128,
                 hidden_dim: int = 128, num_layers: int = 1, dropout: float = 0.25):
        super().__init__()
        self.num_items = num_items
        self.item_embedding = nn.Embedding(num_items + 1, embedding_dim, padding_idx=0)
        self.gru = nn.GRU(
            input_size=embedding_dim,
            hidden_size=hidden_dim,
            num_layers=num_layers,
            batch_first=True,
            dropout=dropout if num_layers > 1 else 0,
        )
        self.output = nn.Linear(hidden_dim, num_items + 1)
        self.dropout = nn.Dropout(dropout)
        nn.init.normal_(self.item_embedding.weight, mean=0, std=0.01)
        nn.init.xavier_uniform_(self.output.weight)
        nn.init.zeros_(self.output.bias)

    def forward(self, sequences, hidden=None):
        x = self.dropout(self.item_embedding(sequences))
        gru_out, hidden = self.gru(x, hidden)
        return self.output(self.dropout(gru_out)), hidden

    def predict_next(self, sequence, top_k=10):
        self.eval()
        with torch.no_grad():
            seq = torch.LongTensor([sequence]).to(next(self.parameters()).device)
            logits, _ = self(seq)
            scores, indices = torch.topk(logits[0, -1, :], k=min(top_k, self.num_items))
            return list(zip(indices.tolist(), scores.tolist()))

class SessionDataset(Dataset):
    """填充或截断到 max_len；输入去掉最后一项，目标去掉第一项。"""

    def __init__(self, sessions, max_len=20):
        self.sessions = sessions
        self.max_len = max_len

    def __len__(self):
        return len(self.sessions)

    def __getitem__(self, idx):
        s = self.sessions[idx][-self.max_len:]
        s = [0] * (self.max_len - len(s)) + s
        return torch.LongTensor(s[:-1]), torch.LongTensor(s[1:])

def train_gru4rec(model, loader, num_epochs=10, lr=1e-3, device="cpu"):
    opt = torch.optim.Adam(model.parameters(), lr=lr)
    loss_fn = nn.CrossEntropyLoss(ignore_index=0)
    model.to(device)
    for epoch in range(num_epochs):
        model.train()
        total = 0.0
        for x, y in loader:
            x, y = x.to(device), y.to(device)
            opt.zero_grad()
            logits, _ = model(x)
            loss = loss_fn(logits.reshape(-1, logits.size(-1)), y.reshape(-1))
            loss.backward()
            opt.step()
            total += loss.item()
        print(f"epoch {epoch + 1}/{num_epochs}  loss={total / len(loader):.4f}")

 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

class Caser(nn.Module):
    """基于 CNN 的序列推荐模型。"""

    def __init__(self, num_items: int, embedding_dim: int = 50,
                 max_len: int = 50, num_horizon: int = 16,
                 num_vertical: int = 8,
                 horizon_sizes: list = [2, 3, 4]):
        super().__init__()
        self.num_items = num_items
        self.max_len = max_len
        self.item_embedding = nn.Embedding(num_items + 1, embedding_dim, padding_idx=0)
        self.horizon_convs = nn.ModuleList([
            nn.Conv2d(1, num_horizon, (h, embedding_dim)) for h in horizon_sizes
        ])
        self.vertical_conv = nn.Conv2d(1, num_vertical, (max_len, 1))
        total = num_horizon * len(horizon_sizes) + num_vertical
        self.fc1 = nn.Linear(total, embedding_dim)
        self.fc2 = nn.Linear(embedding_dim, num_items + 1)
        self.dropout = nn.Dropout(0.5)

    def forward(self, sequences):
        emb = self.item_embedding(sequences).unsqueeze(1)        # (B, 1, L, D)
        h_outs = []
        for conv in self.horizon_convs:
            o = F.relu(conv(emb))
            h_outs.append(F.max_pool2d(o, (o.size(2), 1)).squeeze(-1).squeeze(-1))
        h = torch.cat(h_outs, dim=1)
        v = F.relu(self.vertical_conv(emb)).squeeze(2).squeeze(2)
        x = self.dropout(F.relu(self.fc1(torch.cat([h, v], dim=1))))
        return self.fc2(x)

 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

import math

class PositionalEncoding(nn.Module):
    def __init__(self, d_model, max_len=5000):
        super().__init__()
        pe = torch.zeros(max_len, d_model)
        pos = torch.arange(0, max_len, dtype=torch.float).unsqueeze(1)
        div = torch.exp(torch.arange(0, d_model, 2).float() * (-math.log(10000.0) / d_model))
        pe[:, 0::2] = torch.sin(pos * div)
        pe[:, 1::2] = torch.cos(pos * div)
        self.register_buffer("pe", pe.unsqueeze(0))

    def forward(self, x):
        return x + self.pe[:, : x.size(1), :]

class TransformerBlock(nn.Module):
    def __init__(self, d_model, num_heads, d_ff, dropout=0.1):
        super().__init__()
        self.attn = nn.MultiheadAttention(d_model, num_heads, dropout=dropout, batch_first=True)
        self.norm1 = nn.LayerNorm(d_model)
        self.norm2 = nn.LayerNorm(d_model)
        self.ff = nn.Sequential(
            nn.Linear(d_model, d_ff), nn.ReLU(),
            nn.Dropout(dropout),
            nn.Linear(d_ff, d_model), nn.Dropout(dropout),
        )

    def forward(self, x, mask=None):
        a, _ = self.attn(x, x, x, attn_mask=mask)
        x = self.norm1(x + a)
        x = self.norm2(x + self.ff(x))
        return x

class SASRec(nn.Module):
    def __init__(self, num_items, d_model=128, num_heads=2,
                 num_layers=2, d_ff=256, max_len=50, dropout=0.1):
        super().__init__()
        self.num_items = num_items
        self.d_model = d_model
        self.max_len = max_len
        self.item_embedding = nn.Embedding(num_items + 1, d_model, padding_idx=0)
        self.pos_encoding = PositionalEncoding(d_model, max_len)
        self.blocks = nn.ModuleList([
            TransformerBlock(d_model, num_heads, d_ff, dropout) for _ in range(num_layers)
        ])
        self.dropout = nn.Dropout(dropout)
        self.output = nn.Linear(d_model, num_items + 1)
        nn.init.normal_(self.item_embedding.weight, mean=0, std=0.01)
        nn.init.xavier_uniform_(self.output.weight)

    def forward(self, sequences):
        seq_len = sequences.size(1)
        x = self.item_embedding(sequences) * math.sqrt(self.d_model)
        x = self.dropout(self.pos_encoding(x))
        causal = torch.triu(torch.ones(seq_len, seq_len, device=sequences.device),
                            diagonal=1).bool()
        for block in self.blocks:
            x = block(x, mask=causal)
        return self.output(x)

 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

class BERT4Rec(nn.Module):
    """双向 Transformer + 完形填空式训练"""

    def __init__(self, num_items, d_model=128, num_heads=2,
                 num_layers=2, d_ff=256, max_len=50,
                 dropout=0.1, mask_prob=0.15):
        super().__init__()
        self.num_items = num_items
        self.max_len = max_len
        self.mask_prob = mask_prob
        self.mask_token = num_items + 1                                 # [MASK] id
        self.item_embedding = nn.Embedding(num_items + 2, d_model, padding_idx=0)
        self.pos_encoding = PositionalEncoding(d_model, max_len)
        self.blocks = nn.ModuleList([
            TransformerBlock(d_model, num_heads, d_ff, dropout) for _ in range(num_layers)
        ])
        self.dropout = nn.Dropout(dropout)
        self.output = nn.Linear(d_model, num_items + 1)

    def _mask_sequence(self, sequences):
        import random
        masked = sequences.clone()
        positions = torch.zeros_like(sequences, dtype=torch.bool)
        for i in range(sequences.size(0)):
            for j in range(sequences.size(1)):
                if sequences[i, j] != 0 and random.random() < self.mask_prob:
                    positions[i, j] = True
                    r = random.random()
                    if r < 0.8:
                        masked[i, j] = self.mask_token                  # 80%: [MASK]
                    elif r < 0.9:
                        masked[i, j] = random.randint(1, self.num_items)# 10%: 随机替换
                    # 10%: 保持原样
        return masked, positions

    def forward(self, sequences, training=True):
        if training:
            sequences, positions = self._mask_sequence(sequences)
        else:
            positions = torch.zeros_like(sequences, dtype=torch.bool)
        x = self.item_embedding(sequences) * math.sqrt(self.item_embedding.embedding_dim)
        x = self.dropout(self.pos_encoding(x))
        for block in self.blocks:                                       # 没有因果掩码
            x = block(x)
        return self.output(x), positions

    def predict_next(self, sequence, top_k=10):
        self.eval()
        with torch.no_grad():
            if len(sequence) >= self.max_len:
                sequence = sequence[-(self.max_len - 1):]
            sequence = sequence + [self.mask_token]
            sequence = [0] * (self.max_len - len(sequence)) + sequence
            seq = torch.LongTensor([sequence])
            logits, _ = self(seq, training=False)
            mask_pos = sequence.index(self.mask_token)
            scores, indices = torch.topk(logits[0, mask_pos, :],
                                         k=min(top_k, self.num_items))
            return list(zip(indices.tolist(), scores.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

class FeatureEmbedding(nn.Module):
    """对多个类别型字段分别做 embedding 后拼接。"""

    def __init__(self, feature_dims, embedding_dim):
        super().__init__()
        self.embeddings = nn.ModuleDict({
            name: nn.Embedding(dim + 1, embedding_dim, padding_idx=0)
            for name, dim in feature_dims.items()
        })

    def forward(self, features):
        return torch.cat([self.embeddings[n](t) for n, t in features.items()], dim=-1)

class BST(nn.Module):
    """适用于特征丰富电商场景的行为序列 Transformer。"""

    def __init__(self, item_vocab_size, feature_dims,
                 embedding_dim=64, num_heads=2, num_layers=2,
                 d_ff=256, max_len=50, dropout=0.1):
        super().__init__()
        self.max_len = max_len
        self.item_embedding = nn.Embedding(item_vocab_size + 1, embedding_dim, padding_idx=0)
        self.feature_embedding = FeatureEmbedding(feature_dims, embedding_dim)
        total = embedding_dim * (1 + len(feature_dims))
        self.pos_encoding = PositionalEncoding(total, max_len)
        self.blocks = nn.ModuleList([
            TransformerBlock(total, num_heads, d_ff, dropout) for _ in range(num_layers)
        ])
        self.dropout = nn.Dropout(dropout)
        self.mlp = nn.Sequential(
            nn.Linear(total, d_ff), nn.ReLU(), nn.Dropout(dropout),
            nn.Linear(d_ff, d_ff // 2), nn.ReLU(), nn.Dropout(dropout),
            nn.Linear(d_ff // 2, 1),
        )

    def forward(self, item_ids, features):
        x = torch.cat([self.item_embedding(item_ids),
                       self.feature_embedding(features)], dim=-1)
        x = self.dropout(self.pos_encoding(x))
        seq_len = item_ids.size(1)
        causal = torch.triu(torch.ones(seq_len, seq_len, device=item_ids.device),
                            diagonal=1).bool()
        for block in self.blocks:
            x = block(x, mask=causal)
        return self.mlp(x).squeeze(-1)

 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

class SRGNN(nn.Module):
    """会话即图的推荐模型；简化的 GGNN + 注意力汇总。"""

    def __init__(self, num_items, embedding_dim=100, num_gnn_layers=1, dropout=0.2):
        super().__init__()
        self.num_items = num_items
        self.num_gnn_layers = num_gnn_layers
        self.item_embedding = nn.Embedding(num_items + 1, embedding_dim, padding_idx=0)
        self.W_z = nn.Linear(embedding_dim, embedding_dim)
        self.U_z = nn.Linear(embedding_dim, embedding_dim)
        self.W_r = nn.Linear(embedding_dim, embedding_dim)
        self.U_r = nn.Linear(embedding_dim, embedding_dim)
        self.W_h = nn.Linear(embedding_dim, embedding_dim)
        self.U_h = nn.Linear(embedding_dim, embedding_dim)
        self.attention = nn.Sequential(
            nn.Linear(embedding_dim, embedding_dim),
            nn.ReLU(),
            nn.Linear(embedding_dim, 1),
        )
        self.output = nn.Linear(embedding_dim, num_items + 1)
        self.dropout = nn.Dropout(dropout)

    def _gnn_step(self, node_embs, adj):
        m = torch.matmul(adj, node_embs)
        z = torch.sigmoid(self.W_z(m) + self.U_z(node_embs))
        r = torch.sigmoid(self.W_r(m) + self.U_r(node_embs))
        h = torch.tanh(self.W_h(m) + self.U_h(r * node_embs))
        return (1 - z) * node_embs + z * h

    def forward_session(self, session):
        unique = list(dict.fromkeys(session))                           # 去重保序
        idx = {item: i for i, item in enumerate(unique)}
        n = len(unique)
        adj = torch.zeros(n, n)
        for a, b in zip(session, session[1:]):
            adj[idx[b], idx[a]] += 1
        adj = adj / adj.sum(dim=1, keepdim=True).clamp(min=1)

        h = self.item_embedding(torch.LongTensor(unique))
        for _ in range(self.num_gnn_layers):
            h = self.dropout(self._gnn_step(h, adj))

        seq_emb = h[[idx[i] for i in session]]
        attn = F.softmax(self.attention(seq_emb), dim=0)
        session_emb = (attn * seq_emb).sum(dim=0)
        return self.output(session_emb)

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

import numpy as np
from typing import List

def hit_rate_at_k(preds: List[List[int]], truth: List[int], k: int = 10) -> float:
    return sum(t in p[:k] for p, t in zip(preds, truth)) / len(truth)

def ndcg_at_k(preds: List[List[int]], truth: List[int], k: int = 10) -> float:
    scores = []
    for p, t in zip(preds, truth):
        if t in p[:k]:
            scores.append(1.0 / np.log2(p[:k].index(t) + 2))
        else:
            scores.append(0.0)
    return float(np.mean(scores))

def mrr(preds: List[List[int]], truth: List[int]) -> float:
    scores = [1.0 / (p.index(t) + 1) if t in p else 0.0 for p, t in zip(preds, truth)]
    return float(np.mean(scores))