AI 音乐数据分析：音频特征提取与用户偏好聚类的跨界探索

AI 音乐数据分析：音频特征提取与用户偏好聚类的跨界探索

朱大喜的数据日记 | 第 8 篇

数据分析的尽头不是报表，是让每一首歌找到它该去的人。

上周接到一个需求——音乐平台的运营团队想要搞清楚"用户到底在听什么"，而不是只看播放量排名。说实话，第一次碰到音频数据分析的项目，心里还挺慌的。音频信号处理这块我之前只在统计学课上接触过傅里叶变换，实战经验为零。但转念一想，不就是换了一种数据源嘛？核心逻辑还是特征工程 + 聚类分析，我熟悉的那套 Python 全家桶照样能用。

这篇文章复盘整个项目过程，从音频特征提取到用户偏好聚类，再到最终的推荐策略验证，踩了不少坑但也收获了很多跨界融合的思路。

一、项目背景与需求拆解

音乐平台运营团队的需求说起来很简单：别只给我排行榜，给我用户画像。

但拆解下来，至少三层含义：

传统做法是靠人工标注流派标签 + 协同过滤，但标签覆盖不全、协同过滤对冷启动无力。我们决定走一条新路：用 AI 直接从音频文件提取特征，再用聚类建模用户偏好空间。

二、音频特征提取：从波形到数字

2.1 特征工程的整体思路

音频数据分析的第一步，也是最关键的一步——把 MP3/WAV 文件变成数字特征矩阵。这里我用的是 librosa，Python 音频处理的标配库。

音频特征可以分三大类：

其中 MFCC 是最核心的特征，它模拟人耳对频率的感知方式，13 个系数就能浓缩一首歌的"音色指纹"。

2.2 代码实现：批量音频特征提取

import librosa
import numpy as np
import pandas as pd
from pathlib import Path
from concurrent.futures import ProcessPoolExecutor

# 定义单首歌的特征提取函数
def extract_song_features(audio_path: str) -> dict:
"""
从单个音频文件提取多维特征
返回包含时域、频域、时频域特征的字典
"""
# 加载音频，统一采样率为22050Hz（librosa默认）
# sr=22050 是音乐分析的常用采样率，平衡精度和计算量
y, sr = librosa.load(audio_path, sr=22050, duration=30)
# 只取前30秒，避免尾部沉默影响特征质量

features = {}

# === 时域特征 ===
# RMS能量：反映歌曲的"响度"水平
rms = librosa.feature.rms(y=y)
features['rms_mean'] = np.mean(rms)
features['rms_std'] = np.std(rms) # 标准差反映能量波动

# 零交叉率：信号从正到负的切换频率，与噪声/打击乐相关
zcr = librosa.feature.zero_crossing_rate(y)
features['zcr_mean'] = np.mean(zcr)

# === 频域特征 ===
# MFCC：13个倒谱系数，浓缩音色信息
# 这是音频分析最重要的特征，模拟人耳频率感知
mfccs = librosa.feature.mfcc(y=y, sr=sr, n_mfcc=13)
for i in range(13):
features[f'mfcc_{i}_mean'] = np.mean(mfccs[i])
features[f'mfcc_{i}_std'] = np.std(mfccs[i]) # 标准差捕捉变化

# 频谱质心：频率分布的"重心"，反映声音的"亮度"
# 高质心 = 明亮/尖锐，低质心 = 低沉/浑厚
spectral_centroid = librosa.feature.spectral_centroid(y=y, sr=sr)
features['spectral_centroid_mean'] = np.mean(spectral_centroid)

# 频谱带宽：频率分布的"宽度"，反映音色的丰富程度
spectral_bw = librosa.feature.spectral_bandwidth(y=y, sr=sr)
features['spectral_bandwidth_mean'] = np.mean(spectral_bw)

# 频谱滚降点：能量集中分布的截止频率
# 85%的能量集中在滚降频率以下，衡量声音的高频成分
spectral_rolloff = librosa.feature.spectral_rolloff(y=y, sr=sr)
features['spectral_rolloff_mean'] = np.mean(spectral_rolloff)

# === 时频域特征 ===
# 节拍BPM：用librosa的节拍跟踪算法估计
# 这是用户最直观的节奏感知指标
tempo, beat_frames = librosa.beat.beat_track(y=y, sr=sr)
features['tempo'] = float(tempo)

# 色度特征：12个半音的能量分布，反映和弦/调性
# 用色度可以判断歌曲的"情绪基调"（大调偏明亮，小调偏忧伤）
chroma = librosa.feature.chroma_stft(y=y, sr=sr)
for i in range(12):
features[f'chroma_{i}_mean'] = np.mean(chroma[i])

return features

# 批量提取：用多进程加速（音频处理是CPU密集型任务）
def batch_extract(audio_dir: str, output_csv: str) -> pd.DataFrame:
"""
批量提取目录下所有音频文件的特征
使用多进程并行处理，加速特征提取
"""
audio_files = list(Path(audio_dir).glob('*.mp3'))
print(f"找到 {len(audio_files)} 个音频文件")

