从零实现多模态特征工程:Python实战图像、语音、文本向量化

当你第一次听说"特征向量"这个词时,是否觉得它神秘又遥不可及?实际上,每个开发者都能亲手实现这些看似高深的技术。本文将带你用纯Python代码,不依赖任何现成机器学习框架,一步步实现图像、语音和文本的特征向量转换。这不是又一个调包教程,而是一次真正理解底层原理的深度实践。

1. 图像特征向量:超越像素的数学之美

图像处理新手常犯的错误是直接将像素矩阵输入模型。让我们从更本质的角度出发,实现两种经典特征提取方法:方向梯度直方图(HOG)和局部二值模式(LBP)。

1.1 手动实现HOG特征

HOG的核心思想是捕捉图像的边缘和纹理信息。以下是完整实现步骤:

import numpy as np
from scipy.ndimage import convolve

def compute_gradients(image):
    """计算图像x和y方向的梯度"""
    sobel_x = np.array([[-1, 0, 1], [-2, 0, 2], [-1, 0, 1]])
    sobel_y = np.array([[1, 2, 1], [0, 0, 0], [-1, -2, -1]])
    
    grad_x = convolve(image, sobel_x)
    grad_y = convolve(image, sobel_y)
    
    magnitude = np.sqrt(grad_x**2 + grad_y**2)
    angle = np.arctan2(grad_y, grad_x) * (180 / np.pi) % 180
    
    return magnitude, angle

def compute_hog_features(image, cell_size=8, block_size=2, bins=9):
    """完整HOG特征计算实现"""
    # 预处理:Gamma校正
    image = np.sqrt(image / np.max(image))
    
    # 计算梯度
    mag, ang = compute_gradients(image)
    
    # 初始化特征向量
    height, width = image.shape
    x_cells = width // cell_size
    y_cells = height // cell_size
    hog_features = np.zeros((y_cells, x_cells, bins))
    
    # 构建细胞单元直方图
    for y in range(y_cells):
        for x in range(x_cells):
            cell_mag = mag[y*cell_size:(y+1)*cell_size, 
                          x*cell_size:(x+1)*cell_size]
            cell_ang = ang[y*cell_size:(y+1)*cell_size,
                          x*cell_size:(x+1)*cell_size]
            
            hist = np.zeros(bins)
            for i in range(cell_size):
                for j in range(cell_size):
                    bin_idx = int(cell_ang[i,j] / (180 / bins))
                    hist[bin_idx % bins] += cell_mag[i,j]
            
            hog_features[y,x] = hist
    
    # 块归一化
    norm_features = []
    for y in range(y_cells - block_size + 1):
        for x in range(x_cells - block_size + 1):
            block = hog_features[y:y+block_size, x:x+block_size].ravel()
            norm = np.linalg.norm(block) + 1e-5
            norm_block = block / norm
            norm_features.extend(norm_block)
    
    return np.array(norm_features)

提示:HOG特征对光照变化具有鲁棒性,特别适合行人检测等任务。实际应用中,cell_size通常取8x8像素,block_size为2x2个cell。

1.2 LBP纹理特征实现

局部二值模式(LBP)能有效捕捉纹理信息,以下是简明实现:

def local_binary_pattern(image, radius=1, neighbors=8):
    """LBP特征完整实现"""
    height, width = image.shape
    lbp = np.zeros_like(image, dtype=np.uint8)
    
    for y in range(radius, height-radius):
        for x in range(radius, width-radius):
            center = image[y,x]
            pattern = 0
            for n in range(neighbors):
                angle = 2 * np.pi * n / neighbors
                xn = x + int(radius * np.cos(angle))
                yn = y - int(radius * np.sin(angle))
                pattern |= (image[yn,xn] >= center) << n
            
            lbp[y,x] = pattern
    
    # 计算直方图作为特征向量
    hist, _ = np.histogram(lbp.ravel(), bins=256, range=(0, 256))
    return hist / (hist.sum() + 1e-7)  # 归一化

这两种方法各有千秋,实际应用中常组合使用:

特征类型 维度 计算复杂度 适用场景
HOG 较高 形状识别
LBP 较低 纹理分析

2. 语音特征工程:从声波到数学表示

语音处理远比想象中复杂。我们将实现完整的MFCC提取流程,包括预加重、分帧、加窗等关键步骤。

2.1 音频预处理三部曲

import numpy as np
from scipy.fftpack import dct

def pre_emphasis(signal, alpha=0.97):
    """预加重滤波器,提升高频分量"""
    return np.append(signal[0], signal[1:] - alpha * signal[:-1])

