SmolVLA数据预处理管道搭建：大规模图像-文本对清洗与标注-育师

SmolVLA数据预处理管道搭建：大规模图像-文本对清洗与标注

最近有不少朋友在尝试训练或微调自己的视觉语言模型，比如SmolVLA这类轻量级的架构。大家聊下来，发现最头疼的往往不是模型本身，而是数据——怎么搞到足够多、足够干净、标注又准确的图像-文本对。网上的公开数据集要么不够大，要么质量参差不齐，直接拿来用效果总差那么点意思。

我花了不少时间折腾这块，从零开始搭了一套数据预处理的流程，今天就跟大家分享一下我的做法。这套东西不是什么高深的理论，就是一套能跑通、能出活的工程实践。核心就四步：爬数据、洗数据、标数据、存数据。我会尽量用大白话讲清楚每一步怎么操作，遇到坑怎么绕过去，保证你看完就能动手试试。

1. 从哪找数据？网络爬虫入门与实践

刚开始你可能觉得数据来源是个大问题。其实互联网本身就是个巨大的宝库，关键在于怎么有规矩、有效率地把东西“拿”出来。这里我们不讨论那些有版权限制或者明确禁止爬取的内容，主要聚焦在那些允许合理使用的公开资源上，比如一些开放图库、知识共享平台。

1.1 选择合适的爬取目标

别一上来就想着把整个互联网都爬下来，那不现实，也容易出问题。我的建议是，先明确你的模型要解决什么问题，然后去找对口的、质量相对较高的数据源。

举个例子，如果你想训练一个能理解日常物品和场景的模型，像Flickr、Pexels、Unsplash这类有大量用户上传且附带描述性标签的图片网站就挺合适。它们的图片质量通常不错，而且很多都遵循知识共享许可协议。如果你想做更垂直的领域，比如医学影像配文本报告，那就得去找专业的开源数据集了。

关键点：质量优先于数量。一万张带精准描述的图片，比一百万张标题是“IMG_001.jpg”的图片有价值得多。

1.2 动手写一个简单的图片爬虫

理论说再多不如动手。我们用Python来写，因为它相关的库最丰富。这里我演示一个非常基础但完整的例子，目标是从一个模拟的图片列表页爬取图片和它的标题。

首先，确保你安装了必要的库：

pip install requests beautifulsoup4

假设我们要爬取的页面结构很简单，每张图片在一个<div class="image-item">里，图片链接在<img src="...">，标题在<p class="caption">里。

import requests from bs4 import BeautifulSoup import os import time def simple_image_crawler(base_url, save_dir): """ 一个简单的图片爬虫示例 base_url: 图片列表页的URL save_dir: 图片和文本的保存目录 """ # 创建保存目录 os.makedirs(save_dir, exist_ok=True) # 设置一个简单的请求头，模拟浏览器访问 headers = { 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36' } try: # 1. 获取网页内容 response = requests.get(base_url, headers=headers, timeout=10) response.raise_for_status() # 检查请求是否成功 soup = BeautifulSoup(response.content, 'html.parser') # 2. 找到所有图片项 image_items = soup.find_all('div', class_='image-item') for index, item in enumerate(image_items): # 3. 提取图片链接和标题 img_tag = item.find('img') caption_tag = item.find('p', class_='caption') if not img_tag or not caption_tag: print(f"第{index+1}项数据不完整，跳过") continue img_url = img_tag.get('src') # 处理可能是相对路径的URL if img_url.startswith('//'): img_url = 'https:' + img_url elif img_url.startswith('/'): img_url = 'https://example.com' + img_url # 替换成实际域名 caption_text = caption_tag.get_text(strip=True) # 4. 下载图片 try: img_data = requests.get(img_url, headers=headers, timeout=15).content # 生成文件名，可以用索引或标题的哈希值 img_filename = f"{index:04d}_{hash(caption_text) & 0xFFFFFFFF:08x}.jpg" img_path = os.path.join(save_dir, img_filename) with open(img_path, 'wb') as f: f.write(img_data) # 5. 保存对应的文本描述到单独的文件 txt_filename = f"{index:04d}_{hash(caption_text) & 0xFFFFFFFF:08x}.txt" txt_path = os.path.join(save_dir, txt_filename) with open(txt_path, 'w', encoding='utf-8') as f: f.write(caption_text) print(f"已保存: {img_filename} -> {caption_text[:50]}...") except Exception as e: print(f"下载 {img_url} 失败: {e}") # 礼貌性延迟，避免给服务器造成压力 time.sleep(0.5) except requests.RequestException as e: print(f"请求页面失败: {e}") # 使用示例 (请替换为实际可访问的URL) # simple_image_crawler('https://example.com/images', './downloaded_data')

