verl数据预处理技巧：多模态输入这样处理

verl数据预处理技巧：多模态输入这样处理

verl 是一个专为大型语言模型（LLM）后训练设计的强化学习（RL）框架，由字节跳动火山引擎团队开源，是 HybridFlow 论文的工程落地实现。它不仅支持标准文本 RLHF，更关键的是——原生支持多模态输入的端到端强化学习训练流程。这意味着图像、表格、代码片段、甚至混合格式的输入，都能被统一纳入 RL 数据流中参与策略优化。

但真正决定训练效果上限的，往往不是模型结构本身，而是数据怎么进、怎么对齐、怎么喂给模型。尤其在多模态场景下，文本 prompt 与图像特征如何协同编码？不同来源的图像分辨率、格式、标注方式如何归一化？视觉 token 与语言 token 的序列长度如何平衡？这些都不是“自动处理”能解决的问题。

本文不讲原理推导，不堆参数配置，而是聚焦一个工程师每天真实面对的问题：当你手上有带图的 Geometry3K 几何题、带截图的客服对话、带 UI 界面的自动化任务样本时，该怎么写 preprocessing 逻辑，才能让 verl 正确加载、高效训练、稳定收敛？我们将从 verl 的数据契约出发，拆解四类典型多模态预处理模式，并给出可直接复用的代码片段和避坑指南。

1. verl 多模态数据契约：理解它的“期待”

verl 并不强制要求你把图像转成 base64 或嵌入到 JSON 字段里。相反，它定义了一套轻量但严谨的数据字段契约（Data Contract），只要你的预处理函数输出符合这个结构，后续的 rollout、reward 计算、梯度更新就能无缝衔接。

1.1 核心字段语义解析

verl 的多模态训练样本（datadict）必须包含以下关键键，它们共同构成模型输入的完整上下文：

关键提醒：prompt中的content字段永远只承载纯文本描述。哪怕你写"请分析这张图：<img src='xxx.jpg'>"，verl 也不会解析 HTML 标签——它会把整段字符串当作文本 token 输入，导致模型“看见”的只是字符<img，而非图像内容。真正的图像，必须通过images字段显式传递。

1.2 verl 如何消费images字段？

不同 VLM 模型对images的处理方式不同，verl 通过data.image_key配置项（默认"images"）告诉系统从哪个字段取图像。其内部流程如下：

1. 加载阶段：若images是str列表（如["/path/to/img1.png", "/path/to/img2.jpg"]），verl 调用PIL.Image.open()加载为 RGB 模式 PIL Image；
2. 预处理阶段：根据模型配置（如Qwen2.5-VL的image_processor）进行 resize、normalize、patch embedding；
3. 拼接阶段：将图像 embedding 与文本 embedding 在 sequence 维度拼接，形成[CLS] + text_tokens + image_patches + [SEP]类似结构；
4. 对齐校验：verl 会检查prompt中是否包含占位符（如<image>），并在拼接时插入图像 token 位置。因此，你的promptcontent 中需显式写入<image>占位符，且数量必须与images列表长度严格一致。

# 正确：1 张图 → 1 个 <image> 占位符 data = { "prompt": [ {"role": "user", "content": "这张图展示了一个什么几何图形？<image>"} ], "images": ["/data/geo/fig1.png"], "extra_info": {"original_size": (1024, 768)} } # 正确：2 张图 → 2 个 <image> 占位符（顺序必须对应） data = { "prompt": [ {"role": "user", "content": "对比这两张图：<image> 和 <image>，哪个图形面积更大？"} ], "images": ["/data/geo/fig_a.png", "/data/geo/fig_b.png"] } # ❌ 错误：占位符数量与 images 长度不匹配 → verl 报错 data = { "prompt": [{"role": "user", "content": "<image>"}], "images": [] # 缺少图像 }

1.3 预处理函数的签名与返回规范

