一、系統架構設計
A[數據采集] --> B[預處理模塊]
B --> C[特征提取]
C --> D[多模態融合]
D --> E[情緒分類]
E --> F[系統部署]
F --> G[用戶界面]
二、數據準備與處理
1. 數據收集
- 視頻數據：FER2013（靜態圖像）、RAVDESS（動態視頻）
- 音頻數據：CREMA-D、IEMOCAP
- 自定義采集：使用OpenCV+PyAudio實現同步采集
2. 數據預處理
視頻處理：
import cv2
def process_video(video_path):
? ? cap = cv2.VideoCapture(video_path)
? ? frames = []
? ? while cap.isOpened():
? ? ? ? ret, frame = cap.read()
? ? ? ? if not ret: break
? ? ? ? # 人臉檢測
? ? ? ? face = cv2.CascadeClassifier('haarcascade_frontalface_default.xml')
? ? ? ? gray = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY)
? ? ? ? faces = face.detectMultiScale(gray, 1.3, 5)
? ? ? ? # 裁剪和歸一化
? ? ? ? if len(faces) > 0:
? ? ? ? ? ? (x,y,w,h) = faces[0]
? ? ? ? ? ? roi = cv2.resize(gray[y:y+h, x:x+w], (128,128))
? ? ? ? ? ? frames.append(roi)
? ? return np.array(frames)
?
音頻處理：
import librosa
def extract_audio_features(audio_path):
? ? y, sr = librosa.load(audio_path, sr=16000)
? ? # 分幀處理（30ms窗口）
? ? frames = librosa.util.frame(y, frame_length=480, hop_length=160)
? ? # 提取MFCC特征
? ? mfcc = librosa.feature.mfcc(y=y, sr=sr, n_mfcc=40)
? ? # 動態特征拼接
? ? delta = librosa.feature.delta(mfcc)
? ? ddelta = librosa.feature.delta(mfcc, order=2)
? ? return np.concatenate([mfcc, delta, ddelta], axis=0)
?
3. 數據同步策略
- 使用FFmpeg提取視頻時間戳
- 動態時間規整（DTW）對齊音視頻序列
- 創建時間對齊的元數據文件
?
三、模型設計與訓練
1. 視覺分支（PyTorch實現）
import torch
from torchvision.models import resnet34
?
class VisualNet(nn.Module):
? ? def __init__(self):
? ? ? ? super().__init__()
? ? ? ? self.base = resnet34(pretrained=True)
? ? ? ? self.base.fc = nn.Identity() # 移除全連接層
? ? ? ? self.temporal = nn.LSTM(512, 256, bidirectional=True)
? ? ? ??
? ? def forward(self, x):
? ? ? ? # x: (B, T, C, H, W)
? ? ? ? B, T = x.shape[:2]
? ? ? ? x = x.view(B*T, *x.shape[2:])
? ? ? ? features = self.base(x) # (B*T, 512)
? ? ? ? features = features.view(B, T, -1)
? ? ? ? out, _ = self.temporal(features)
? ? ? ? return out[:, -1] # 取最后時刻輸出
?
2. 音頻分支
class AudioNet(nn.Module):
? ? def __init__(self):
? ? ? ? super().__init__()
? ? ? ? self.conv = nn.Sequential(
? ? ? ? ? ? nn.Conv1d(120, 64, 3, padding=1),
? ? ? ? ? ? nn.BatchNorm1d(64),
? ? ? ? ? ? nn.ReLU(),
? ? ? ? ? ? nn.MaxPool1d(2))
? ? ? ? self.lstm = nn.LSTM(64, 128, bidirectional=True)
? ? ? ??
? ? def forward(self, x):
? ? ? ? # x: (B, T, Features)
? ? ? ? x = x.permute(0,2,1) # (B, Features, T)
? ? ? ? x = self.conv(x)
? ? ? ? x = x.permute(2,0,1) # (T, B, Features)
? ? ? ? out, _ = self.lstm(x)
? ? ? ? return out[-1]
3. 多模態融合
注意力融合層：
class FusionModule(nn.Module):
? ? def __init__(self, v_dim, a_dim):
? ? ? ? super().__init__()
? ? ? ? self.v_proj = nn.Linear(v_dim, 256)
? ? ? ? self.a_proj = nn.Linear(a_dim, 256)
? ? ? ? self.attention = nn.MultiheadAttention(256, 4)
? ? ? ??
? ? def forward(self, v_feat, a_feat):
? ? ? ? v = self.v_proj(v_feat).unsqueeze(0) # (1,B,256)
? ? ? ? a = self.a_proj(a_feat).unsqueeze(0)
? ? ? ? combined = torch.cat([v, a], dim=0) # (2,B,256)
? ? ? ? attn_out, _ = self.attention(combined, combined, combined)
? ? ? ? return attn_out.mean(dim=0)
?
?四、訓練策略
1. 損失函數設計
class MultimodalLoss(nn.Module):
? ? def __init__(self):
? ? ? ? super().__init__()
? ? ? ? self.ce = nn.CrossEntropyLoss()
? ? ? ? self.kl = nn.KLDivLoss()
? ? ? ??
? ? def forward(self, pred, label, v_out, a_out):
? ? ? ? # 主損失
? ? ? ? main_loss = self.ce(pred, label)
? ? ? ? # 模態一致性損失
? ? ? ? p_v = F.log_softmax(v_out, dim=1)
? ? ? ? p_a = F.softmax(a_out, dim=1)
? ? ? ? consistency_loss = self.kl(p_v, p_a.detach())
? ? ? ? return main_loss + 0.5 * consistency_loss
?
2. 訓練技巧
- 分階段訓練：先單模態預訓練，再聯合微調
- 數據增強策略：
? - 視覺：隨機遮擋、色彩抖動
? - 音頻：添加噪聲、時移變換
- 優化器配置：
? optimizer = torch.optim.AdamW([
? ? ? {'params': visual_net.parameters(), 'lr': 1e-4},
? ? ? {'params': audio_net.parameters(), 'lr': 3e-4},
? ? ? {'params': fusion_module.parameters(), 'lr': 5e-4}
? ], weight_decay=1e-5)
五、實時處理與部署
1. 實時處理架構
import queue
from threading import Thread
?
class RealTimeProcessor:
? ? def __init__(self):
? ? ? ? self.video_queue = queue.Queue(maxsize=30)
? ? ? ? self.audio_queue = queue.Queue(maxsize=100)
? ? ? ??
? ? def video_capture(self):
? ? ? ? cap = cv2.VideoCapture(0)
? ? ? ? while True:
? ? ? ? ? ? ret, frame = cap.read()
? ? ? ? ? ? processed = process_frame(frame)
? ? ? ? ? ? self.video_queue.put(processed)
? ? ? ? ? ??
? ? def audio_capture(self):
? ? ? ? p = pyaudio.PyAudio()
? ? ? ? stream = p.open(format=pyaudio.paInt16, channels=1,
? ? ? ? ? ? ? ? ? ? ? ? rate=16000, input=True,
? ? ? ? ? ? ? ? ? ? ? ? frames_per_buffer=1024)
? ? ? ? while True:
? ? ? ? ? ? data = stream.read(1024)
? ? ? ? ? ? features = extract_features(data)
? ? ? ? ? ? self.audio_queue.put(features)
? ? ? ? ? ??
? ? def sync_processor(self):
? ? ? ? while True:
? ? ? ? ? ? # 動態時間對齊算法
? ? ? ? ? ? video_batch = self.get_video_window()
? ? ? ? ? ? audio_batch = self.get_audio_window()
? ? ? ? ? ? aligned_data = dtw_align(video_batch, audio_batch)
? ? ? ? ? ? yield aligned_data
?
2. 部署優化方案
- 使用TensorRT進行模型量化：
? trtexec --onnx=model.onnx --saveEngine=model.engine \
? ? ? ? ?--fp16 --workspace=2048
- 邊緣設備優化：
? import torch_tensorrt
? traced_model = torch.jit.trace(model, example_input)
? trt_model = torch_tensorrt.compile(traced_model,
? ? ? inputs= [torch_tensorrt.Input((1, 3, 128, 128),?
? ? ? ? ? ? ? ?torch_tensorrt.Input((1, 100, 120))],
? ? ? enabled_precisions= {torch.float16})
?
六、評估與調優
?1. 評估指標
from sklearn.metrics import f1_score, confusion_matrix
?
def evaluate(y_true, y_pred):
? ? acc = (y_true == y_pred).mean()
? ? f1 = f1_score(y_true, y_pred, average='macro')
? ? cm = confusion_matrix(y_true, y_pred)
? ? return {'accuracy': acc, 'f1': f1, 'confusion_matrix': cm}
?
2. 模型分析工具
import shap
?
def explain_sample(video, audio):
? ? explainer = shap.DeepExplainer(model)
? ? shap_values = explainer.shap_values([video, audio])
? ??
? ? # 可視化各模態貢獻度
? ? shap.image_plot(shap_values[0], video)
? ? shap.summary_plot(shap_values[1], audio)
?
七、系統集成方案
1. 服務端架構
from fastapi import FastAPI
from pydantic import BaseModel
app = FastAPI()
class Request(BaseModel):
? ? video_url: str
? ? audio_url: str
@app.post("/analyze")
async def analyze(data: Request):
? ? video = download_and_process(data.video_url)
? ? audio = process_audio(data.audio_url)
? ? with torch.no_grad():
? ? ? ? prediction = model(video, audio)
? ? return {"emotion": class_names[prediction.argmax()]}
?2. 前端界面示例
// React組件示例
function EmotionDetector() {
? const [result, setResult] = useState(null);
? const handleUpload = async (files) => {
? ? const formData = new FormData();
? ? formData.append('video', files[0]);
? ? formData.append('audio', files[1]);
? ? const res = await fetch('/analyze', {
? ? ? method: 'POST',
? ? ? body: formData
? ? });
? ? setResult(await res.json());
? };
? return (
? ? <div>
? ? ? <input type="file" onChange={e => handleUpload(e.target.files)} />
? ? ? {result && <EmotionChart data={result}/>}
? ? </div>
? );
}
八、挑戰解決方案
1. 模態異步問題：
? ?- 采用雙緩沖隊列+動態時間規整
? ?- 設置最大等待時延（200ms），超時使用插值補償
2. 噪聲處理：
? ?def denoise_audio(audio):
? ? ? ?return nr.reduce_noise(y=audio, sr=16000,?
? ? ? ? ? ? ? ? ? ? ? ? ? ?stationary=True,
? ? ? ? ? ? ? ? ? ? ? ? ? ?prop_decrease=0.8)? ? ? ? ? ? ? ??
? ?def enhance_video(frame):
? ? ? ?clahe = cv2.createCLAHE(clipLimit=3.0, tileGridSize=(8,8))
? ? ? ?return clahe.apply(frame)
3. 資源優化：
? ?- 使用模型蒸餾技術：
? ?distiller = Distiller(teacher=teacher_model, student=student_model)
? ?distiller.train_with_distillation(train_loader,?
? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ?alpha=0.3,?
? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ?temperature=4)
總結：
該方案完整覆蓋了從數據采集到部署的全流程，重點解決了多模態系統中的關鍵挑戰。實際部署時可根據硬件資源調整模型復雜度，推薦使用NVIDIA Jetson系列設備進行邊緣部署。
： Transformer 架構)