需要注意的几个地方：

遵守robots.txt：在爬取任何网站前，先访问网站域名/robots.txt，看看对方允许爬取哪些内容。
设置延迟：time.sleep(0.5)这样的延迟是基本礼仪，避免高频请求把别人服务器搞垮。
错误处理：网络请求充满不确定性，一定要用try...except把可能出错的地方包起来。
保存对应关系：图片和文本一定要能对应上。这里用了相同的哈希前缀来命名文件，确保一对一的关联。

这只是个起点。真实情况要复杂得多，比如分页处理、动态加载的页面（可能需要用Selenium）、登录验证等等。但核心思路是一样的：解析页面结构，定位元素，提取内容，妥善保存。

2. 数据清洗：从“原材料”到“半成品”

爬下来的数据，我们称之为“原材料”，里面混杂着各种“杂质”：重复的图片、牛头不对马嘴的文本、低质量的图片、甚至是不相关的内容。这一步清洗的目的，就是把明显不合格的“杂质”过滤掉。

2.1 基于规则的快速过滤

先用一些简单快速的规则筛掉明显有问题的数据，这能大大减少后续复杂处理的计算量。

文本过滤：
- 长度过滤：描述太短的（比如少于3个单词）可能信息量不足，太长的（比如超过200个单词）可能包含了太多无关信息或本身就是一篇文章。
- 语言过滤：如果你的模型只针对中文或英文，可以用简单的词典或开源库（如langdetect）过滤掉其他语言的描述。
- 关键词黑名单：过滤掉包含“广告”、“点击这里”、“版权归...所有”等明显是广告或版权声明的文本。
图像过滤：
- 尺寸过滤：分辨率过低的图片（如小于224x224）对于现代模型来说可能信息不够。
- 长宽比过滤：极端长宽比（如超过1:10）的图片可能是横幅广告或者图标，不适合作为自然场景。
- 文件大小过滤：极小的文件可能是损坏的图片或图标。

import os from PIL import Image def rule_based_filtering(data_dir, min_text_words=3, max_text_words=200, min_img_size=224): """ 基于规则的初步过滤 """ filtered_pairs = [] # 假设图片是.jpg，文本是.txt，且文件名前缀相同（不包括扩展名） image_files = [f for f in os.listdir(data_dir) if f.endswith('.jpg')] for img_file in image_files: base_name = os.path.splitext(img_file)[0] txt_file = base_name + '.txt' txt_path = os.path.join(data_dir, txt_file) # 1. 检查文本文件是否存在且可读 if not os.path.exists(txt_path): continue with open(txt_path, 'r', encoding='utf-8', errors='ignore') as f: text = f.read().strip() # 2. 文本长度过滤 word_count = len(text.split()) if word_count < min_text_words or word_count > max_text_words: continue # 3. 简单关键词黑名单过滤 (示例) blacklist = ['advertisement', 'click here', 'copyright', '广告', '版权所有'] if any(bad_word in text.lower() for bad_word in blacklist): continue # 4. 图像基本属性过滤 img_path = os.path.join(data_dir, img_file) try: with Image.open(img_path) as img: width, height = img.size # 尺寸过滤 if width < min_img_size or height < min_img_size: continue # 极端长宽比过滤 aspect_ratio = max(width, height) / min(width, height) if aspect_ratio > 10: continue except Exception: # 无法打开的图片文件，跳过 continue # 通过所有规则，保留此数据对 filtered_pairs.append((img_file, txt_file, text)) print(f"原始数据量: {len(image_files)}") print(f"规则过滤后剩余: {len(filtered_pairs)}") return filtered_pairs