verl 的数据集构建依赖Dataset.map()，你的预处理函数必须返回符合上述契约的data字典。函数签名建议如下：

def process_multimodal_example(example: Dict, idx: int) -> Dict: """ 多模态样本预处理主函数 Args: example: 原始数据样本（来自 HuggingFace Dataset 或自定义 JSONL） idx: 样本索引（可用于日志或 debug） Returns: Dict: 符合 verl 多模态契约的 data 字典 """ # 步骤1：提取并清洗文本 prompt # 步骤2：加载/转换图像数据 # 步骤3：构造 reward_model 和 extra_info # 步骤4：返回最终 data 字典 return data

核心原则：预处理函数应是纯函数（Pure Function）——无副作用、不修改全局状态、输入相同则输出确定。这保证了分布式训练时数据加载的一致性。

2. 四类典型多模态预处理实战

下面以 verl 官方支持的三大 VLM（Qwen2.5-VL、Kimi-VL、自定义 VLM）为背景，给出四类高频场景的完整预处理方案。所有代码均可直接集成到Dataset.map()流程中。

2.1 场景一：本地图像路径 → verl 可加载格式（最常用）

适用场景：你的数据集是 JSONL 文件，每行含"image_path": "xxx.png"和"question": "..."字段。

挑战：路径可能相对、损坏、格式不一（jpg/png/webp）；需确保 verl 能稳定加载。

解决方案：封装健壮的图像加载器，自动处理异常并提供 fallback。

from pathlib import Path from PIL import Image import logging logger = logging.getLogger(__name__) def load_and_validate_image(image_path: str, max_retries: int = 3) -> Image.Image: """ 健壮加载单张图像，支持重试和常见错误处理 Returns: PIL.Image.Image: RGB 模式图像，失败时返回空白图（避免 pipeline 中断） """ path = Path(image_path) if not path.exists(): logger.warning(f"Image path not found: {image_path}") # 返回 1x1 白色占位图，避免训练中断 return Image.new("RGB", (1, 1), color="white") for attempt in range(max_retries): try: img = Image.open(path).convert("RGB") # 验证尺寸（防极端小图） if img.size[0] < 32 or img.size[1] < 32: logger.warning(f"Image too small: {image_path} -> {img.size}") return Image.new("RGB", (224, 224), color="gray") return img except Exception as e: logger.warning(f"Failed to load {image_path}, attempt {attempt+1}: {e}") if attempt == max_retries - 1: return Image.new("RGB", (224, 224), color="red") time.sleep(0.1) # 避免快速重试冲击磁盘 return Image.new("RGB", (224, 224), color="red") def preprocess_local_images(example: Dict, idx: int) -> Dict: """本地路径多模态预处理""" # 1. 构造 prompt（含 <image> 占位符） question = example.get("question", "").strip() if not question: question = "请描述这张图片。" prompt = [{"role": "user", "content": f"{question}<image>"}] # 2. 加载图像（支持单图/多图） image_paths = example.get("image_paths", []) # 支持列表 if isinstance(image_paths, str): image_paths = [image_paths] images = [] for path in image_paths: img = load_and_validate_image(path) images.append(img) # verl 会自动调用 image_processor # 3. 构造 reward_model（示例：规则奖励） ground_truth = example.get("answer", "") reward_model = {"style": "rule", "ground_truth": ground_truth} # 4. extra_info 存放原始信息 extra_info = { "original_image_paths": image_paths, "sample_id": example.get("id", f"idx_{idx}"), "source_dataset": example.get("dataset", "unknown") } return { "prompt": prompt, "images": images, # ← verl 期望的格式 "reward_model": reward_model, "extra_info": extra_info } # 使用示例 # dataset = load_dataset("json", data_files="geo3k_train.jsonl") # processed_ds = dataset.map(preprocess_local_images, num_proc=8)

2.2 场景二：Base64 编码图像 → verl 可加载格式（API/爬虫数据）