def framing(signal, sample_rate, frame_size=0.025, frame_stride=0.01):
    """将音频信号分帧处理"""
    frame_length = int(round(frame_size * sample_rate))
    frame_step = int(round(frame_stride * sample_rate))
    signal_length = len(signal)
    num_frames = int(np.ceil(float(np.abs(signal_length - frame_length)) / frame_step))
    
    pad_length = num_frames * frame_step + frame_length
    padding = np.zeros((pad_length - signal_length))
    padded_signal = np.concatenate((signal, padding))
    
    indices = np.tile(np.arange(0, frame_length), (num_frames, 1)) + \
              np.tile(np.arange(0, num_frames * frame_step, frame_step), (frame_length, 1)).T
    frames = padded_signal[indices.astype(np.int32, copy=False)]
    
    return frames * np.hamming(frame_length)

def compute_power_spectrum(frames, nfft=512):
    """计算功率谱"""
    mag_frames = np.absolute(np.fft.rfft(frames, nfft))
    return ((1.0 / nfft) * (mag_frames ** 2))

2.2 梅尔滤波器组实现

def mel_to_hz(mel):
    """梅尔频率转换为Hz"""
    return 700 * (10**(mel / 2595.0) - 1)

def hz_to_mel(hz):
    """Hz转换为梅尔频率"""
    return 2595 * np.log10(1 + hz / 700.0)

def get_filter_banks(nfilt=26, nfft=512, sample_rate=16000):
    """创建梅尔滤波器组"""
    low_freq_mel = hz_to_mel(0)
    high_freq_mel = hz_to_mel(sample_rate/2)
    
    mel_points = np.linspace(low_freq_mel, high_freq_mel, nfilt+2)
    hz_points = mel_to_hz(mel_points)
    
    bin = np.floor((nfft + 1) * hz_points / sample_rate)
    fbank = np.zeros((nfilt, int(np.floor(nfft / 2 + 1))))
    
    for m in range(1, nfilt + 1):
        f_m_minus = int(bin[m - 1])
        f_m = int(bin[m])
        f_m_plus = int(bin[m + 1])
        
        for k in range(f_m_minus, f_m):
            fbank[m - 1, k] = (k - bin[m - 1]) / (bin[m] - bin[m - 1])
        for k in range(f_m, f_m_plus):
            fbank[m - 1, k] = (bin[m + 1] - k) / (bin[m + 1] - bin[m])
    
    return fbank

2.3 完整MFCC提取流程

def extract_mfcc(signal, sample_rate=16000, nfilt=26, ncep=13):
    """完整MFCC特征提取实现"""
    # 预处理
    emphasized_signal = pre_emphasis(signal)
    frames = framing(emphasized_signal, sample_rate)
    pow_frames = compute_power_spectrum(frames)
    
    # 梅尔滤波器组
    filter_banks = get_filter_banks(nfilt=nfilt, nfft=512, sample_rate=sample_rate)
    filter_banks = np.dot(pow_frames, filter_banks.T)
    filter_banks = np.where(filter_banks == 0, np.finfo(float).eps, filter_banks)
    filter_banks = 20 * np.log10(filter_banks)  # dB
    
    # DCT变换得到MFCC
    mfcc = dct(filter_banks, type=2, axis=1, norm='ortho')[:, 1:(ncep+1)]
    
    # 倒谱均值归一化
    mfcc -= (np.mean(mfcc, axis=0) + 1e-8)
    
    return mfcc

注意:实际应用中,MFCC通常取13-26个系数,前13个包含大部分语音特征信息。采样率一般设为16kHz,帧长为25ms,帧移为10ms。

3. 文本特征工程:从词袋到语义向量

文本向量化经历了从统计方法到神经方法的演进。我们将实现TF-IDF和Word2Vec两种经典方法。

3.1 手动实现TF-IDF

import math
from collections import defaultdict

def tokenize(text):
    """简单分词函数"""
    return text.lower().split()

def compute_tf(docs):
    """计算词频(TF)"""
    tf_dict = []
    vocab = set()
    
    for doc in docs:
        tokens = tokenize(doc)
        tf = defaultdict(int)
        for token in tokens:
            tf[token] += 1
            vocab.add(token)
        
        # 归一化
        max_count = max(tf.values()) if tf else 1
        for token in tf:
            tf[token] /= max_count
        
        tf_dict.append(tf)
    
    return tf_dict, vocab

def compute_idf(docs, vocab):
    """计算逆文档频率(IDF)"""
    idf = {}
    N = len(docs)
    
    for term in vocab:
        doc_count = sum(1 for doc in docs if term in tokenize(doc))
        idf[term] = math.log(N / (doc_count + 1))
    
    return idf