2.2 利用预训练模型进行语义清洗

规则过滤后，数据干净了一些，但还不够。比如，一张猫的图片，配文却是“今天天气真好”，这种图文不匹配的问题规则发现不了。这时候就需要用AI来对付AI了。

一个常用的方法是使用一个现成的、强大的图文匹配模型（比如CLIP）来给每个图文对打分。分数高的，说明图文相关性高；分数很低的，就可能是不匹配的“噪声”。

# 假设你已经安装了 transformers 和 torch # pip install transformers torch pillow from transformers import CLIPProcessor, CLIPModel from PIL import Image import torch def clip_based_filtering(data_dir, pair_list, threshold=0.2): """ 使用CLIP模型计算图文相似度，过滤低分对 pair_list: 经过规则过滤后的数据对列表，每个元素是(img_file, txt_file, text) threshold: 相似度阈值，低于此值则过滤 """ # 加载预训练的CLIP模型和处理器 model = CLIPModel.from_pretrained("openai/clip-vit-base-patch32") processor = CLIPProcessor.from_pretrained("openai/clip-vit-base-patch32") device = "cuda" if torch.cuda.is_available() else "cpu" model.to(device) model.eval() filtered_pairs = [] with torch.no_grad(): for img_file, txt_file, text in pair_list: img_path = os.path.join(data_dir, img_file) try: # 预处理图像和文本 image = Image.open(img_path).convert("RGB") inputs = processor(text=[text], images=image, return_tensors="pt", padding=True) inputs = {k: v.to(device) for k, v in inputs.items()} # 前向传播，获取图文相似度 outputs = model(**inputs) logits_per_image = outputs.logits_per_image # 图像到文本的相似度 similarity = logits_per_image.softmax(dim=1).cpu().numpy()[0][0] # 取概率值 if similarity >= threshold: filtered_pairs.append((img_file, txt_file, text, similarity)) else: print(f"过滤低相似度对: {img_file} - {text[:30]}... (分数: {similarity:.3f})") except Exception as e: print(f"处理 {img_file} 时出错: {e}") continue print(f"CLIP过滤后剩余: {len(filtered_pairs)}") # 可以按相似度排序，保留高质量数据 filtered_pairs.sort(key=lambda x: x[3], reverse=True) return [(p[0], p[1], p[2]) for p in filtered_pairs] # 返回时去掉分数

这个方法计算量比较大，但效果通常比纯规则好很多。你可以根据数据量和计算资源调整阈值。如果数据量巨大，可以考虑先对文本进行嵌入（embedding）聚类，对图片进行特征提取去重，再进行CLIP打分，这样效率更高。

3. 数据标注：效率与质量的平衡

清洗后的数据，图文基本是相关的，但描述的质量还有提升空间。网络爬取的描述可能很简短、不准确，或者风格不一。为了得到更高质量的训练数据，我们往往需要进行标注或重标注。全人工标注成本太高，这里介绍一种“半自动”的思路。

3.1 设计一个半自动标注工具

核心思想是：让模型先猜，人来复核和修改。我们用一个图像描述生成模型（比如BLIP、GIT）为每张图片生成一个候选描述，然后提供一个简单的界面让标注员在候选描述的基础上进行修正、润色或重写。