适用场景：数据来自 Web API、爬虫或标注平台，图像以 base64 字符串形式存储在"image_b64"字段。

挑战：base64 解码失败、非图像 MIME 类型、超大尺寸内存溢出。

解决方案：安全解码 + 内存限制 + 格式标准化。

import base64 import io from PIL import Image def decode_base64_image(b64_str: str, max_size_mb: int = 10) -> Image.Image: """ 安全解码 base64 图像，限制最大内存占用 Args: b64_str: base64 编码字符串（不含 data:image/...;base64, 前缀） max_size_mb: 最大允许解码后图像内存（MB） Returns: PIL.Image.Image: RGB 图像，失败时返回灰色占位图 """ try: # 移除可能的 data URL 前缀 if b64_str.startswith("data:"): b64_str = b64_str.split(",", 1)[-1] # 估算解码后内存（粗略：假设 RGBA 4 bytes/pixel） # base64 编码膨胀约 4/3，故原始字节数 ≈ len * 3/4 approx_raw_bytes = len(b64_str) * 3 // 4 if approx_raw_bytes > max_size_mb * 1024 * 1024: raise ValueError(f"Base64 too large: ~{approx_raw_bytes//1024//1024}MB > {max_size_mb}MB") # 解码 img_bytes = base64.b64decode(b64_str) img_buffer = io.BytesIO(img_bytes) img = Image.open(img_buffer).convert("RGB") # 再次检查尺寸（防恶意超大图） if img.size[0] > 4096 or img.size[1] > 4096: img = img.resize((2048, 2048), Image.LANCZOS) return img except Exception as e: logger.error(f"Base64 decode failed: {e}") return Image.new("RGB", (224, 224), color="gray") def preprocess_base64_images(example: Dict, idx: int) -> Dict: """Base64 多模态预处理""" b64_list = example.get("image_b64", []) if isinstance(b64_list, str): b64_list = [b64_list] images = [] for b64_str in b64_list: img = decode_base64_image(b64_str) images.append(img) # prompt 构造同上... prompt = [{"role": "user", "content": f"{example.get('text', '请描述')}<image>"}] return { "prompt": prompt, "images": images, "reward_model": {"style": "rule", "ground_truth": example.get("label", "")}, "extra_info": {"b64_source": True, "sample_idx": idx} }

2.3 场景三：OCR 文本 + 图像 → 多模态增强（提升图文对齐）

适用场景：图像含大量文字（如文档、截图、UI 界面），单纯靠 VLM 理解易出错。需将 OCR 提取的文本作为辅助 prompt。

挑战：OCR 结果噪声大、排版丢失、与图像区域未对齐。

解决方案：将 OCR 文本作为独立消息加入prompt，并用extra_info传递结构化 OCR 结果供 Reward Model 使用。

def preprocess_ocr_enhanced(example: Dict, idx: int) -> Dict: """OCR 增强型多模态预处理""" # 1. 加载图像（同场景一） img = load_and_validate_image(example["image_path"]) # 2. 获取 OCR 文本（假设已预处理好，存于 example["ocr_text"]） ocr_text = example.get("ocr_text", "").strip() if ocr_text: # 将 OCR 文本作为 system message 注入，引导模型关注文字内容 prompt = [ {"role": "system", "content": f"以下是从图像中识别出的文字内容，请结合图像一起理解：\n{ocr_text}"}, {"role": "user", "content": f"{example.get('question', '请回答')}<image>"} ] else: prompt = [{"role": "user", "content": f"{example.get('question', '请回答')}<image>"}] # 3. 将原始 OCR 结构化数据存入 extra_info，供 Reward Model 使用 ocr_structured = example.get("ocr_boxes", []) # 如 [{"text": "Submit", "bbox": [10,20,100,40]}] return { "prompt": prompt, "images": [img], "reward_model": {"style": "rule", "ground_truth": example.get("answer")}, "extra_info": { "ocr_text": ocr_text, "ocr_boxes": ocr_structured, "has_ocr": bool(ocr_text) } } # 效果提示：此方法在 UI 自动化、文档问答等任务中，显著提升 verl 对图文混合指令的理解准确率。

