🎓 第30课：毕业项目——工业级推荐系统

🎯 学习目标

🗂️ 课程定位

📖 核心理论

本课是整个课程的毕业项目，将30课所学整合为完整的工业级推荐系统。从数据层到评估层，覆盖推荐系统的全部关键环节。

┌──────────────────────────────────────────┐
│                  API网关                   │
├──────────────────────────────────────────┤
│  数据层 → 召回层 → 排序层 → 重排层 → 评估层  │
│  ┌───┐   ┌───┐   ┌───┐   ┌───┐   ┌───┐  │
│  │特征│   │多路│   │精排│   │多样│   │A/B│  │
│  │存储│   │召回│   │模型│   │性  │   │测试│  │
│  └───┘   └───┘   └───┘   └───┘   └───┘  │
└──────────────────────────────────────────┘

📐 数学公式

💻 代码实现

import numpy as np
from collections import defaultdict
import time
np.random.seed(42)
print("="*60+"\n工业级推荐系统(毕业项目)\n"+"="*60)
# ===== 系统配置 =====
n_users=100;n_items=80;n_features=10
print(f"规模:用户={n_users} 物品={n_items}")
# ===== 数据层 =====
print("\n【数据层】")
# 用户特征
user_features=np.random.randn(n_users,n_features)
user_active=np.random.poisson(5,n_users)+1
# 物品特征
item_features=np.random.randn(n_items,n_features)
item_category=np.random.randint(5,size=n_items)
item_popularity=np.random.poisson(3,n_items)+1
item_freshness=np.random.randint(1,72,n_items)  # 小时
# 交互数据
interactions={}
for u in range(n_users):
    n_int=user_active[u]
    for i in np.random.choice(n_items,size=min(n_int,n_items),replace=False):
        interactions[(u,i)]=np.random.randint(1,6)
print(f"交互数:{len(interactions)} 密度:{len(interactions)/(n_users*n_items):.4f}")
# ===== 召回层 =====
print("\n【召回层】")
t0=time.time()
# 多路召回
recall_results=defaultdict(set)
# 1. 协同过滤召回
uv=defaultdict(dict)
for(u,i),r in interactions.items():uv[u][i]=r
def cosine_sim(a,b):
    ck=set(a.keys())&set(b.keys())
    if not ck:return 0
    va=np.array([a[k]for k in ck]);vb=np.array([b[k]for k in ck])
    return np.dot(va,vb)/(np.linalg.norm(va)*np.linalg.norm(vb)+1e-10)
# ItemCF
item_users=defaultdict(dict)
for(u,i),r in interactions.items():item_users[i][u]=r
item_sim=defaultdict(dict)
for i1 in range(n_items):
    for i2 in range(i1+1,n_items):
        s=cosine_sim(item_users.get(i1,{}),item_users.get(i2,{}))
        if s>0.1:item_sim[i1][i2]=item_sim[i2][i1]=s
for u in range(n_users):
    for i,r in uv[u].items():
        for j,s in item_sim.get(i,{}).items():
            if j not in uv[u]:recall_results[u].add(j)
# 2. 向量召回
K_emb=8
P=np.random.randn(n_users,K_emb)*0.1;Q=np.random.randn(n_items,K_emb)*0.1
obs=list(interactions.keys())
for ep in range(30):
    for u,i in obs:
        r=interactions[(u,i)];pred=P[u]@Q[i];err=r-pred
        P[u]+=0.005*(err*Q[i]-0.02*P[u]);Q[i]+=0.005*(err*P[u]-0.02*Q[i])
for u in range(n_users):
    scores=P[u]@Q.T
    top=np.argsort(-scores)[:20]
    recall_results[u].update(top)
# 3. 热门召回
hot_items=np.argsort(-item_popularity)[:10]
for u in range(n_users):recall_results[u].update(hot_items)
recall_time=time.time()-t0
print(f"平均每用户召回:{np.mean([len(v)for v in recall_results.values()]):.0f}个 耗时:{recall_time*1000:.0f}ms")
# ===== 排序层 =====
print("\n【排序层】")
t1=time.time()
rank_results={}
for u in range(n_users):
    candidates=recall_results[u]-set(uv[u].keys())
    if not candidates:continue
    scores={}
    for i in candidates:
        # 特征:向量得分+热度+新鲜度+类目偏好
        vec_score=sigmoid(P[u]@Q[i])
        pop_score=item_popularity[i]/max(item_popularity)
        fresh_score=1/(1+item_freshness[i]/24)
        cat_pref=1 if item_category[i]==np.argmax([sum(1for j in uv[u]if item_category[j]==c)for c in range(5)])else 0.3
        scores[i]=0.4*vec_score+0.2*pop_score+0.2*fresh_score+0.2*cat_pref
    rank_results[u]=sorted(scores.items(),key=lambda x:-x[1])[:20]
def sigmoid(x):return 1/(1+np.exp(-x))
rank_time=time.time()-t1
print(f"排序耗时:{rank_time*1000:.0f}ms")
# ===== 重排层 =====
print("\n【重排层】")
t2=time.time()
final_results={}
for u in range(n_users):
    if u not in rank_results:continue
    rec_items=[i for i,_ in rank_results[u]]
    # 类目打散(每类最多3个)
    cat_count=defaultdict(int);reranked=[]
    for i in rec_items:
        c=item_category[i]
        if cat_count[c]<3:reranked.append(i);cat_count[c]+=1
    # 时效性调整
    final_results[u]=reranked[:10]
rerank_time=time.time()-t2
print(f"重排耗时:{rerank_time*1000:.0f}ms")
# ===== 评估层 =====
print("\n【评估层】")
# 留一法评估
test_data={}
for u in range(n_users):
    if len(uv[u])<2:continue
    items=list(uv[u].keys())
    hidden=items[np.random.randint(len(items))]
    test_data[u]=hidden
hits=0;ndcgs=[]
for u,hidden in test_data.items():
    if u not in final_results:continue
    if hidden in final_results[u]:
        hits+=1
        rank=final_results[u].index(hidden)
        ndcgs.append(1/np.log2(rank+2))
    else:ndcgs.append(0)
hr=hits/len(test_data)if test_data else 0
ndcg=np.mean(ndcgs)if ndcgs else 0
# 覆盖率
all_rec=set()
for u in final_results:all_rec.update(final_results[u])
coverage=len(all_rec)/n_items
# 多样性
diversities=[]
for u in final_results:
    cats=set(item_category[i]for i in final_results[u])
    diversities.append(len(cats)/5)
print(f"HitRate@10={hr:.4f} NDCG@10={ndcg:.4f} 覆盖率={coverage:.2%} 多样性={np.mean(diversities):.4f}")
# ===== 性能统计 =====
print("\n【性能统计】")
total_time=recall_time+rank_time+rerank_time
print(f"总延迟:{total_time*1000:.0f}ms (召回:{recall_time*1000:.0f}ms 排序:{rank_time*1000:.0f}ms 重排:{rerank_time*1000:.0f}ms)")
print(f"目标:<100ms (工业级要求)")

🔴 运行结果

✅ 验证通过

🔍 代码解读

💡 关键洞察

📝 本章小结

🧪 练习

📚 延伸阅读

💼 面试考点

🏭 工业案例