# 这是一个简化的命令行交互示例，实际应用可以做成Web界面 from transformers import BlipProcessor, BlipForConditionalGeneration from PIL import Image import os def generate_caption_with_blip(image_path): """使用BLIP模型为图片生成描述""" processor = BlipProcessor.from_pretrained("Salesforce/blip-image-captioning-base") model = BlipForConditionalGeneration.from_pretrained("Salesforce/blip-image-captioning-base") raw_image = Image.open(image_path).convert('RGB') inputs = processor(raw_image, return_tensors="pt") out = model.generate(**inputs, max_length=50) caption = processor.decode(out[0], skip_special_tokens=True) return caption def semi_auto_annotation_workflow(data_dir, pair_list, output_file='annotated_data.jsonl'): """ 半自动标注工作流示例 """ annotated_data = [] for img_file, txt_file, original_text in pair_list[:100]: # 先处理前100个做演示 img_path = os.path.join(data_dir, img_file) print(f"\n--- 图片: {img_file} ---") # 显示图片 (在实际GUI中) # Image.open(img_path).show() print(f"原始文本: {original_text}") # 1. 模型生成候选描述 candidate_caption = generate_caption_with_blip(img_path) print(f"模型生成: {candidate_caption}") # 2. 人工交互 print("\n请选择操作:") print("1. 接受模型生成描述") print("2. 在模型描述基础上修改") print("3. 重新手动输入描述") print("4. 跳过此图片") choice = input("请输入选择 (1/2/3/4): ").strip() final_caption = "" if choice == '1': final_caption = candidate_caption elif choice == '2': print(f"当前描述: {candidate_caption}") modification = input("请输入修改后的描述: ").strip() final_caption = modification if modification else candidate_caption elif choice == '3': final_caption = input("请输入新的描述: ").strip() elif choice == '4': print("跳过此图片。") continue if final_caption: # 保存标注结果 data_entry = { 'image_path': img_path, 'original_text': original_text, 'annotated_text': final_caption, 'image_file': img_file } annotated_data.append(data_entry) print(f"已保存标注: {final_caption}") # 将标注结果保存为JSON Lines格式，方便后续处理 import json with open(output_file, 'w', encoding='utf-8') as f: for item in annotated_data: f.write(json.dumps(item, ensure_ascii=False) + '\n') print(f"\n标注完成！结果已保存至 {output_file}") return annotated_data

这个流程将人工从“从零开始写描述”的重度劳动，变成了“审核和修正模型输出”的轻度劳动，效率能提升好几倍。你可以把这个命令行脚本扩展成一个带有图形界面（用Gradio或Streamlit很简单）的小工具，让标注体验更好。

3.2 制定标注规范与质量控制

即使有工具辅助，如果没有统一的规范，不同标注员产出的数据风格也会差异很大。在开始大规模标注前，最好能制定一个简单的指南：

描述要具体：避免“一张图片”这种泛泛之谈，描述图中的主体、动作、场景、颜色、材质等。
保持客观：描述图中可见的内容，而不是主观感受或推断（除非非常明显）。
语法正确：使用完整的句子，避免拼写错误。
风格一致：确定好是用短句还是长句，是口语化还是书面化，并保持一致。

可以定期抽样检查标注结果，对共性的问题进行反馈和校准，确保最终数据集的质量。

4. 数据格式化与划分：为训练做好准备

经过清洗和标注，我们得到了一份高质量的数据列表。最后一步，就是把它转换成模型训练时能直接“吃”下去的格式，并合理地分成训练集、验证集和测试集。

4.1 转换为标准数据集格式

SmolVLA这类模型训练时，通常期望数据是特定的格式。一种非常通用且简单的格式是JSON Lines（.jsonl），每行是一个独立的JSON对象，代表一个数据样本。

