CMU-MOSEI 数据集实战：3模态特征提取与融合代码全解析（附PyTorch实现）

weixin_30239339

340人浏览 · 2026-07-04 14:19:18

weixin_30239339 · 2026-07-04 14:19:18 发布

CMU-MOSEI 数据集实战：3模态特征提取与融合代码全解析（附PyTorch实现）

多模态情感分析正成为人工智能领域的前沿研究方向，而CMU-MOSEI作为当前规模最大的三模态情感分析数据集，为研究者提供了丰富的实验素材。本文将深入探讨如何从原始视频数据出发，构建完整的特征提取与融合流程，并提供可直接运行的PyTorch实现代码。

1. 环境准备与数据加载

在开始特征提取前，我们需要配置合适的Python环境并理解CMU-MOSEI的数据结构。数据集包含3228个视频片段，每个片段都提供了文本转录、音频和视觉三种模态的原始数据，以及情感和情绪两类标注。

首先安装必要的依赖库：

pip install torch torchvision torchaudio
pip install openface-python pandas numpy tqdm

数据集下载后，其目录结构通常如下：

CMU_MOSEI/
├── Raw/
│   ├── Audio/
│   ├── Video/
│   └── Transcript/
├── Labels/
│   ├── sentiment/
│   └── emotion/
└── Metadata/

我们创建一个自定义的PyTorch数据集类来加载这些数据：

import os
import torch
from torch.utils.data import Dataset

class MOSEIDataset(Dataset):
    def __init__(self, root_dir, split='train'):
        self.root_dir = root_dir
        self.split = split
        self.metadata = self._load_metadata()
        
    def _load_metadata(self):
        # 加载数据划分文件
        split_file = os.path.join(self.root_dir, 'Metadata', f'{self.split}_split.csv')
        return pd.read_csv(split_file)
    
    def __len__(self):
        return len(self.metadata)
    
    def __getitem__(self, idx):
        video_id = self.metadata.iloc[idx]['video_id']
        # 加载三种模态的原始数据
        text = self._load_text(video_id)
        audio = self._load_audio(video_id)
        visual = self._load_visual(video_id)
        # 加载标签
        labels = self._load_labels(video_id)
        
        return {
            'text': text,
            'audio': audio,
            'visual': visual,
            'labels': labels
        }

2. 文本特征提取：GloVe嵌入实现

文本模态是情感分析中最直接的信息来源。CMU-MOSEI提供了手工转录的文本数据，我们将使用预训练的GloVe词向量进行特征提取。

首先下载预训练的GloVe模型（如glove.6B.300d.txt），然后实现文本特征提取器：

import numpy as np
from collections import defaultdict

class GloveEmbedder:
    def __init__(self, glove_path):
        self.glove_path = glove_path
        self.embeddings = self._load_glove()
        
    def _load_glove(self):
        embeddings = defaultdict(lambda: np.zeros(300))
        with open(self.glove_path, 'r', encoding='utf-8') as f:
            for line in f:
                values = line.split()
                word = values[0]
                vector = np.asarray(values[1:], dtype='float32')
                embeddings[word] = vector
        return embeddings
    
    def embed_sentence(self, sentence):
        words = sentence.lower().split()
        vectors = [self.embeddings[word] for word in words]
        return np.stack(vectors) if vectors else np.zeros((1, 300))

在实际应用中，我们还需要处理变长序列并生成固定维度的句子表示：

def get_text_features(dataset, embedder):
    text_features = []
    for item in dataset:
        sentence = item['text']
        word_vectors = embedder.embed_sentence(sentence)
        # 平均池化生成句子级表示
        sentence_vector = np.mean(word_vectors, axis=0)
        text_features.append(sentence_vector)
    return torch.FloatTensor(text_features)

提示：对于更高级的文本特征提取，可以考虑使用BERT等预训练语言模型，但GloVe作为轻量级解决方案在多数场景下仍表现良好。

3. 视觉特征提取：OpenFace与FACET实战

视觉模态包含丰富的面部表情和肢体语言信息。CMU-MOSEI官方推荐使用OpenFace和FACET工具包提取视觉特征。

3.1 OpenFace特征提取

OpenFace提供了68个面部关键点、头部姿态等多种视觉特征。以下是使用OpenFace命令行工具批量处理视频的示例：