# ProcessPoolExecutor：每个进程独立加载librosa，避免GIL限制
# max_workers=4：音频IO+计算混合，4核刚好压满
with ProcessPoolExecutor(max_workers=4) as executor:
results = list(executor.map(extract_song_features,
[str(f) for f in audio_files]))

# 组装DataFrame，song_id用文件名（去掉.mp3后缀）
df = pd.DataFrame(results)
df['song_id'] = [f.stem for f in audio_files]

# 保存到CSV，方便后续复用（特征提取很耗时，别重复跑）
df.to_csv(output_csv, index=False)
print(f"特征提取完成，共 {len(df)} 首，{len(df.columns)} 个特征维度")
return df

# 执行批量提取
song_features_df = batch_extract('/data/music_library', '/data/song_features.csv')

踩坑记录：librosa.load 默认会加载完整音频，一首5分钟的歌要处理好几秒。我只取前30秒（duration=30），既节省时间又能抓住歌曲的主旋律段落。另外，MP3 解码是 CPU 密集型操作，必须用 ProcessPoolExecutor 多进程才能加速，ThreadPoolExecutor 因为 GIL 没效果。

2.3 特征降维：从40维到可解释的6维

提取了40多个特征维度，直接聚类会有维度灾难问题。我用 PCA 降到6维，但关键是——这6维必须能映射到可解释的听觉属性。

from sklearn.preprocessing import StandardScaler
from sklearn.decomposition import PCA

# 特征标准化：MFCC和其他特征量级差异极大，必须先归一化
scaler = StandardScaler()
X_scaled = scaler.fit_transform(song_features_df.drop(columns=['song_id']))

# PCA降维：保留95%方差，自动选择维度数
pca = PCA(n_components=0.95)
X_pca = pca.fit_transform(X_scaled)
print(f"原始维度: {X_scaled.shape[1]}, 降维后: {X_pca.shape[1]}")

# 查看每个主成分的贡献率，判断可解释性
for i, ratio in enumerate(pca.explained_variance_ratio_):
print(f"PC{i+1}: {ratio:.2%} 方差贡献")

# 分析主成分与原始特征的相关性，给PC起"人能懂的名字"
# PC1 与 RMS、spectral_centroid 强相关 → "能量与亮度"
# PC2 与 tempo 强相关 → "节奏速度"
# PC3 与 chroma_0,1 强相关 → "调性明暗"
# PC4 与 MFCC_0,1 强相关 → "音色粗糙度"
# PC5 与 zcr 强相关 → "噪声/打击感"
# PC6 与 spectral_bandwidth 强相关 → "频率丰富度"

最终保留6个主成分，累计方差贡献率约92%，而且每个主成分都能对应一个人类能理解的听觉维度。这步很重要——数据分析的结果如果不能翻译成业务语言，就是无效输出。

三、用户偏好聚类：从听歌记录到画像分组

3.1 偏好向量构建

有了歌曲特征矩阵，下一步是把用户的播放记录转换成偏好向量。核心思路：用户偏好 = 他常听歌曲特征的加权平均。

def build_user_preference_vector(user_logs: pd.DataFrame,
song_features: pd.DataFrame) -> pd.DataFrame:
"""
根据用户播放日志构建偏好向量
权重 = 播放次数 × 完播率（听完才算真正喜欢）
"""
# 计算每首歌的加权播放分数：播放次数 × 完播率
# 完播率低于0.3的记录直接过滤（跳歌不算偏好）
valid_logs = user_logs[user_logs['completion_rate'] >= 0.3].copy()
valid_logs['weight'] = valid_logs['play_count'] * valid_logs['completion_rate']

# 合并歌曲特征，按用户聚合
merged = valid_logs.merge(song_features, on='song_id', how='left')

# 每个用户：对所有听过的歌的特征做加权平均
feature_cols = [c for c in song_features.columns if c != 'song_id']
user_prefs = merged.groupby('user_id').apply(
lambda g: np.average(g[feature_cols].values,
weights=g['weight'].values, axis=0)
)

# 转成DataFrame
pref_df = pd.DataFrame(user_prefs.tolist(),
columns=feature_cols,
index=user_prefs.index)
return pref_df

user_pref_df = build_user_preference_vector(play_logs, song_features_df_pca)
print(f"用户偏好矩阵: {user_pref_df.shape}") # (15000, 6)

3.2 KMeans聚类与画像解读

from sklearn.cluster import KMeans
from sklearn.metrics import silhouette_score

# 聚类数选择：用轮廓系数评估不同K值
# 音乐偏好不像零售画像那么离散，测试K=4到8
silhouette_scores = {}
for k in range(4, 9):
km = KMeans(n_clusters=k, random_state=42, n_init=10)
labels = km.fit_predict(user_pref_df)
score = silhouette_score(user_pref_df, labels)
silhouette_scores[k] = score
print(f"K={k}, 轮廓系数={score:.3f}")

