一、为什么需要多模态数据预处理
在实际开发中，我们对接文心一言这类多模态大模型时，经常要处理各种不同格式的数据输入。文本是文本，图片是图片，音频是音频，视频又是视频，它们的结构差异非常大。如果直接把原始数据扔给模型，往往会出现格式不对、尺寸不匹配、编码有问题等各种幺蛾子。
所以多模态数据预处理这一步就特别关键。它的作用就是把各种"五花八门"的原始数据，统一整理成模型能理解和处理的"标准件"。你可以把它想象成做菜之前的备菜环节——食材得洗干净、切好、码整齐，后面才能下锅。
这篇文章就跟大家聊聊，针对不同类型的数据格式，我们应该怎么去做预处理，才能和文心一言的接口对接得顺顺当当。
1.1 多模态数据面临的挑战
不同模态的数据在存储格式、数据结构、编码方式上都有很大差别。举几个例子：
- 文本数据可能有不同编码（UTF-8、GBK等），还可能包含特殊字符和换行符
- 图像数据有PNG、JPG、BMP、WebP等格式，分辨率各异
- 音频数据有WAV、MP3、FLAC等格式，采样率和声道数也不同
- 视频数据涉及编码方式（H.264、H.265等）、帧率、分辨率等多维参数
这些差异如果不提前处理，模型接口那边大概率会报错或者直接返回错误结果。
二、文本数据的预处理方案
文本看着最简单，但坑也不少。比如你从网页上抓下来的文本可能带有HTML标签，从本地文件读的可能编码不对，用户输入的文本可能包含表情符号或者特殊控制字符。
2.1 文本清洗与编码统一
下面这个例子展示了如何把各种来源的文本统一处理成干净的UTF-8编码字符串：
# 技术栈：Python 3.9+，使用标准库 + re 正则表达式
import re
import unicodedata
def preprocess_text(raw_text: str, source_encoding: str = None) -> str:
"""
文本预处理函数
raw_text: 原始文本字符串
source_encoding: 原始编码（如果从bytes解码则传入）
返回：清洗后的干净文本
"""
# 如果传入的是bytes类型，先按指定编码解码
if isinstance(raw_text, bytes):
enc = source_encoding or 'utf-8'
try:
raw_text = raw_text.decode(enc)
except UnicodeDecodeError:
# 编码不对就退回utf-8，忽略错误字符
raw_text = raw_text.decode('utf-8', errors='ignore')
# 第一步：去除HTML标签（适用于从网页抓取的内容）
clean_text = re.sub(r'<[^>]+>', '', raw_text)
# 第二步：去除多余的空格和换行
clean_text = re.sub(r'[ \t]+', ' ', clean_text)
clean_text = re.sub(r'\n{3,}', '\n\n', clean_text) # 最多保留两个换行
# 第三步：去除控制字符（保留常见空白字符）
clean_text = ''.join(
ch for ch in clean_text
if unicodedata.category(ch) != 'Cc' or ch in '\n\t'
)
# 第四步：首尾去空格
clean_text = clean_text.strip()
return clean_text
# ---------- 使用示例 ----------
# 模拟从网页抓取的脏文本
dirty_html = '<p>你好世界</p><div>这是<div>嵌套</div>内容</div>\n\n\n\n 多余空格 '
result = preprocess_text(dirty_html)
print("清洗前:", repr(dirty_html))
print("清洗后:", repr(result))
# 输出结果：'你好世界 这是嵌套内容 多余空格'
# 模拟GBK编码的字节数据
gbk_bytes = '这是一个GBK编码的文本'.encode('gbk')
result2 = preprocess_text(gbk_bytes, source_encoding='gbk')
print("GBK解码结果:", result2)
# 输出结果：'这是一个GBK编码的文本'
# 模拟包含控制字符的文本
control_text = '正常文本\x00\x01\x02特殊字符'
result3 = preprocess_text(control_text)
print("去控制符结果:", result3)
# 输出结果：'正常文本特殊字符'
2.2 分词处理
分词是把长文本切成有意义的片段，这对模型理解语义非常有用。中文尤其需要分词，因为中文不像英文那样天然用空格分隔单词。
# 技术栈：Python 3.9+，使用 jieba 分词库
import jieba
import jieba.analyse
def preprocess_for_tokenization(text: str, mode: str = 'detail') -> list:
"""
对文本进行分词预处理
text: 已清洗的文本
mode: 'cut'=简单切词, 'detail'=保留停用词标记
返回：分词后的token列表
"""
if mode == 'cut':
# 简单分词，直接返回词列表
return list(jieba.cut(text))
elif mode == 'detail':
# 详细模式，区分中文词、英文词、数字、标点
tokens = []
for word in jieba.cut(text):
if re.match(r'^[a-zA-Z]+$', word):
tokens.append({'word': word, 'type': 'english'})
elif re.match(r'^\d+$', word):
tokens.append({'word': word, 'type': 'number'})
elif re.match(r'^[\W_]+$', word):
tokens.append({'word': word, 'type': 'punctuation'})
else:
tokens.append({'word': word, 'type': 'chinese'})
return tokens
# ---------- 使用示例 ----------
sample_text = "2024年AI技术发展很快，GPT和文心一言都在进步"
# 简单分词
simple_tokens = preprocess_for_tokenization(sample_text, mode='cut')
print("简单分词:", simple_tokens)
# 输出: ['2024', '年', 'AI', '技术', '发展', '很', '快', '，', 'GPT', '和', '文心一言', '都', '在', '进步']
# 详细分词（带类型标记）
detailed_tokens = preprocess_for_tokenization(sample_text, mode='detail')
print("详细分词:", detailed_tokens)
# 输出包含每个词的词性和类型标签
# 提取关键词（用jieba的TF-IDF算法）
# 输出: ['AI技术', '文心一言', '发展', '2024', '进步']
三、图像数据的预处理方案
图像预处理是多模态处理中比较常见的一环。不同来源的图片尺寸、格式、通道数都不一样，需要统一处理后才能送给模型。
3.1 图像格式转换与尺寸调整
# 技术栈：Python 3.9+，使用 Pillow (PIL) 库
from PIL import Image
import os
import base64
def preprocess_image(
image_path: str = None,
image_bytes: bytes = None,
target_size: tuple = (512, 512),
target_format: str = 'RGB',
max_size_mb: float = 2.0
) -> dict:
"""
图像预处理函数
image_path: 图片文件路径
image_bytes: 图片二进制数据（二选一）
target_size: 目标尺寸 (宽, 高)
target_format: 目标颜色模式 (RGB/RGBA/Gray)
max_size_mb: 压缩后最大文件大小(MB)
返回：预处理后的图像信息字典
"""
# 加载图像
if image_path and os.path.exists(image_path):
img = Image.open(image_path)
original_path = image_path
elif image_bytes:
import io
img = Image.open(io.BytesIO(image_bytes))
original_path = "binary_input"
else:
raise ValueError("必须提供 image_path 或 image_bytes")
# 记录原始信息
original_info = {
'source': original_path,
'original_size': img.size, # (宽, 高)
'original_mode': img.mode, # RGB/RGBA/L 等
'original_format': img.format # PNG/JPEG/BMP 等
}
# 第一步：统一颜色模式
if img.mode not in ['RGB', 'RGBA', 'L']:
img = img.convert(target_format)
elif img.mode != target_format:
img = img.convert(target_format)
# 第二步：调整尺寸（保持宽高比）
if img.size != target_size:
img = img.resize(target_size, Image.LANCZOS)
# LANCZOS 是高质量的重采样算法，适合缩图
# 第三步：压缩输出，控制文件大小
output_buffer = io.BytesIO()
if target_format == 'RGB':
quality = 85 # JPEG质量参数
img.save(output_buffer, format='JPEG', quality=quality)
output_format = 'JPEG'
else:
img.save(output_buffer, format='PNG')
output_format = 'PNG'
# 如果文件太大，降低质量重新压缩
current_size = output_buffer.tell()
if current_size > max_size_mb * 1024 * 1024:
if target_format == 'RGB':
quality = max(10, quality - 30)
output_buffer.seek(0)
output_buffer.truncate()
img.save(output_buffer, format='JPEG', quality=quality)
output_buffer.seek(0)
image_data = output_buffer.read()
# 第四步：生成Base64编码（API调用常用）
base64_data = base64.b64encode(image_data).decode('utf-8')
result_info = {
'original_info': original_info,
'processed_size': img.size,
'processed_mode': img.mode,
'output_format': output_format,
'output_size_bytes': len(image_data),
'base64_data': base64_data,
'quality': quality if target_format == 'RGB' else 'N/A'
}
return result_info
# ---------- 使用示例 ----------
# 场景1：处理本地PNG图片
result = preprocess_image(
image_path='photo.png',
target_size=(512, 512),
target_format='RGB',
max_size_mb=2.0
)
print("原始尺寸:", result['original_info']['original_size'])
print("处理后尺寸:", result['processed_size'])
print("输出格式:", result['output_format'])
print("文件大小(KB):", round(result['output_size_bytes'] / 1024, 2))
# 场景2：处理网络下载的JPEG图片（假设已有bytes数据）
# image_bytes = requests.get(url).content
# result2 = preprocess_image(
# image_bytes=image_bytes,
# target_size=(224, 224), # 模型要求的输入尺寸
# target_format='RGB'
# )
# print("Base64长度:", len(result2['base64_data']))
3.2 图像预处理中的EXIF信息处理
手机拍摄的图片通常带有EXIF元数据，里面记录了拍摄设备、拍摄时间、GPS位置等信息。有时候我们需要清除这些信息以保护隐私。
# 技术栈：Python 3.9+，使用 Pillow (PIL) 库
from PIL import Image, ExifTags
import io
def strip_exif_info(image_path: str, save_path: str = None) -> str:
"""
清除图像中的EXIF元数据
image_path: 原始图片路径
save_path: 保存路径（为None则返回内存中的图片）
返回：处理后的图片路径或内存Buffer
"""
img = Image.open(image_path)
# 读取原始EXIF信息（用于日志记录）
exif_data = img._getexif()
if exif_data:
print("原始EXIF标签:")
for tag_id, value in exif_data.items():
tag_name = ExifTags.TAGS.get(tag_id, tag_id)
print(f" {tag_name}: {value}")
# 创建新图像（不携带EXIF）
clean_img = Image.new(img.mode, img.size)
clean_img.putdata(list(img.getdata()))
if save_path:
clean_img.save(save_path)
return save_path
else:
buffer = io.BytesIO()
clean_img.save(buffer, format=img.format)
buffer.seek(0)
return buffer
# ---------- 使用示例 ----------
# 处理手机拍摄的照片，清除GPS等隐私信息
# strip_exif_info('DSC_001.jpg', 'DSC_001_clean.jpg')
# 输出示例:
# 原始EXIF标签:
# Make: Apple
# Model: iPhone 15 Pro
# DateTime: 2024:03:15 14:30:22
# GPSLatitude: (31, 1234567)
# GPSLongitude: (121, 1234567)
四、音频数据的预处理方案
音频数据的特点是它是一个时间序列，包含采样率、声道数、位深等多个参数。不同的音频格式（WAV、MP3、FLAC）在这些参数上差异很大。
4.1 音频格式统一与重采样
# 技术栈：Python 3.9+，使用 librosa 库
import librosa
import numpy as np
def preprocess_audio(
audio_path: str,
target_sr: int = 16000,
mono: bool = True,
trim_silence: bool = True,
top_db: float = 20
) -> dict:
"""
音频预处理函数
audio_path: 音频文件路径
target_sr: 目标采样率（Hz），文心一言常用16000
mono: 是否转为单声道
trim_silence: 是否去除首尾静音
top_db: 静音判定阈值（dB）
返回：处理后的音频数据及元信息
"""
# 第一步：加载音频文件
# librosa默认以mono和22050Hz加载，所以我们先不指定
y_original, sr_original = librosa.load(audio_path, sr=None, mono=False)
original_info = {
'source': audio_path,
'original_sr': sr_original,
'original_duration': len(y_original) / sr_original, # 秒
'original_channels': 1 if y_original.ndim == 1 else y_original.shape[1]
}
# 第二步：如果需要，进行声道合并
if mono and y_original.ndim == 2:
y = y_original.mean(axis=1) # 多声道取平均转为单声道
else:
y = y_original
# 第三步：重采样到目标采样率
if sr_original != target_sr:
y = librosa.resample(y, orig_sr=sr_original, target_sr=target_sr)
current_sr = target_sr
# 第四步：可选的静音裁剪
if trim_silence:
y_trimmed, (start_frame, end_frame) = librosa.effects.trim(
y, top_db=top_db
)
start_time = start_frame / current_sr
end_time = end_frame / current_sr
trim_info = {
'trimmed': True,
'start': round(start_time, 3),
'end': round(end_time, 3),
'removed_duration': round(
original_info['original_duration'] - len(y_trimmed) / current_sr, 3
)
}
y = y_trimmed
else:
trim_info = {'trimmed': False}
# 第五步：归一化（将音量统一到一个范围）
y_normalized = librosa.util.normalize(y, norm=np.inf)
result_info = {
'original_info': original_info,
'processing_info': {
'target_sr': current_sr,
'mono': mono,
'trimmed': trim_info,
'normalized': True
},
'processed_duration': round(len(y_normalized) / current_sr, 3),
'audio_array': y_normalized, # numpy数组，范围[-1, 1]
'dtype': str(y_normalized.dtype)
}
return result_info
# ---------- 使用示例 ----------
# 场景：处理一段用户录音，准备送给语音识别API
result = preprocess_audio(
audio_path='user_recording.mp3',
target_sr=16000, # 目标采样率
mono=True, # 转单声道
trim_silence=True, # 去掉首尾静音
top_db=20 # 静音阈值
)
print("原始采样率:", result['original_info']['original_sr'])
print("处理后采样率:", result['processing_info']['target_sr'])
print("原始时长(秒):", result['original_info']['original_duration'])
print("处理后时长(秒):", result['processed_duration'])
print("裁剪信息:", result['processing_info']['trimmed'])
print("音频数据类型:", result['dtype'])
print("音频数据长度:", len(result['audio_array']))
# 如果要去保存为WAV文件：
# import soundfile as sf
# sf.write('processed.wav', result['audio_array'], result['processing_info']['target_sr'])
4.2 音频特征提取
除了原始波形数据，有时候还需要提取音频的特征向量，比如梅尔频谱图，这在音乐理解、语音分析场景中很常见。
# 技术栈：Python 3.9+，使用 librosa 库
import librosa
import numpy as np
def extract_audio_features(
audio_path: str,
sr: int = 16000,
n_mels: int = 80,
n_fft: int = 2048,
hop_length: int = 512
) -> dict:
"""
提取音频的多种特征
audio_path: 音频文件路径
sr: 采样率
n_mels: 梅尔滤波器数量
n_fft: FFT窗口大小
hop_length: 帧移长度
返回：包含多种特征的字典
"""
y, sr = librosa.load(audio_path, sr=sr)
features = {}
# 特征1：梅尔频谱图（最常用，模拟人耳感知）
melspectrogram = librosa.feature.melspectrogram(
y=y, sr=sr, n_mels=n_mels, n_fft=n_fft, hop_length=hop_length
)
melspec_db = librosa.power_to_db(melspectrogram, ref=np.max)
features['mel_spectrogram'] = {
'data': melspec_db,
'shape': melspec_db.shape, # (n_mels, frames)
'description': f'{n_mels} bands, {melspec_db.shape[1]} frames'
}
# 特征2：MFCC（梅尔频率倒谱系数，语音识别常用）
mfcc = librosa.feature.mfcc(y=y, sr=sr, n_mfcc=13)
features['mfcc'] = {
'data': mfcc,
'shape': mfcc.shape, # (13, frames)
'description': '13 MFCC coefficients'
}
# 特征3：RMS能量包络（衡量音量变化）
rms = librosa.feature.rms(y=y, hop_length=hop_length)[0]
features['rms_energy'] = {
'data': rms,
'shape': rms.shape,
'description': 'Root Mean Square energy per frame'
}
# 特征4：过零率（判断是否有人声/乐器声）
zcr = librosa.feature.zero_crossing_rate(y, frame_length=n_fft, hop_length=hop_length)[0]
features['zero_crossing_rate'] = {
'data': zcr,
'shape': zcr.shape,
'description': 'Zero Crossing Rate per frame'
}
return features
# ---------- 使用示例 ----------
features = extract_audio_features('speech_sample.wav')
print("=== 音频特征提取结果 ===")
for name, info in features.items():
print(f"{name}: shape={info['shape']}, {info['description']}")
# 典型输出:
# === 音频特征提取结果 ===
# mel_spectrogram: shape=(80, 128), 80 bands, 128 frames
# mfcc: shape=(13, 128), 13 MFCC coefficients
# rms_energy: shape=(128,), Root Mean Square energy per frame
# zero_crossing_rate: shape=(128,), Zero Crossing Rate per frame
五、视频数据的预处理方案
视频处理是所有模态里最复杂的，因为它同时涉及图像和时间两个维度。一帧一帧的图片，再加上时序关系，处理起来工作量很大。
5.1 视频抽帧与预处理
# 技术栈：Python 3.9+，使用 OpenCV (cv2) + Pillow
import cv2
import numpy as np
from PIL import Image
import os
def preprocess_video(
video_path: str,
frame_interval: int = 1,
target_fps: float = None,
target_resolution: tuple = (640, 480),
max_frames: int = 100,
output_format: str = 'jpg',
quality: int = 90
) -> dict:
"""
视频预处理函数：抽帧并统一处理
video_path: 视频文件路径
frame_interval: 每N帧抽取一帧
target_fps: 目标帧率（与frame_interval二选一）
target_resolution: 目标分辨率 (宽, 高)
max_frames: 最多抽取的帧数
output_format: 输出帧格式
quality: 图像压缩质量(1-100)
返回：抽帧结果及视频元信息
"""
cap = cv2.VideoCapture(video_path)
# 读取视频元信息
original_fps = cap.get(cv2.CAP_PROP_FPS)
total_frames = int(cap.get(cv2.CAP_PROP_FRAME_COUNT))
original_width = int(cap.get(cv2.CAP_PROP_FRAME_WIDTH))
original_height = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT))
total_duration = total_frames / original_fps if original_fps > 0 else 0
original_info = {
'source': video_path,
'fps': original_fps,
'total_frames': total_frames,
'resolution': (original_width, original_height),
'duration_seconds': round(total_duration, 2)
}
# 确定实际抽帧间隔
if target_fps and original_fps:
actual_interval = max(1, int(original_fps / target_fps))
else:
actual_interval = frame_interval
# 开始抽帧
frames_info = []
frame_idx = 0
frames_saved = 0
while frames_saved < max_frames:
ret, frame = cap.read()
if not ret:
break
if frame_idx % actual_interval == 0:
# 第一步：调整尺寸
if (original_width, original_height) != target_resolution:
frame = cv2.resize(frame, target_resolution, interpolation=cv2.INTER_AREA)
# 第二步：调整质量并编码
encode_param = [cv2.IMWRITE_JPEG_QUALITY, quality]
encode_format = f'.{output_format}'
# 保存到临时文件（如果需要）
# success, buffer = cv2.imencode(encode_format, frame, encode_param)
# if success:
# buffer_path = f'frame_{frame_idx:05d}{encode_format}'
# with open(buffer_path, 'wb') as f:
# f.write(buffer)
# 计算帧的简单特征
gray = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY)
mean_brightness = np.mean(gray)
std_brightness = np.std(gray)
frame_info = {
'frame_index': frame_idx,
'timestamp': round(frame_idx / original_fps, 3) if original_fps > 0 else 0,
'resolution': target_resolution,
'mean_brightness': round(float(mean_brightness), 2),
'std_brightness': round(float(std_brightness), 2),
'data_shape': frame.shape # (高, 宽, 通道数)
}
frames_info.append(frame_info)
frames_saved += 1
frame_idx += 1
cap.release()
result = {
'original_info': original_info,
'processing_info': {
'actual_interval': actual_interval,
'target_resolution': target_resolution,
'output_format': output_format,
'quality': quality
},
'extracted_frames': frames_info,
'total_extracted': len(frames_info)
}
return result
# ---------- 使用示例 ----------
# 场景：从一个会议视频中抽取关键帧，分析内容
result = preprocess_video(
video_path='meeting_record.mp4',
target_fps=1.0, # 每秒抽1帧
target_resolution=(640, 480),
max_frames=60, # 最多抽60帧
quality=85
)
print("=== 视频原始信息 ===")
print(f"时长: {result['original_info']['duration_seconds']}秒")
print(f"原始帧数: {result['original_info']['total_frames']}")
print(f"原始分辨率: {result['original_info']['resolution']}")
print(f"原始帧率: {result['original_info']['fps']:.1f}fps")
print()
print("=== 处理信息 ===")
print(f"抽帧间隔: 每{result['processing_info']['actual_interval']}帧")
print(f"目标分辨率: {result['processing_info']['target_resolution']}")
print(f"抽取帧数: {result['total_extracted']}")
print()
print("=== 前3帧信息 ===")
for f in result['extracted_frames'][:3]:
print(f"帧{f['frame_index']}: 时间={f['timestamp']}s, "
f"亮度均值={f['mean_brightness']}, "
f"分辨率={f['resolution']}")
5.2 视频场景变化检测
有时候我们需要知道视频中的"镜头切换"时刻，这在短视频分析和监控系统中非常有用。
# 技术栈：Python 3.9+，使用 OpenCV (cv2)
import cv2
import numpy as np
def detect_scene_changes(
video_path: str,
threshold: float = 30.0,
min_scene_duration: float = 0.5
) -> dict:
"""
检测视频中的场景切换
video_path: 视频文件路径
threshold: 帧间差异阈值（越大越不容易判定为切换）
min_scene_duration: 最短场景时长（秒）
返回：场景切换点和场景列表
"""
cap = cv2.VideoCapture(video_path)
fps = cap.get(cv2.CAP_PROP_FPS)
scene_changes = [] # 场景切换的时间点
previous_frame = None
frame_idx = 0
current_scene_start = 0
while True:
ret, frame = cap.read()
if not ret:
break
# 将帧缩小以加速处理，同时保留足够的细节
small_frame = cv2.resize(frame, (160, 120))
if previous_frame is not None:
# 计算帧间差异（归一化直方图差异）
hist_prev = cv2.calcHist([previous_frame], [0], None, [256], [0, 256])
hist_curr = cv2.calcHist([small_frame], [0], None, [256], [0, 256])
# 使用巴氏距离计算直方图差异
diff = cv2.compareHist(hist_prev, hist_curr, cv2.HISTCMP_BHATTACHARYYA)
if diff > threshold:
change_time = frame_idx / fps
scene_duration = change_time - current_scene_start
# 只有满足最小时长的切换才记录
if scene_duration >= min_scene_duration:
scene_changes.append({
'frame_index': frame_idx,
'time_seconds': round(change_time, 3),
'histogram_diff': round(diff, 4),
'previous_scene_duration': round(scene_duration, 3)
})
current_scene_start = change_time
previous_frame = small_frame.copy()
frame_idx += 1
cap.release()
total_duration = frame_idx / fps if fps > 0 else 0
return {
'total_frames': frame_idx,
'total_duration': round(total_duration, 2),
'scene_changes': scene_changes,
'num_scenes': len(scene_changes) + 1 # +1是因为最后一个场景没有后续切换
}
# ---------- 使用示例 ----------
# 场景：分析一段Vlog视频，找出所有镜头切换
analysis = detect_scene_changes(
video_path='my_vlog.mp4',
threshold=0.3, # 归一化后的阈值
min_scene_duration=0.5 # 至少0.5秒的场景才记录
)
print("=== 场景变化检测结果 ===")
print(f"总时长: {analysis['total_duration']}秒")
print(f"检测到的场景数: {analysis['num_scenes']}")
print(f"镜头切换次数: {len(analysis['scene_changes'])}")
print()
for i, sc in enumerate(analysis['scene_changes'][:5]):
print(f"切换 {i+1}: 时间={sc['time_seconds']}s, "
f"前一个场景时长={sc['previous_scene_duration']}s, "
f"差异值={sc['histogram_diff']}")
六、混合多模态数据的统一预处理框架
在实际项目中，我们往往需要同时处理多种模态的数据。比如一个视频分析任务可能需要同时处理视频帧、音频轨道和字幕文本。这就需要搭建一个统一的预处理框架。
6.1 统一预处理调度器
# 技术栈：Python 3.9+，使用标准库 + 前面各章节的工具函数
import os
import json
import time
from dataclasses import dataclass, field, asdict
from typing import List, Dict, Any, Optional
from pathlib import Path
# ---------- 定义预处理任务的数据结构 ----------
@dataclass
class PreprocessTask:
"""一个预处理任务的定义"""
task_id: str
modality: str # text/image/audio/video
file_path: str # 输入文件路径
output_dir: str # 输出目录
params: Dict[str, Any] = field(default_factory=dict)
priority: int = 0 # 优先级，数字越大越优先
@dataclass
class PreprocessResult:
"""一个预处理任务的结果"""
task_id: str
success: bool
output_path: Optional[str]
info: Dict[str, Any]
error_message: Optional[str] = None
processing_time: float = 0.0
class MultimodalPreprocessor:
"""
多模态统一预处理器
负责调度和管理不同类型的预处理任务
"""
def __init__(self, base_output_dir: str = './preprocessed_output'):
self.base_output_dir = base_output_dir
os.makedirs(base_output_dir, exist_ok=True)
self.tasks: List[PreprocessTask] = []
self.results: List[PreprocessResult] = []
def add_text_task(self, task_id: str, text_content: str,
output_filename: str = None, **params) -> PreprocessTask:
"""添加文本预处理任务"""
if not output_filename:
output_filename = f'{task_id}_text.json'
task = PreprocessTask(
task_id=task_id,
modality='text',
file_path='__TEXT_CONTENT__' + json.dumps(text_content, ensure_ascii=False),
output_dir=os.path.join(self.base_output_dir, 'text'),
params=params
)
self.tasks.append(task)
return task
def add_image_task(self, task_id: str, image_path: str, **params) -> PreprocessTask:
"""添加图像预处理任务"""
task = PreprocessTask(
task_id=task_id,
modality='image',
file_path=image_path,
output_dir=os.path.join(self.base_output_dir, 'image'),
params=params
)
self.tasks.append(task)
return task
def add_audio_task(self, task_id: str, audio_path: str, **params) -> PreprocessTask:
"""添加音频预处理任务"""
task = PreprocessTask(
task_id=task_id,
modality='audio',
file_path=audio_path,
output_dir=os.path.join(self.base_output_dir, 'audio'),
params=params
)
self.tasks.append(task)
return task
def add_video_task(self, task_id: str, video_path: str, **params) -> PreprocessTask:
"""添加视频预处理任务"""
task = PreprocessTask(
task_id=task_id,
modality='video',
file_path=video_path,
output_dir=os.path.join(self.base_output_dir, 'video'),
params=params
)
self.tasks.append(task)
return task
def run_all_tasks(self, verbose: bool = True) -> List[PreprocessResult]:
"""执行所有预处理任务"""
# 按优先级排序
self.tasks.sort(key=lambda t: -t.priority)
for task in self.tasks:
os.makedirs(task.output_dir, exist_ok=True)
start_time = time.time()
try:
if task.modality == 'text':
result = self._process_text(task)
elif task.modality == 'image':
result = self._process_image(task)
elif task.modality == 'audio':
result = self._process_audio(task)
elif task.modality == 'video':
result = self._process_video(task)
else:
raise ValueError(f"不支持的模态类型: {task.modality}")
elapsed = time.time() - start_time
pre_result = PreprocessResult(
task_id=task.task_id,
success=True,
output_path=result.get('output_path'),
info=result.get('info', {}),
processing_time=round(elapsed, 3)
)
except Exception as e:
elapsed = time.time() - start_time
pre_result = PreprocessResult(
task_id=task.task_id,
success=False,
output_path=None,
info={},
error_message=str(e),
processing_time=round(elapsed, 3)
)
self.results.append(pre_result)
if verbose:
status = "✓" if pre_result.success else "✗"
print(f"[{status}] {task.task_id} ({task.modality}): "
f"{pre_result.processing_time:.2f}s" +
(f" - {pre_result.error_message}" if not pre_result.success else ""))
# 保存汇总报告
self._save_report()
return self.results
def _process_text(self, task: PreprocessTask) -> Dict[str, Any]:
"""文本处理内部方法"""
text_content = json.loads(task.file_path.replace('__TEXT_CONTENT__', ''))
output_path = os.path.join(task.output_dir, f'{task.task_id}_cleaned.json')
# 调用前面定义的文本清洗函数
cleaned = preprocess_text(text_content)
# 保存结果
with open(output_path, 'w', encoding='utf-8') as f:
json.dump({
'original_length': len(text_content),
'cleaned_length': len(cleaned),
'cleaned_text': cleaned
}, f, ensure_ascii=False, indent=2)
return {
'output_path': output_path,
'info': {
'original_chars': len(text_content),
'cleaned_chars': len(cleaned)
}
}
def _process_image(self, task: PreprocessTask) -> Dict[str, Any]:
"""图像处理内部方法"""
result = preprocess_image(
image_path=task.file_path,
target_size=tuple(task.params.get('target_size', (512, 512))),
target_format=task.params.get('target_format', 'RGB'),
max_size_mb=task.params.get('max_size_mb', 2.0)
)
output_path = os.path.join(task.output_dir, f'{task.task_id}_processed.json')
with open(output_path, 'w', encoding='utf-8') as f:
# 不保存base64到文件（太大），只保存元信息
info_copy = {k: v for k, v in result.items() if k != 'base64_data'}
info_copy['base64_length'] = len(result['base64_data'])
json.dump(info_copy, f, ensure_ascii=False, indent=2)
return {'output_path': output_path, 'info': info_copy}
def _process_audio(self, task: PreprocessTask) -> Dict[str, Any]:
"""音频处理内部方法"""
result = preprocess_audio(
audio_path=task.file_path,
target_sr=task.params.get('target_sr', 16000),
mono=task.params.get('mono', True),
trim_silence=task.params.get('trim_silence', True)
)
output_path = os.path.join(task.output_dir, f'{task.task_id}_meta.json')
meta_info = {
'original_info': result['original_info'],
'processing_info': result['processing_info'],
'processed_duration': result['processed_duration'],
'sample_count': len(result['audio_array'])
}
with open(output_path, 'w', encoding='utf-8') as f:
json.dump(meta_info, f, ensure_ascii=False, indent=2)
return {'output_path': output_path, 'info': meta_info}
def _process_video(self, task: PreprocessTask) -> Dict[str, Any]:
"""视频处理内部方法"""
result = preprocess_video(
video_path=task.file_path,
target_fps=task.params.get('target_fps', 1.0),
target_resolution=tuple(task.params.get('target_resolution', (640, 480))),
max_frames=task.params.get('max_frames', 50),
quality=task.params.get('quality', 85)
)
output_path = os.path.join(task.output_dir, f'{task.task_id}_frames_info.json')
with open(output_path, 'w', encoding='utf-8') as f:
json.dump(result, f, ensure_ascii=False, indent=2)
return {'output_path': output_path, 'info': {
'original_duration': result['original_info']['duration_seconds'],
'extracted_frames': result['total_extracted'],
'target_resolution': result['processing_info']['target_resolution']
}}
def _save_report(self):
"""保存处理报告"""
report_path = os.path.join(self.base_output_dir, 'processing_report.json')
report = {
'total_tasks': len(self.results),
'successful': sum(1 for r in self.results if r.success),
'failed': sum(1 for r in self.results if not r.success),
'total_time': round(sum(r.processing_time for r in self.results), 2),
'results': [asdict(r) for r in self.results]
}
with open(report_path, 'w', encoding='utf-8') as f:
json.dump(report, f, ensure_ascii=False, indent=2)
print(f"\n📊 处理报告已保存: {report_path}")
# ---------- 使用示例：批量处理多种模态 ----------
# 创建预处理器
processor = MultimodalPreprocessor(base_output_dir='./my_preprocessed_data')
# 添加各种预处理任务
processor.add_text_task(
task_id='t_001',
text_content='<p>这是一段<em>需要清洗的</em>文本内容</p> 多余空格\n\n\n\n\n尾部',
priority=2
)
processor.add_image_task(
task_id='img_001',
image_path='./images/photo_001.jpg',
target_size=(512, 512),
target_format='RGB',
max_size_mb=1.5,
priority=3
)
processor.add_audio_task(
task_id='aud_001',
audio_path='./audio/recording_001.mp3',
target_sr=16000,
mono=True,
trim_silence=True,
priority=1
)
processor.add_video_task(
task_id='vid_001',
video_path='./videos/interview_001.mp4',
target_fps=2.0,
target_resolution=(640, 480),
max_frames=30,
quality=80,
priority=1
)
# 执行所有任务
results = processor.run_all_tasks(verbose=True)
# 查看汇总
print(f"\n总计: {len(results)} 个任务")
print(f"成功: {sum(1 for r in results if r.success)}")
print(f"失败: {sum(1 for r in results if not r.success)}")
七、应用场景
7.1 智能客服多轮对话系统
在智能客服场景中，用户可能通过文字、语音、截图等多种方式描述问题。系统需要把这些不同格式的信息统一预处理后送给大模型，才能给出准确的回答。比如用户上传了一张报错截图，系统需要先对图片做格式统一和尺寸调整，再让模型去分析截图中的错误信息。
7.2 教育领域的多模态内容分析
教育平台上有很多视频课程，教师希望系统能自动分析视频中的知识点分布、学生提问的语音内容、课件中的文字信息。这时候就需要一个统一的多模态预处理框架，把视频抽帧、音频转写、文本提取都处理好，再交给模型分析。
7.3 医疗影像辅助诊断
医生上传的患者检查数据可能包括X光片、CT序列、病理报告文本、超声视频等。这些数据格式各异，需要分别做预处理后才能整合分析。预处理环节要特别注意数据的标准化处理和隐私信息脱敏。
7.4 内容审核与风控
社交平台需要审核用户上传的图文、短视频等内容。多模态预处理在这里的作用是统一把各种格式的内容提取成模型可分析的表征，然后再做违规内容检测。
八、技术优缺点
8.1 优点
统一接口降低复杂度：通过统一预处理框架，不同模态的数据处理被封装成标准化的接口，业务代码不需要关心底层的格式差异，调用方式简单统一。
数据质量有保障：预处理阶段做了清洗、归一化、格式转换等操作，能有效降低后续模型处理出错的概率。尤其是编码统一和尺寸调整这类处理，能避免很多"低级错误"。
可扩展性强：框架设计成可插拔的模式，添加新的模态类型或者新的预处理策略时，只需要新增对应的处理器类，不需要修改已有代码。
批处理能力：支持批量任务提交和并行处理，在处理大量数据时效率更高。处理报告能清晰记录每个任务的执行情况和耗时，方便排查问题。
8.2 缺点
内存消耗较大：尤其是视频和音频处理，如果一次性加载太多帧或者太长的音频到内存中，可能会触发内存溢出。需要做好分批处理和流式读取。
预处理耗时不可控：不同数据文件的复杂程度差异很大，一个10分钟的4K视频和一个30秒的手机录像，处理时间可能差几十倍。在实际部署时需要考虑超时机制和资源调度。
格式兼容性仍有盲区：虽然有各种库支持，但总会遇到一些冷门或者非标准的格式。比如某些特殊编码的PDF嵌入图片、某些厂商自定义的音频封装格式等，可能需要额外的适配工作。
预处理质量影响下游效果：预处理环节的参数设置（比如采样率、压缩质量、抽帧间隔）会直接影响后续模型的理解效果。参数设得不好，可能丢失关键信息或者引入噪声。
九、注意事项
9.1 文件编码要提前确认
处理文本和文件时，编码问题是最高频的坑。不要假设所有文件都是UTF-8编码，尤其是从Windows系统或者某些老旧系统导出的文件，很可能是GBK编码。建议先用file命令或者读取文件的BOM头来判断编码。
9.2 大文件要分块处理
视频和音频文件经常有几百MB甚至几个GB，不要试图一次性全部读入内存。对于视频，可以边读边处理帧数据；对于音频，可以分段加载并逐段处理。OpenCV的VideoCapture本身就是流式读取的，但要注意不要把所有帧都缓存在列表里。
9.3 异常处理不能少
实际项目中，输入文件可能是损坏的、路径可能不存在、权限可能不够。每个预处理步骤都应该有完善的异常捕获和错误报告机制，不要因为一个文件出错就导致整个批处理任务崩溃。
9.4 预处理参数的选择需要实验验证
图像压缩质量设80还是90，音频采样率用16000还是22050，视频抽帧间隔用1秒还是2秒——这些参数没有"标准答案"，需要根据具体的业务场景和模型要求做实验对比。建议先在小规模数据集上跑对比实验，找到效果和质量之间的最佳平衡点。
9.5 注意版本依赖
多模态处理涉及的库比较多（Pillow、OpenCV、librosa等），不同版本的API可能不兼容。建议在项目中使用虚拟环境或者Docker容器来锁定依赖版本，避免"在我机器上能跑"的问题。
9.6 隐私数据脱敏
图像中可能包含人脸和敏感信息，音频中可能包含个人隐私，文本中可能包含电话号码等敏感数据。在预处理阶段就应该做脱敏处理，尤其是医疗和金融等敏感行业。
十、文章总结
多模态数据预处理看似是一个"杂活"，但其实是大模型应用开发中非常关键的一环。预处理做得好，后续模型的表现会更稳定、更准确；预处理做得不好，再强大的模型也可能给出错误结果。
这篇文章介绍了文本、图像、音频、视频四种常见模态的预处理方案，并给出了一个统一调度框架的完整实现。核心思路是：针对不同格式的数据，先做清洗和标准化，再提取有用的特征，最后以统一的格式输出，供下游模型使用。
实际操作中，建议大家根据具体的业务需求来裁剪预处理流程——不是每个场景都需要做全部的预处理步骤。比如纯文本对话场景就不需要处理图像和音频；简单的图像问答可能只需要格式转换和尺寸调整，不需要做复杂的特征提取。
记住一个原则：预处理的目的是让数据"干净、规范、对模型友好"，而不是为了处理而处理。每一个预处理步骤都应该有其存在的意义。
评论
围绕“文心一言多模态数据处理适配：针对不同格式输入的预处理方案”参与讨论