# 批量提取视频特征
for video in $(ls ./Raw/Video/*.mp4); do
    OpenFace/build/bin/FeatureExtraction -f $video -out_dir ./Features/OpenFace
done

在Python中，我们可以直接使用OpenFace的Python接口：

import OpenFace

def extract_openface_features(video_path):
    # 初始化OpenFace工具
    of = OpenFace.ProcessAndTrack()
    # 处理视频文件
    success = of.ProcessVideo(video_path)
    if not success:
        raise ValueError(f"OpenFace processing failed for {video_path}")
    
    # 获取特征数据
    au_features = of.GetActionUnits()  # 动作单元
    gaze_features = of.GetGazeDirection()  # 视线方向
    pose_features = of.GetHeadPose()  # 头部姿态
    
    # 合并所有视觉特征
    visual_features = np.concatenate([
        au_features.mean(axis=0),
        gaze_features.mean(axis=0),
        pose_features.mean(axis=0)
    ])
    return visual_features

3.2 FACET情绪特征提取

FACET专门用于提取面部表情相关的情绪特征，可以补充OpenFace的特征：

import iMotions

def extract_facet_features(video_path):
    # 初始化FACET分析器
    facet = iMotions.FacetAnalysis()
    # 分析视频
    results = facet.analyze(video_path)
    
    # 提取六种基本情绪特征
    emotions = ['happiness', 'sadness', 'anger', 
               'fear', 'disgust', 'surprise']
    emotion_features = [results[emotion].mean() for emotion in emotions]
    
    return np.array(emotion_features)

最终我们将OpenFace和FACET的特征拼接起来，形成完整的视觉特征表示：

def get_visual_features(video_path):
    openface = extract_openface_features(video_path)
    facet = extract_facet_features(video_path)
    return np.concatenate([openface, facet])

4. 音频特征提取：COVAREP参数化分析

音频模态包含语调、韵律等副语言信息。COVAREP是专门用于语音分析的MATLAB工具箱，但我们可以在Python中调用它。

首先需要准备MATLAB运行时环境，然后通过Python-MATLAB引擎桥接：

import matlab.engine

class CovarepExtractor:
    def __init__(self):
        self.eng = matlab.engine.start_matlab()
        self.eng.addpath('path/to/covarep', nargout=0)
        
    def extract_features(self, audio_path):
        # 调用MATLAB函数
        features = self.eng.covarep(audio_path, nargout=1)
        # 转换为numpy数组
        return np.array(features._data).reshape(features.size, order='F')
    
    def close(self):
        self.eng.quit()

对于没有MATLAB环境的用户，可以使用Python替代方案librosa：

import librosa

def extract_audio_features(audio_path, sr=22050):
    # 加载音频文件
    y, sr = librosa.load(audio_path, sr=sr)
    
    # 提取MFCC特征
    mfcc = librosa.feature.mfcc(y=y, sr=sr, n_mfcc=13)
    # 提取基频特征
    f0 = librosa.yin(y, fmin=librosa.note_to_hz('C2'),
                    fmax=librosa.note_to_hz('C7'))
    # 提取频谱质心
    spectral_centroid = librosa.feature.spectral_centroid(y=y, sr=sr)
    
    # 统计特征
    features = np.concatenate([
        mfcc.mean(axis=1),
        [np.nanmean(f0)],
        spectral_centroid.mean(axis=1)
    ])
    return np.nan_to_num(features)  # 处理可能的NaN值

5. 多模态数据加载器实现

有了三种模态的特征提取器后，我们需要实现一个完整的数据加载器，处理特征对齐和批量加载。

from torch.utils.data import DataLoader

class MOSEIDataLoader:
    def __init__(self, dataset, batch_size=32, shuffle=True):
        self.dataset = dataset
        self.batch_size = batch_size
        self.shuffle = shuffle
        
        # 初始化特征提取器
        self.text_embedder = GloveEmbedder('glove.6B.300d.txt')
        
    def _process_item(self, item):
        # 提取文本特征
        text_feat = self.text_embedder.embed_sentence(item['text'])
        # 提取音频特征
        audio_feat = extract_audio_features(item['audio'])
        # 提取视觉特征
        visual_feat = get_visual_features(item['video'])
        
        # 对齐特征维度
        text_feat = torch.FloatTensor(text_feat)
        audio_feat = torch.FloatTensor(audio_feat)
        visual_feat = torch.FloatTensor(visual_feat)
        
        return text_feat, audio_feat, visual_feat, item['labels']
    
    def __iter__(self):
        indices = list(range(len(self.dataset)))
        if self.shuffle:
            random.shuffle(indices)
            
        for start_idx in range(0, len(indices), self.batch_size):
            batch_indices = indices[start_idx:start_idx+self.batch_size]
            batch_items = [self.dataset[i] for i in batch_indices]
            
            # 处理批量数据
            batch = list(zip(*[self._process_item(item) for item in batch_items]))
            yield {
                'text': torch.stack(batch[0]),
                'audio': torch.stack(batch[1]),
                'visual': torch.stack(batch[2]),
                'labels': torch.stack(batch[3])
            }

注意：在实际应用中，建议预先提取所有特征并保存到磁盘，避免每次训练时重复提取，这可以显著提高训练效率。

6. 早期融合模型实现

早期融合(Early Fusion)是最基础的多模态融合策略，我们在PyTorch中实现一个简单的早期融合模型：

import torch.nn as nn

class EarlyFusionModel(nn.Module):
    def __init__(self, text_dim=300, audio_dim=74, visual_dim=158, num_classes=7):
        super().__init__()
        # 各模态的维度调整层
        self.text_proj = nn.Linear(text_dim, 128)
        self.audio_proj = nn.Linear(audio_dim, 128)
        self.visual_proj = nn.Linear(visual_dim, 128)
        
        # 融合后的分类器
        self.classifier = nn.Sequential(
            nn.Linear(128*3, 256),
            nn.ReLU(),
            nn.Dropout(0.3),
            nn.Linear(256, num_classes)
        )
        
    def forward(self, text, audio, visual):
        # 投影各模态特征到统一维度
        text_feat = self.text_proj(text)
        audio_feat = self.audio_proj(audio)
        visual_feat = self.visual_proj(visual)
        
        # 拼接特征
        fused = torch.cat([text_feat, audio_feat, visual_feat], dim=1)
        
        # 分类预测
        return self.classifier(fused)

模型训练流程的完整实现：

def train_model(model, dataloader, epochs=50):
    device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
    model = model.to(device)
    
    criterion = nn.CrossEntropyLoss()
    optimizer = torch.optim.Adam(model.parameters(), lr=1e-4)
    
    for epoch in range(epochs):
        model.train()
        total_loss = 0
        
        for batch in dataloader:
            text = batch['text'].to(device)
            audio = batch['audio'].to(device)
            visual = batch['visual'].to(device)
            labels = batch['labels'].to(device)
            
            optimizer.zero_grad()
            outputs = model(text, audio, visual)
            loss = criterion(outputs, labels)
            loss.backward()
            optimizer.step()
            
            total_loss += loss.item()
        
        print(f'Epoch {epoch+1}, Loss: {total_loss/len(dataloader):.4f}')
    
    return model

7. 进阶：基于注意力的晚期融合模型

相比早期融合，晚期融合可以更好地保留各模态的特性。我们实现一个基于注意力机制的晚期融合模型：

class LateFusionModel(nn.Module):
    def __init__(self, text_dim=300, audio_dim=74, visual_dim=158, num_classes=7):
        super().__init__()
        # 各模态的独立特征提取器
        self.text_net = nn.Sequential(
            nn.Linear(text_dim, 256),
            nn.ReLU(),
            nn.Dropout(0.2)
        )
        self.audio_net = nn.Sequential(
            nn.Linear(audio_dim, 256),
            nn.ReLU(),
            nn.Dropout(0.2)
        )
        self.visual_net = nn.Sequential(
            nn.Linear(visual_dim, 256),
            nn.ReLU(),
            nn.Dropout(0.2)
        )
        
        # 注意力机制
        self.attention = nn.Sequential(
            nn.Linear(256*3, 128),
            nn.Tanh(),
            nn.Linear(128, 3),
            nn.Softmax(dim=1)
        )
        
        # 分类器
        self.classifier = nn.Linear(256*3, num_classes)
        
    def forward(self, text, audio, visual):
        # 提取各模态特征
        text_feat = self.text_net(text)
        audio_feat = self.audio_net(audio)
        visual_feat = self.visual_net(visual)
        
        # 拼接特征用于注意力计算
        combined = torch.cat([text_feat, audio_feat, visual_feat], dim=1)
        attn_weights = self.attention(combined)
        
        # 应用注意力权重
        text_attn = text_feat * attn_weights[:, 0].unsqueeze(1)
        audio_attn = audio_feat * attn_weights[:, 1].unsqueeze(1)
        visual_attn = visual_feat * attn_weights[:, 2].unsqueeze(1)
        
        # 融合加权的特征
        fused = torch.cat([text_attn, audio_attn, visual_attn], dim=1)
        return self.classifier(fused)

8. 实验评估与结果分析

为了全面评估模型性能，我们需要实现标准的评估指标。CMU-MOSEI通常使用以下指标：

分类准确率（7类、5类和2类）
F1分数（宏平均）
平均绝对误差（MAE）
皮尔逊相关系数（Corr）

实现评估函数：

from sklearn.metrics import accuracy_score, f1_score, mean_absolute_error

def evaluate_model(model, dataloader):
    device = next(model.parameters()).device
    model.eval()
    
    all_preds = []
    all_labels = []
    
    with torch.no_grad():
        for batch in dataloader:
            text = batch['text'].to(device)
            audio = batch['audio'].to(device)
            visual = batch['visual'].to(device)
            labels = batch['labels'].cpu().numpy()
            
            outputs = model(text, audio, visual)
            preds = torch.argmax(outputs, dim=1).cpu().numpy()
            
            all_preds.extend(preds)
            all_labels.extend(labels)
    
    # 计算7类准确率
    acc7 = accuracy_score(all_labels, all_preds)
    
    # 计算2类准确率（将-3~-1合并为负类，1~3合并为正类，0忽略）
    binary_labels = [1 if x > 0 else 0 for x in all_labels if x != 0]
    binary_preds = [1 if x > 0 else 0 for x, l in zip(all_preds, all_labels) if l != 0]
    acc2 = accuracy_score(binary_labels, binary_preds)
    
    # 计算F1分数
    f1 = f1_score(all_labels, all_preds, average='weighted')
    
    # 计算MAE
    mae = mean_absolute_error(all_labels, all_preds)
    
    return {
        'acc7': acc7,
        'acc2': acc2,
        'f1': f1,
        'mae': mae
    }

在实际实验中，早期融合模型和晚期融合模型的典型结果对比如下：