# K=5 轮廓系数最高（0.42），选择5组
best_k = max(silhouette_scores, key=silhouette_scores.get)
kmeans = KMeans(n_clusters=best_k, random_state=42, n_init=10)
user_pref_df['cluster'] = kmeans.fit_predict(user_pref_df)

# 解读每个聚类的特征中心，翻译成"画像标签"
cluster_centers = pd.DataFrame(kmeans.cluster_centers_,
columns=user_pref_df.columns[:-1])

# 根据各聚类在各主成分上的偏离程度，人工命名
# Cluster 0: 高能量+快节奏 → "活力摇滚党"
# Cluster 1: 低能量+慢节奏+暗调性 → "深夜emo组"
# Cluster 2: 高频丰富度+中能量 → "精致文艺派"
# Cluster 3: 高噪声感+高节奏 → "电子舞曲控"
# Cluster 4: 中等能量+明亮调性 → "阳光流行粉"
cluster_names = {
0: "活力摇滚党",
1: "深夜emo组",
2: "精致文艺派",
3: "电子舞曲控",
4: "阳光流行粉"
}
user_pref_df['cluster_name'] = user_pref_df['cluster'].map(cluster_names)

结果出来了——15000个用户分成5组，每组特征中心差异明显。运营看了聚类画像后说："这个'深夜emo组'和我们后台看到的凌晨1-3点活跃用户高度吻合。"数据验证了直觉，这才是分析的价值。

四、推荐策略验证：A/B 测试与效果评估

4.1 分组推荐策略设计

有了用户画像分组，下一步是设计差异化的推荐策略。不是简单地把同类用户常听的歌推给他，而是结合歌曲特征空间做匹配。

def recommend_by_cluster(user_cluster: int,
song_features: pd.DataFrame,
cluster_centers: np.ndarray,
user_history: set,
top_n: int = 20) -> list:
"""
基于聚类中心的推荐策略
核心逻辑：在歌曲特征空间中，找离用户聚类中心最近的歌
排除用户已听过的，避免重复推荐
"""
# 取该聚类的中心向量
center = cluster_centers[user_cluster]

# 计算每首歌到聚类中心的距离
song_vectors = song_features.drop(columns=['song_id']).values
distances = np.linalg.norm(song_vectors – center, axis=1)

# 距离排序：越近越匹配
ranked_indices = np.argsort(distances)

# 过滤已听过的歌，避免"你刚听完又推给你"
recommendations = []
for idx in ranked_indices:
song_id = song_features.iloc[idx]['song_id']
if song_id not in user_history:
recommendations.append({
'song_id': song_id,
'distance': distances[idx],
'match_score': 1 / (1 + distances[idx]) # 转成0-1匹配分数
})
if len(recommendations) >= top_n:
break

return recommendations

# 对每个聚类分组生成推荐池
for cluster_id, name in cluster_names.items():
# 取该分组所有用户的历史听歌集合
group_users = user_pref_df[user_pref_df['cluster'] == cluster_id].index
group_history = set(play_logs[play_logs['user_id'].isin(group_users)]['song_id'])

# 生成推荐列表
recs = recommend_by_cluster(cluster_id, song_features_df_pca,
kmeans.cluster_centers_, group_history)
print(f"{name}: 推荐池 {len(recs)} 首，平均匹配分数 {np.mean([r['match_score'] for r in recs]):.3f}")

4.2 A/B 测试与结果

我们做了为期两周的 A/B 测试：

• 对照组：热门排行榜推荐（编辑精选 + 播放量排序）
• 实验组：聚类画像匹配推荐

冷启动用户的提升最亮眼——这些新用户没有播放历史，协同过滤完全没辙，但聚类画像可以根据注册时填的偏好标签快速定位分组，再从分组中心做匹配推荐。

def cold_start_recommend(user_tags: list,
song_features: pd.DataFrame,
tag_cluster_map: dict,
cluster_centers: np.ndarray,
top_n: int = 15) -> list:
"""
冷启动用户推荐：根据注册标签映射到聚类分组
新用户没有播放历史，只能靠注册时的偏好标签判断
"""
# 根据用户选择的标签，映射到最近的聚类分组
# 标签如 "摇滚" → Cluster 0, "电子" → Cluster 3
mapped_clusters = [tag_cluster_map.get(tag, 4) for tag in user_tags]
# 取最常见的映射分组作为主分组
main_cluster = max(set(mapped_clusters), key=mapped_clusters.count)

# 从主分组中心推荐
recs = recommend_by_cluster(main_cluster, song_features,
cluster_centers, set(), top_n)
return recs

五、总结

这个项目让我体会到音频数据分析的特殊性，也让我对"跨界融合"有了更深的理解。总结几点经验：

下次如果再碰到音频项目，我会直接跳过"要不要试试"的犹豫阶段，上手就 librosa + PCA + KMeans，这套组合已经验证过了。数据分析的核心竞争力不是会多少工具，而是能把方法论快速适配到新领域的能力。