2.4 场景四：动态生成图像 → verl 实时处理（工具调用链路）

适用场景：你的 RL 任务涉及工具调用（如 SandboxFusion 执行代码生成图表），需将工具输出的图像实时喂给 Actor 模型。

挑战：图像生成是异步过程，需在Interaction生命周期内完成；不能提前写死路径。

解决方案：利用extra_info传递生成逻辑，在Interaction.generate_response()中动态执行并注入images。

# 在 Interaction 类中（如 Gsm8kInteraction） async def generate_response(self, instance_id: str, messages: list[dict], **kwargs) -> tuple[bool, str, float, dict]: # ... 原有逻辑 ... # 新增：检查是否需要动态生成图像 if self._instance_dict[instance_id].get("need_dynamic_image", False): # 1. 从 extra_info 获取生成参数 gen_params = self._instance_dict[instance_id]["gen_params"] # 2. 调用工具生成图像（如 matplotlib 画图） img_pil = await self._generate_chart(gen_params) # 自定义方法 # 3. 将 PIL Image 直接注入到当前 step 的 images 中 # verl 的 rollout 会自动处理 kwargs["images"] = [img_pil] # ... 后续调用 actor model ... return should_terminate, response, reward, {} # 预处理函数只需标记需求，不实际生成 def preprocess_dynamic_image(example: Dict, idx: int) -> Dict: """为动态图像生成任务准备数据""" return { "prompt": [{"role": "user", "content": f"{example['query']}<image>"}], # images 字段留空，由 Interaction 运行时填充 "reward_model": {"style": "rule", "ground_truth": example["answer"]}, "extra_info": { "need_dynamic_image": True, "gen_params": example.get("chart_params", {}) } }

3. 关键避坑指南：那些让 verl 训练崩溃的细节

即使代码逻辑正确，几个微小的疏忽也会导致 verl 训练报错、OOM 或结果诡异。以下是高频踩坑点及修复方案。

3.1 图像尺寸不一致 → Batch 内存爆炸

现象：训练时 GPU 显存占用忽高忽低，CUDA out of memory随机出现。

原因：verl 默认按 batch 内最大图像尺寸做 padding。若一个 batch 中有 1024x1024 和 50x50 的图，小图会被 pad 到大图尺寸，浪费大量显存。

解决方案：预处理时统一缩放，或启用 verl 的动态分辨率。

# 方案1：预处理时统一 resize（推荐用于固定任务） def resize_for_batch(image: Image.Image, target_size: int = 448) -> Image.Image: w, h = image.size scale = min(target_size / w, target_size / h) new_w, new_h = int(w * scale), int(h * scale) return image.resize((new_w, new_h), Image.LANCZOS) # 方案2：配置 verl 启用动态分辨率（需模型支持） # 在训练命令中添加： # +actor_rollout_ref.model.image_processor_kwargs.dynamic_resize=True \ # actor_rollout_ref.model.image_processor_kwargs.target_size=448

3.2<image>占位符缺失或错位 → 模型“看不见”图

现象：loss 不下降，生成结果完全忽略图像内容，reward 为 0。

原因：prompt中未写<image>，或写了但数量与images不匹配。

验证方法：在预处理函数末尾加日志：

num_placeholders = sum(content.count("<image>") for msg in prompt for content in [msg.get("content", "")]) assert num_placeholders == len(images), f"Mismatch: {num_placeholders} vs {len(images)}"

3.3extra_info传入非法类型 → Ray 序列化失败

现象：ray.exceptions.RayTaskError，提示Object cannot be serialized。

原因：extra_info中存了不可序列化的对象（如open()的 file handle、lambda 函数、数据库连接）。

