🎵 第28课：音乐推荐项目

🎯 学习目标

🗂️ 课程定位

📖 核心理论

音乐推荐有独特性：用户消费是连续会话式的（听歌是一个序列），且受心情强烈影响。Spotify和网易云音乐的推荐成功证明了音乐推荐的商业价值。

📐 数学公式

💻 代码实现

import numpy as np
from collections import defaultdict
np.random.seed(42)
print("="*60+"\n音乐推荐项目\n"+"="*60)
n_users=30;n_songs=40;n_moods=4;n_genres=5
moods=["开心","忧伤","放松","活力"]
genres=["流行","摇滚","电子","古典","嘻哈"]
# 歌曲特征
song_genre=np.random.randint(n_genres,size=n_songs)
song_mood=np.random.randint(n_moods,size=n_songs)
song_energy=np.random.random(n_songs)
song_tempo=np.random.randint(60,180,n_songs)
song_pop=np.random.poisson(5,n_songs)+1
song_emb=np.random.randn(n_songs,8)
print(f"用户:{n_users} 歌曲:{n_songs}")
# 听歌序列
user_sequences=defaultdict(list)
for u in range(n_users):
    n_songs_u=np.random.randint(10,30)
    current_mood=np.random.randint(n_moods)
    seq=[]
    for _ in range(n_songs_u):
        mood_songs=np.where(song_mood==current_mood)[0]
        if len(mood_songs)==0:mood_songs=np.arange(n_songs)
        song=np.random.choice(mood_songs)
        play_ratio=np.random.random()  # 播放完成率
        seq.append((song,play_ratio))
        if np.random.random()<0.2:current_mood=np.random.randint(n_moods)
    user_sequences[u]=seq
print(f"总听歌数:{sum(len(s)for s in user_sequences.values())}")
# 协同过滤
print("\n--- 协同过滤(基于完整播放) ---")
uv=defaultdict(dict)
for u,seq in user_sequences.items():
    for song,play_ratio in seq:
        if play_ratio>0.5:uv[u][song]=uv[u].get(song,0)+1
# ItemCF
item_co=defaultdict(lambda:defaultdict(int))
for u,songs in uv.items():
    for s1 in songs:
        for s2 in songs:
            if s1!=s2:item_co[s1][s2]+=1
target_songs=list(uv.get(0,{}).keys())
cf_scores=defaultdict(float)
for s in target_songs:
    for s2,count in item_co[s].items():
        if s2 not in uv.get(0,{}):cf_scores[s2]+=count
cf_rec=sorted(cf_scores.items(),key=lambda x:-x[1])[:5]
print(f"ItemCF推荐:{[f'歌曲{s}'for s,_ in cf_rec]}")
# 基于音频特征推荐
print("\n--- 音频特征推荐 ---")
user_profile=np.zeros(8)
for song,_ in user_sequences[0]:
    user_profile+=song_emb[song]
user_profile/=max(1,len(user_sequences[0]))
audio_scores=song_emb@user_profile
audio_rec=sorted([(i,audio_scores[i])for i in range(n_songs)if i not in uv.get(0,{})],key=lambda x:-x[1])[:5]
print(f"音频特征推荐:{[(f'歌曲{i}',f'{s:.3f}')for i,s in audio_rec]}")
# 心情推荐
print("\n--- 心情推荐 ---")
# 根据用户最近听歌推断心情
recent_songs=[s for s,_ in user_sequences[0][-5:]]
recent_moods=[song_mood[s]for s in recent_songs]
from collections import Counter
current_mood=Counter(recent_moods).most_common(1)[0][0]
mood_songs=[s for s in range(n_songs)if song_mood[s]==current_mood and s not in uv.get(0,{})]
mood_rec=sorted(mood_songs,key=lambda s:-song_pop[s])[:5]
print(f"推断心情:{moods[current_mood]} 推荐:{[f'歌曲{s}'for s in mood_rec]}")
# 会话推荐
print("\n--- 会话推荐 ---")
def session_rec(sequence,k=5):
    if len(sequence)<2:return[]
    last_song=sequence[-1][0]
    similar=sorted([(s,np.dot(song_emb[last_song],song_emb[s])/(np.linalg.norm(song_emb[last_song])*np.linalg.norm(song_emb[s])+1e-10))for s in range(n_songs)],key=lambda x:-x[1])
    return[s for s,_ in similar[:k]]
sr=session_rec(user_sequences[0])
print(f"会话推荐:{[f'歌曲{s}'for s in sr]}")

🔴 运行结果

✅ 验证通过

🔍 代码解读

💡 关键洞察

📝 本章小结

🧪 练习

📚 延伸阅读

💼 面试考点

🏭 工业案例