🔄 第9课：序列推荐（GRU4Rec）

🎯 学习目标

🗂️ 课程定位

📖 核心理论

序列推荐（Sequential Recommendation）捕捉用户行为的时间顺序模式。与传统方法将用户历史视为无序集合不同，序列推荐认为用户的下一个行为与最近的行为序列密切相关。

GRU4Rec是第一个将RNN应用于会话推荐的模型，用GRU处理用户点击序列，预测下一个点击。后续的SASRec用Self-Attention替代GRU，BERT4Rec用双向Transformer。

📐 数学公式

💻 代码实现

import numpy as np
np.random.seed(42)
print("="*60+"\n序列推荐模型(GRU4Rec)\n"+"="*60)
n_users=50;n_items=30;seq_len=5
# 生成用户行为序列
item_emb=np.random.randn(n_items,8)
sequences=[];labels=[]
for u in range(n_users):
    seq_len_u=np.random.randint(3,8)
    start=np.random.randint(n_items)
    seq=[start]
    for s in range(seq_len_u-1):
        last=seq[-1]
        sim=item_emb[last]@item_emb.T
        sim[last]=-999
        next_item=np.argmax(sim+np.random.randn(n_items)*0.5)
        seq.append(next_item)
    if len(seq)>=3:
        sequences.append(seq[:-1])
        labels.append(seq[-1])
print(f"用户序列数:{len(sequences)} 物品数:{n_items}")
print(f"平均序列长度:{np.mean([len(s)for s in sequences]):.1f}")
# GRU单元
def sigmoid(x):return 1/(1+np.exp(-np.clip(x,-500,500)))
def tanh(x):return np.tanh(x)
class SimpleGRU:
    def __init__(self,input_size,hidden_size):
        self.hidden_size=hidden_size
        scale=0.01
        self.Wz=np.random.randn(input_size+hidden_size,hidden_size)*scale
        self.Wr=np.random.randn(input_size+hidden_size,hidden_size)*scale
        self.Wh=np.random.randn(input_size+hidden_size,hidden_size)*scale
        self.bz=np.zeros(hidden_size)
        self.br=np.zeros(hidden_size)
        self.bh=np.zeros(hidden_size)
    def forward(self,x,h):
        combined=np.concatenate([x,h])
        z=sigmoid(combined@self.Wz+self.bz)
        r=sigmoid(combined@self.Wr+self.br)
        combined_r=np.concatenate([x,r*h])
        h_tilde=tanh(combined_r@self.Wh+self.bh)
        h_new=(1-z)*h+z*h_tilde
        return h_new
    def forward_seq(self,x_seq):
        h=np.zeros(self.hidden_size)
        for x in x_seq:h=self.forward(x,h)
        return h
# GRU4Rec模型
emb_size=8;hidden_size=16
item_embeddings=np.random.randn(n_items,emb_size)*0.1
gru=SimpleGRU(emb_size,hidden_size)
output_W=np.random.randn(hidden_size,n_items)*0.01
output_b=np.zeros(n_items)
# 训练
print("\n--- 训练GRU4Rec ---")
lr=0.005
for ep in range(50):
    total_loss=0;correct=0
    indices=np.random.permutation(len(sequences))
    for idx in indices:
        seq=sequences[idx];target=labels[idx]
        x_seq=[item_embeddings[s]for s in seq]
        h=gru.forward_seq(x_seq)
        logits=h@output_W+output_b
        # Softmax + NLL
        logits-=logits.max()
        exp_logits=np.exp(logits)
        probs=exp_logits/exp_logits.sum()
        loss=-np.log(probs[target]+1e-10)
        total_loss+=loss
        pred=np.argmax(probs)
        if pred==target:correct+=1
        # 简化梯度更新
        grad=probs.copy();grad[target]-=1
        dW=np.outer(h,grad)
        output_W-=lr*dW;output_b-=lr*grad
        dh=grad@output_W.T
        # 近似更新embedding
        for s in seq:
            item_embeddings[s]-=lr*0.1*dh[:emb_size]
    if(ep+1)%10==0:
        acc=correct/len(sequences)
        print(f"  Epoch{ep+1}:Loss={total_loss/len(sequences):.4f} Acc={acc:.4f}")
# 评估
print("\n--- 评估 ---")
hits=0;mrr=0
for seq,target in zip(sequences,labels):
    x_seq=[item_embeddings[s]for s in seq]
    h=gru.forward_seq(x_seq)
    logits=h@output_W+output_b
    top10=np.argsort(logits)[-10:][::-1]
    if target in top10:hits+=1;mrr+=1/(np.where(top10==target)[0][0]+1)
hr10=hits/len(sequences);mrr10=mrr/len(sequences)
print(f"HitRate@10={hr10:.4f} MRR@10={mrr10:.4f}")
# 序列模式分析
print("\n序列模式分析:")
for u in range(5):
    print(f"  用户{u}:序列{sequences[u]} → 预测{labels[u]}")
# GRU vs 简单最近邻
print("\n--- 对比: GRU4Rec vs 最近邻 ---")
nn_hits=0
for seq,target in zip(sequences,labels):
    last=seq[-1]
    sim=item_emb[last]@item_emb.T;sim[last]=-999
    top10=np.argsort(sim)[-10:][::-1]
    if target in top10:nn_hits+=1
nn_hr10=nn_hits/len(sequences)
print(f"最近邻 HR@10={nn_hr10:.4f}")
print(f"GRU4Rec HR@10={hr10:.4f}")
print(f"提升:{(hr10-nn_hr10)/max(nn_hr10,0.001):.1%}")

🔴 运行结果

✅ 验证通过

🔍 代码解读

💡 关键洞察

📝 本章小结

🧪 练习

📚 延伸阅读

💼 面试考点

🏭 工业案例