解决方案：extra_info只存基本类型（str, int, float, list, dict, bool, None）或 numpy array（verl 支持）。

# ❌ 错误 extra_info = {"file_handle": open("log.txt")} # 正确 extra_info = {"log_file": "log.txt", "processed_at": time.time()}

3.4 多图顺序错乱 → 图文语义错配

现象：模型对多图比较任务（如“图A和图B哪个更清晰？”）回答错误率高。

原因：images列表顺序与prompt中<image>占位符顺序不一致。

解决方案：严格按<image>出现顺序组织images，并在预处理中加断言：

# 在 preprocess_xxx 函数中 placeholders = [] for msg in prompt: if "content" in msg: placeholders.extend([p for p in msg["content"].split("<image>") if p.strip()]) # 占位符数量 = len(prompt_content) 中 <image> 个数 assert len(images) == len(placeholders), "Image count must match <image> count"

4. 性能优化：让多模态预处理快起来

预处理速度直接影响整体训练吞吐。针对 verl 的多模态 pipeline，我们推荐以下优化组合。

4.1 并行加载与缓存

from datasets import Features, Value, Sequence, Image as HFImage # 定义 HuggingFace Dataset Features，启用内置图像缓存 features = Features({ "image_path": Value("string"), "question": Value("string"), "answer": Value("string"), "image": HFImage() # ← 启用 HF 的图像缓存和解码优化 }) # 加载时自动缓存解码后的 PIL Image dataset = load_dataset( "json", data_files="data.jsonl", features=features, keep_in_memory=False # 大数据集用磁盘缓存 )

4.2 预计算图像特征（高级）

对于固定图像集，可预计算并缓存 VLM 的image_processor输出，跳过训练时重复计算：

# 预处理脚本：生成 .arrow 文件 def precompute_image_features(example): img = load_and_validate_image(example["image_path"]) # 使用 verl 的 image_processor（需 import） from verl.trainer.utils import get_image_processor processor = get_image_processor("Qwen/Qwen2.5-VL-7B-Instruct") pixel_values = processor(images=img, return_tensors="pt").pixel_values return {"pixel_values": pixel_values.squeeze(0)} # 移除 batch dim # dataset = dataset.map(precompute_image_features, num_proc=16) # dataset.save_to_disk("precomputed_geo3k")

然后在训练时，preprocess函数直接读取pixel_values字段，不再调用image_processor。

总结

verl 的多模态能力强大，但它的威力能否释放，80% 取决于你如何准备数据。本文没有罗列晦涩的 API 参数，而是直击工程落地中最常遇到的四个核心问题：

• 契约理解：明确prompt与images的分工，<image>占位符是图文对齐的唯一桥梁；
• 路径处理：用健壮加载器应对损坏路径、格式混杂，避免 pipeline 中断；
• 动态增强：通过 OCR 文本注入或工具链路生成，让多模态输入更“聪明”；
• 避坑实践：从尺寸对齐、序列一致性到序列化安全，覆盖训练崩溃的绝大多数原因。

记住一个原则：verl 的预处理不是“把数据变漂亮”，而是“把意图变明确”。每一次对prompt的精炼、对<image>的精准放置、对extra_info的合理利用，都是在为 RL 的 reward signal 铺设更清晰的路径。

当你下次面对一堆带图的数据时，不必再纠结“verl 能不能支持”，而是思考：“我该如何用最简洁的代码，把图像的语义、文本的意图、任务的目标，全部无损地编码进那几个 key-value 字段里？”

这才是多模态 RL 工程师真正的起点。

--- > **获取更多AI镜像** > > 想探索更多AI镜像和应用场景？访问 [CSDN星图镜像广场](https://ai.csdn.net/?utm_source=mirror_blog_end)，提供丰富的预置镜像，覆盖大模型推理、图像生成、视频生成、模型微调等多个领域，支持一键部署。