import json import random from sklearn.model_selection import train_test_split def create_final_dataset(annotated_data_list, output_dir='./final_dataset'): """ 将标注后的数据转换为标准格式并划分 annotated_data_list: 来自半自动标注流程的数据列表 """ os.makedirs(output_dir, exist_ok=True) # 准备数据条目 dataset_entries = [] for item in annotated_data_list: # 这里我们使用人工标注后的文本作为最终文本 # 你也可以选择保留原始文本，或者将两者结合 entry = { "image": item['image_file'], # 图片文件名 "text": item['annotated_text'] # 最终使用的文本 # 可以添加更多字段，如原始文本、来源URL等 } dataset_entries.append(entry) # 随机打乱数据 random.shuffle(dataset_entries) # 划分数据集 (例如 80%训练, 10%验证, 10%测试) train_val, test = train_test_split(dataset_entries, test_size=0.1, random_state=42) train, val = train_test_split(train_val, test_size=0.111, random_state=42) # 0.111 * 0.9 ≈ 0.1 print(f"总样本数: {len(dataset_entries)}") print(f"训练集: {len(train)}") print(f"验证集: {len(val)}") print(f"测试集: {len(test)}") # 保存为JSON Lines格式 def save_jsonl(data, filename): with open(os.path.join(output_dir, filename), 'w', encoding='utf-8') as f: for entry in data: f.write(json.dumps(entry, ensure_ascii=False) + '\n') save_jsonl(train, 'train.jsonl') save_jsonl(val, 'validation.jsonl') save_jsonl(test, 'test.jsonl') # 同时，将图片文件复制到对应的目录（可选，但更清晰） import shutil for split_name, split_data in [('train', train), ('validation', val), ('test', test)]: split_img_dir = os.path.join(output_dir, split_name, 'images') os.makedirs(split_img_dir, exist_ok=True) # 假设原始图片都在一个叫‘raw_images’的文件夹里 source_img_dir = './raw_images' for entry in split_data: src = os.path.join(source_img_dir, entry['image']) dst = os.path.join(split_img_dir, entry['image']) if os.path.exists(src): shutil.copy2(src, dst) print("数据集已成功创建并划分！") return output_dir

4.2 创建数据加载器

有了标准格式的数据集，在训练时就可以方便地加载了。这里以PyTorch为例，写一个简单的数据加载器。

import torch from torch.utils.data import Dataset from PIL import Image import json class ImageTextDataset(Dataset): """自定义图文对数据集类""" def __init__(self, jsonl_file, image_dir, transform=None): self.data = [] with open(jsonl_file, 'r', encoding='utf-8') as f: for line in f: self.data.append(json.loads(line.strip())) self.image_dir = image_dir self.transform = transform # 图像增强变换 def __len__(self): return len(self.data) def __getitem__(self, idx): item = self.data[idx] image_path = os.path.join(self.image_dir, item['image']) # 加载图像 image = Image.open(image_path).convert('RGB') if self.transform: image = self.transform(image) # 文本 text = item['text'] # 注意：这里返回的是原始文本，需要tokenizer在collate_fn中处理 return image, text # 使用示例 # from torchvision import transforms # transform = transforms.Compose([ # transforms.Resize((224, 224)), # transforms.ToTensor(), # ]) # train_dataset = ImageTextDataset('./final_dataset/train.jsonl', './final_dataset/train/images', transform=transform) # train_loader = torch.utils.data.DataLoader(train_dataset, batch_size=32, shuffle=True, collate_fn=custom_collate_fn)

这个Dataset类会读取我们之前生成的.jsonl文件，并根据文件名加载对应的图片。图像预处理（如缩放、归一化）和文本的tokenization通常在transform参数和自定义的collate_fn函数中完成，以适配不同的模型需求。

5. 总结与建议

走完这一整套流程，你会发现准备数据确实是个细致活，但每一步都有它的价值。爬虫帮你获取了原始的“矿石”，规则清洗进行了初步的“筛选”，预训练模型完成了深度的“提纯”，半自动标注实现了“精加工”，最后的格式化则是打包成标准“产品”。这套组合拳打下来，得到的数据集质量会比直接用原始网络数据好上一个档次。

实际做的时候，有几点心得可以分享。首先，别想着一口吃成胖子，先从一个小规模、高质量的数据子集开始实验你的整个流程，跑通了再扩大。其次，清洗和标注的阈值、规则都不是一成不变的，要根据你模型的实际表现来动态调整。如果模型在某些类别上表现不好，回头看看是不是那部分数据质量有问题。最后，工具只是工具，最重要的还是你对数据的理解和判断。多看看你的数据，有时候肉眼观察能发现自动流程发现不了的问题。

数据是模型的上限，花时间把数据处理好，后面模型训练会顺利很多。希望这套流程能给你提供一个清晰的起点，你可以根据自己的具体需求和资源，对每个环节进行加强或简化。