def compute_tfidf(docs):
    """完整TF-IDF实现"""
    tf_dict, vocab = compute_tf(docs)
    idf = compute_idf(docs, vocab)
    
    tfidf_vectors = []
    for tf in tf_dict:
        vector = []
        for term in sorted(vocab):
            vector.append(tf.get(term, 0) * idf[term])
        tfidf_vectors.append(vector)
    
    return np.array(tfidf_vectors)

3.2 实现简易Word2Vec

import numpy as np
from collections import deque

class SimpleWord2Vec:
    def __init__(self, vocab_size, embedding_dim=100, window_size=2):
        self.W1 = np.random.randn(vocab_size, embedding_dim) * 0.01
        self.W2 = np.random.randn(embedding_dim, vocab_size) * 0.01
        self.window = window_size
    
    def softmax(self, x):
        e_x = np.exp(x - np.max(x))
        return e_x / e_x.sum(axis=0)
    
    def forward(self, x):
        h = np.dot(self.W1.T, x)
        u = np.dot(self.W2.T, h)
        y_pred = self.softmax(u)
        return y_pred, h
    
    def train(self, texts, epochs=100, learning_rate=0.01):
        vocab = sorted(set(word for text in texts for word in text))
        word2idx = {word: i for i, word in enumerate(vocab)}
        
        for epoch in range(epochs):
            loss = 0
            for text in texts:
                for i, target_word in enumerate(text):
                    # 构建上下文窗口
                    context = deque(maxlen=2*self.window+1)
                    for j in range(i-self.window, i+self.window+1):
                        if j >=0 and j < len(text) and j != i:
                            context.append(text[j])
                    
                    # 准备输入输出
                    x = np.zeros(len(vocab))
                    x[word2idx[target_word]] = 1
                    
                    y_true = np.zeros(len(vocab))
                    for word in context:
                        y_true[word2idx[word]] = 1
                    
                    # 前向传播
                    y_pred, h = self.forward(x)
                    
                    # 反向传播
                    EI = y_pred - y_true
                    dW2 = np.outer(h, EI)
                    dW1 = np.outer(x, np.dot(self.W2, EI))
                    
                    # 参数更新
                    self.W1 -= learning_rate * dW1
                    self.W2 -= learning_rate * dW2
                    
                    loss += -np.sum(y_true * np.log(y_pred + 1e-10))
            
            print(f"Epoch {epoch}, Loss: {loss}")
        
        return self.W1

文本特征方法对比:

方法 维度 是否保留语义 计算复杂度 适用场景
TF-IDF 短文本分类
Word2Vec 可调节 语义相关任务

4. 多模态特征融合实战

当我们将不同模态的特征向量组合使用时,关键在于解决特征尺度不一致和维度差异问题。

4.1 特征标准化与降维

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

def normalize_features(feature_list):
    """特征标准化处理"""
    scaler = StandardScaler()
    normalized = []
    for feature in feature_list:
        if len(feature.shape) == 1:
            feature = feature.reshape(1, -1)
        normalized.append(scaler.fit_transform(feature))
    return normalized

def reduce_dimension(features, n_components=64):
    """PCA降维"""
    pca = PCA(n_components=n_components)
    reduced = []
    for feature in features:
        reduced.append(pca.fit_transform(feature))
    return reduced

4.2 多模态融合策略

def feature_fusion(image_feat, audio_feat, text_feat, method='concat'):
    """多模态特征融合"""
    if method == 'concat':
        # 简单拼接
        return np.concatenate([image_feat, audio_feat, text_feat], axis=1)
    elif method == 'weighted':
        # 加权平均
        weights = [0.4, 0.3, 0.3]  # 可调节权重
        min_len = min(len(image_feat), len(audio_feat), len(text_feat))
        weighted = (weights[0]*image_feat[:min_len] + 
                   weights[1]*audio_feat[:min_len] + 
                   weights[2]*text_feat[:min_len])
        return weighted
    elif method == 'attention':
        # 简易注意力机制
        features = np.stack([image_feat, audio_feat, text_feat])
        attention = np.exp(features) / np.sum(np.exp(features), axis=0)
        return np.sum(features * attention, axis=0)
    else:
        raise ValueError("Unknown fusion method")

融合方法效果对比:

方法 保持特性 计算复杂度 适用场景
拼接(concat) 最好 各模态特征同等重要
加权平均 中等 最低 已知模态重要性
注意力机制 最好 动态权重分配

在真实项目中,我通常会先尝试简单的拼接方法,然后根据模型表现逐步尝试更复杂的融合策略。记得在融合前务必将各模态特征标准化到相同尺度,否则数值范围大的特征会主导模型训练。

Logo

码道开发者社区,聚焦华为云码道 CodeArts 代码智能体,沉淀 Agent、Skill、鸿蒙开发实战内容,供开发者查阅资料、交流技术、分享工程实践

更多推荐