企业网站建设兴田德润电话,公司产品营销广告宣传,wordpress 安装语言设置中文乱码,群辉做网站服务器配置Ostrakon-VL-8B数据处理实战#xff1a;利用Python进行大规模训练数据清洗 最近有不少朋友在尝试对Ostrakon-VL-8B这类视觉语言大模型进行微调#xff0c;结果发现效果总是不太理想。很多时候#xff0c;问题并不出在模型本身或者训练代码上#xff0c;而是源头——数据没…Ostrakon-VL-8B数据处理实战利用Python进行大规模训练数据清洗最近有不少朋友在尝试对Ostrakon-VL-8B这类视觉语言大模型进行微调结果发现效果总是不太理想。很多时候问题并不出在模型本身或者训练代码上而是源头——数据没准备好。我见过太多团队花了大把时间调参、优化训练策略最后才发现是训练数据里混入了大量低质量、重复甚至错误的样本。这就好比用浑浊的水源去酿酒再怎么调整工艺也酿不出好酒。今天咱们就来聊聊数据清洗这个看似基础却至关重要的环节。我会用最直白的Python代码带你走一遍大规模图文数据清洗的完整流程。无论你是数据工程师、算法研究员还是刚接触模型微调的新手这篇文章都能帮你避开那些常见的坑。1. 为什么数据清洗如此关键在开始写代码之前我们先搞清楚一件事为什么数据清洗对Ostrakon-VL-8B的微调这么重要Ostrakon-VL-8B是个视觉语言模型它需要同时理解图像和对应的文本描述。如果你喂给它的数据是混乱的——图片模糊不清、文本描述牛头不对马嘴、或者同一个样本重复出现几十次——模型就会学到错误的知识。想象一下你正在教一个孩子认识动物。如果你给他看一张模糊的猫的照片却告诉他是“狗”还反复给他看同样的错误例子他当然学不会正确的知识。模型训练也是同样的道理。数据清洗的目标很明确从原始的海量数据中筛选出那些清晰、准确、多样的图文对。清晰指的是图片质量要高准确指的是文本描述要真实反映图片内容多样指的是数据要覆盖不同的场景和主题。这个过程通常能帮你解决几个核心问题去掉那些根本没法用的垃圾数据比如损坏的图片文件发现并处理重复的数据避免模型“偏科”检查图文是否匹配剔除那些“挂羊头卖狗肉”的样本把数据整理成模型能直接“吃”的标准格式做好了这些你的微调成功率至少能提升30%。这不是我瞎说的是很多实际项目验证过的经验。2. 搭建你的数据处理环境工欲善其事必先利其器。我们先来把需要的工具准备好。2.1 基础环境配置我建议直接用Python 3.8或更高版本这个版本比较稳定各种库的支持也最好。如果你还没有安装Python去官网下载一个安装包一路下一步就行。安装好Python后打开你的终端Windows用户叫命令提示符或者PowerShell我们来安装几个核心的库。# 数据处理和分析的瑞士军刀 pip install pandas numpy # 处理图片必备 pip install Pillow opencv-python # 文本处理相关 pip install jieba # 中文分词如果你的数据是英文可以不用 pip install langdetect # 检测文本语言 # 进度显示处理大数据时很实用 pip install tqdm # 如果需要处理特殊格式的数据 pip install pyarrow # 处理Parquet格式这些库基本上覆盖了数据清洗的各个方面。Pandas用来整理和筛选数据OpenCV和Pillow用来检查和处理图片其他几个则在特定场景下很有用。2.2 准备你的数据在开始清洗之前你得先有个地方放数据。我通常这样组织文件夹your_data_project/ ├── raw_data/ # 原始数据不动它 │ ├── images/ # 原始图片 │ └── metadata.csv # 原始标注文件 ├── processed_data/ # 清洗后的数据 ├── scripts/ # 处理脚本 │ └── data_clean.py # 今天要写的清洗脚本 └── logs/ # 处理日志原始数据放在raw_data里我们不对它做任何修改所有操作都在内存或生成新文件。这样万一处理过程中出了什么问题我们还能回到最原始的状态。你的标注文件可能是CSV、JSON、或者TXT格式这都没关系。关键是要知道它的结构至少应该包含图片路径和对应的文本描述。3. 第一步基础清洗——去掉明显的问题数据现在让我们开始写第一个清洗函数。我们从最简单但最有效的基础清洗开始。3.1 加载和查看数据首先我们看看手头的数据长什么样。import pandas as pd import os from tqdm import tqdm import logging # 设置日志方便查看处理过程 logging.basicConfig(levellogging.INFO, format%(asctime)s - %(levelname)s - %(message)s) logger logging.getLogger(__name__) def load_and_inspect_data(metadata_path): 加载数据并查看基本情况 logger.info(f正在加载数据: {metadata_path}) # 根据文件类型选择加载方式 if metadata_path.endswith(.csv): df pd.read_csv(metadata_path) elif metadata_path.endswith(.json): df pd.read_json(metadata_path) elif metadata_path.endswith(.parquet): df pd.read_parquet(metadata_path) else: # 尝试其他格式或者报错 raise ValueError(f不支持的格式: {metadata_path}) logger.info(f数据加载完成共有 {len(df)} 条记录) logger.info(f数据列名: {df.columns.tolist()}) logger.info(f前几条数据:\n{df.head()}) # 查看缺失值情况 missing_info df.isnull().sum() logger.info(f缺失值统计:\n{missing_info[missing_info 0]}) return df # 使用示例 if __name__ __main__: # 假设你的标注文件是CSV格式 df load_and_inspect_data(raw_data/metadata.csv)运行这段代码你就能看到数据的基本情况有多少条记录、有哪些列、有没有缺失值。这是了解数据的第一步。3.2 处理缺失值和损坏文件接下来我们处理那些明显有问题的数据。import cv2 from PIL import Image import numpy as np def check_image_file(img_path): 检查图片文件是否有效 返回: (是否有效, 错误信息) if not os.path.exists(img_path): return False, 文件不存在 try: # 尝试用PIL打开 with Image.open(img_path) as img: img.verify() # 验证文件完整性 # 再尝试用OpenCV读取确保能正常加载 img_cv cv2.imread(img_path) if img_cv is None: return False, OpenCV无法读取 # 检查图片尺寸是否合理 height, width img_cv.shape[:2] if height 10 or width 10: return False, f图片尺寸过小: {width}x{height} return True, except Exception as e: return False, f图片损坏: {str(e)} def check_text_description(text): 检查文本描述是否有效 if not isinstance(text, str): return False, 不是字符串类型 # 去除空白字符后检查长度 cleaned_text text.strip() if len(cleaned_text) 0: return False, 文本为空 # 检查是否包含有效字符至少有一个非空格字符 if not any(c.isalnum() for c in cleaned_text): return False, 无有效字符 # 文本长度检查根据实际情况调整 if len(cleaned_text) 3: return False, 文本过短 return True, def basic_cleaning(df, image_dirraw_data/images): 基础清洗移除缺失值、损坏文件、无效文本 logger.info(开始基础清洗...) # 记录原始数量 original_count len(df) # 1. 移除缺失值 df_clean df.dropna(subset[image_path, caption]) # 假设你的列名是这些 # 2. 检查图片文件 valid_indices [] img_errors [] logger.info(检查图片文件...) for idx, row in tqdm(df_clean.iterrows(), totallen(df_clean)): img_path os.path.join(image_dir, row[image_path]) is_valid, error_msg check_image_file(img_path) if is_valid: valid_indices.append(idx) else: img_errors.append((row[image_path], error_msg)) df_clean df_clean.loc[valid_indices] # 3. 检查文本描述 valid_indices [] text_errors [] logger.info(检查文本描述...) for idx, row in tqdm(df_clean.iterrows(), totallen(df_clean)): text row[caption] is_valid, error_msg check_text_description(text) if is_valid: valid_indices.append(idx) else: text_errors.append((text[:50] ... if len(text) 50 else text, error_msg)) df_clean df_clean.loc[valid_indices] # 记录清洗结果 logger.info(f基础清洗完成:) logger.info(f 原始数据量: {original_count}) logger.info(f 清理后数据量: {len(df_clean)}) logger.info(f 移除无效图片: {len(img_errors)}) logger.info(f 移除无效文本: {len(text_errors)}) # 保存错误日志可选 if img_errors: with open(logs/image_errors.log, w) as f: for img_path, error in img_errors: f.write(f{img_path}: {error}\n) if text_errors: with open(logs/text_errors.log, w) as f: for text, error in text_errors: f.write(f{text}: {error}\n) return df_clean # 使用示例 df_clean basic_cleaning(df, image_dirraw_data/images)这个基础清洗能帮你过滤掉大约10-30%的明显问题数据。别小看这一步很多训练失败就是因为这些基础问题没处理好。4. 第二步去重——避免数据“偏科”数据重复是个隐形杀手。想象一下如果你的训练数据里同样的图片出现了20次模型就会过度关注这些样本导致泛化能力变差。4.1 基于图片内容的去重import hashlib from collections import defaultdict def calculate_image_hash(img_path, hash_size16): 计算图片的感知哈希 相似的图片会有相似的哈希值 try: # 读取图片并调整大小加快计算速度 img cv2.imread(img_path) if img is None: return None # 转换为灰度图 gray cv2.cvtColor(img, cv2.COLOR_BGR2GRAY) # 调整大小 resized cv2.resize(gray, (hash_size, hash_size)) # 计算平均值 avg resized.mean() # 生成哈希大于平均值为1否则为0 hash_str .join([1 if pixel avg else 0 for pixel in resized.flatten()]) return hash_str except Exception as e: logger.warning(f计算图片哈希失败: {img_path}, 错误: {str(e)}) return None def find_similar_images(df, image_dir, hash_size16, similarity_threshold0.9): 查找相似的图片 logger.info(开始查找相似图片...) # 计算所有图片的哈希 image_hashes {} for idx, row in tqdm(df.iterrows(), totallen(df)): img_path os.path.join(image_dir, row[image_path]) img_hash calculate_image_hash(img_path, hash_size) if img_hash: image_hashes[idx] img_hash # 按哈希值分组 hash_groups defaultdict(list) for idx, img_hash in image_hashes.items(): hash_groups[img_hash].append(idx) # 查找相似图片汉明距离小于阈值 similar_groups [] # 将哈希值转换为整数以便比较 hash_list list(hash_groups.items()) for i in range(len(hash_list)): hash1, indices1 hash_list[i] if len(indices1) 1: # 相同哈希值肯定是重复的 similar_groups.append(indices1) continue for j in range(i 1, len(hash_list)): hash2, indices2 hash_list[j] # 计算汉明距离 hamming_distance sum(c1 ! c2 for c1, c2 in zip(hash1, hash2)) similarity 1 - (hamming_distance / len(hash1)) if similarity similarity_threshold: similar_groups.append(indices1 indices2) # 从每组相似图片中保留一个 indices_to_keep [] indices_to_remove [] for group in similar_groups: # 保留第一个移除其他的 indices_to_keep.append(group[0]) indices_to_remove.extend(group[1:]) logger.info(f找到 {len(similar_groups)} 组相似图片) logger.info(f将移除 {len(indices_to_remove)} 个重复图片) # 创建新的DataFrame只保留不重复的 keep_mask df.index.isin(indices_to_keep) | ~df.index.isin(indices_to_remove) df_deduped df[keep_mask].copy() logger.info(f去重后数据量: {len(df_deduped)}) return df_deduped, similar_groups # 使用示例 df_deduped, similar_groups find_similar_images( df_clean, image_dirraw_data/images, similarity_threshold0.95 # 相似度阈值可以根据需要调整 )4.2 基于文本的去重图片去重了文本也得去重。同样的描述反复出现也会影响模型学习。from sklearn.feature_extraction.text import TfidfVectorizer from sklearn.metrics.pairwise import cosine_similarity import re def preprocess_text(text): 文本预处理清洗、分词等 if not isinstance(text, str): return # 转换为小写 text text.lower() # 移除特殊字符和多余空格 text re.sub(r[^\w\s], , text) text re.sub(r\s, , text).strip() return text def find_similar_texts(df, text_columncaption, similarity_threshold0.8): 查找相似的文本描述 logger.info(开始查找相似文本...) # 预处理文本 texts df[text_column].apply(preprocess_text).tolist() # 使用TF-IDF计算文本相似度 vectorizer TfidfVectorizer(max_features5000, stop_wordsNone) tfidf_matrix vectorizer.fit_transform(texts) # 计算余弦相似度 cosine_sim cosine_similarity(tfidf_matrix) # 查找相似文本对 similar_pairs [] n len(texts) for i in range(n): for j in range(i 1, n): if cosine_sim[i, j] similarity_threshold: similar_pairs.append((i, j, cosine_sim[i, j])) # 分组相似的文本 groups [] visited set() for i, j, sim in similar_pairs: if i in visited or j in visited: continue group [i, j] visited.update([i, j]) # 查找与这个组相似的其他文本 for k in range(n): if k in visited: continue # 检查k是否与组内任一文本相似 if any(cosine_sim[k, member] similarity_threshold for member in group): group.append(k) visited.add(k) if len(group) 1: groups.append(group) # 从每组中保留一个文本 indices_to_keep [] indices_to_remove [] for group in groups: # 保留第一个移除其他的 indices_to_keep.append(df.index[group[0]]) for idx in group[1:]: indices_to_remove.append(df.index[idx]) logger.info(f找到 {len(groups)} 组相似文本) logger.info(f将移除 {len(indices_to_remove)} 个重复文本) # 创建新的DataFrame keep_mask ~df.index.isin(indices_to_remove) df_text_deduped df[keep_mask].copy() logger.info(f文本去重后数据量: {len(df_text_deduped)}) return df_text_deduped, groups # 使用示例 df_final_deduped, text_groups find_similar_texts( df_deduped, text_columncaption, similarity_threshold0.85 )去重完成后你的数据量可能会减少10-50%这取决于原始数据的质量。别心疼这些被去掉的数据它们只会干扰模型学习。5. 第三步质量检查——确保图文匹配对于视觉语言模型来说图文匹配是最关键的质量指标。如果图片和描述对不上模型就会学到错误的知识。5.1 基于规则的质量检查def check_text_image_relevance(row, image_dir): 基于规则检查图文相关性 返回: (是否相关, 置信度, 问题描述) img_path os.path.join(image_dir, row[image_path]) text str(row[caption]).lower() # 检查1: 文本长度与图片复杂度是否匹配 try: img cv2.imread(img_path) if img is not None: height, width img.shape[:2] img_size height * width # 大图片通常需要更详细的描述 text_length len(text.split()) if img_size 1000000 and text_length 10: # 百万像素以上但描述很短 return False, 0.3, 图片大但描述过短 if img_size 50000 and text_length 50: # 小图片但描述很长 return False, 0.4, 图片小但描述过长 except: pass # 检查2: 常见的不匹配模式 mismatch_patterns [ (rphoto of.*man, [woman, female, girl]), (rphoto of.*woman, [man, male, boy]), (rred.*car, [blue, green, yellow]), (rindoor.*scene, [outdoor, street, park]), (rdaytime, [night, dark, evening]), ] for pattern, contradictions in mismatch_patterns: if re.search(pattern, text): # 检查图片文件名或路径是否包含矛盾词 img_path_lower img_path.lower() for contradiction in contradictions: if contradiction in img_path_lower: return False, 0.7, f图文矛盾: {pattern} vs {contradiction} # 检查3: 通用质量指标 issues [] # 文本过于通用 generic_phrases [a picture, an image, photo, image, picture of] if any(phrase in text for phrase in generic_phrases) and len(text.split()) 5: issues.append(描述过于通用) # 文本包含明显错误 if test in text and image in text: issues.append(可能是测试数据) if issues: return False, 0.5, ; .join(issues) return True, 0.8, 质量合格 def quality_filtering(df, image_dir, min_confidence0.6): 质量过滤移除低质量的图文对 logger.info(开始质量过滤...) quality_results [] for idx, row in tqdm(df.iterrows(), totallen(df)): is_relevant, confidence, issue check_text_image_relevance(row, image_dir) quality_results.append({ index: idx, is_relevant: is_relevant, confidence: confidence, issue: issue }) # 创建质量DataFrame quality_df pd.DataFrame(quality_results) # 过滤低质量数据 high_quality_mask (quality_df[is_relevant]) (quality_df[confidence] min_confidence) df_high_quality df.loc[quality_df[high_quality_mask][index]].copy() # 统计结果 low_quality_count len(df) - len(df_high_quality) logger.info(f质量过滤完成:) logger.info(f 过滤前数据量: {len(df)}) logger.info(f 过滤后数据量: {len(df_high_quality)}) logger.info(f 移除低质量数据: {low_quality_count}) # 分析主要问题 if low_quality_count 0: low_quality_issues quality_df[~high_quality_mask][issue].value_counts() logger.info(主要质量问题分布:) for issue, count in low_quality_issues.head(10).items(): logger.info(f {issue}: {count}条) return df_high_quality, quality_df # 使用示例 df_high_quality, quality_info quality_filtering( df_final_deduped, image_dirraw_data/images, min_confidence0.7 )5.2 使用现有模型进行质量评分进阶如果你有计算资源可以用一个现成的图文匹配模型来评分。# 注意这需要额外的模型和计算资源 # 这里只是一个示例框架 def score_with_clip_model(df, image_dir, batch_size32): 使用CLIP模型评估图文匹配度 这是一个高级功能需要安装transformers和torch try: from transformers import CLIPProcessor, CLIPModel import torch logger.info(加载CLIP模型进行质量评分...) # 加载模型这里以中文CLIP为例 model_name OFA-Sys/chinese-clip-vit-base-patch16 model CLIPModel.from_pretrained(model_name) processor CLIPProcessor.from_pretrained(model_name) device cuda if torch.cuda.is_available() else cpu model.to(device) model.eval() scores [] # 分批处理 for i in tqdm(range(0, len(df), batch_size)): batch_df df.iloc[i:ibatch_size] batch_images [] batch_texts [] valid_indices [] # 准备批次数据 for idx, row in batch_df.iterrows(): img_path os.path.join(image_dir, row[image_path]) try: image Image.open(img_path).convert(RGB) batch_images.append(image) batch_texts.append(row[caption]) valid_indices.append(idx) except: continue if not batch_images: continue # 处理输入 inputs processor( textbatch_texts, imagesbatch_images, return_tensorspt, paddingTrue ).to(device) # 计算相似度 with torch.no_grad(): outputs model(**inputs) logits_per_image outputs.logits_per_image # 图像-文本相似度 batch_scores logits_per_image.cpu().numpy().diagonal() # 收集分数 for idx, score in zip(valid_indices, batch_scores): scores.append((idx, float(score))) # 创建分数DataFrame scores_df pd.DataFrame(scores, columns[index, clip_score]) scores_df.set_index(index, inplaceTrue) # 合并到原始数据 df_with_scores df.copy() df_with_scores[clip_score] scores_df[clip_score] logger.info(fCLIP评分完成平均分: {df_with_scores[clip_score].mean():.3f}) return df_with_scores except ImportError: logger.warning(未安装transformers或torch跳过CLIP评分) return df except Exception as e: logger.error(fCLIP评分失败: {str(e)}) return df # 使用示例可选 # df_with_scores score_with_clip_model(df_high_quality, image_dirraw_data/images)6. 第四步格式标准化——让数据准备好被“投喂”清洗好的数据需要转换成模型能接受的格式。Ostrakon-VL-8B通常需要特定的数据格式。6.1 转换为标准格式def convert_to_standard_format(df, output_dirprocessed_data): 将清洗后的数据转换为标准格式 logger.info(开始格式转换...) # 创建输出目录 os.makedirs(output_dir, exist_okTrue) os.makedirs(os.path.join(output_dir, images), exist_okTrue) # 准备标准格式的数据 standard_data [] for idx, row in tqdm(df.iterrows(), totallen(df)): try: # 原始图片路径 src_img_path os.path.join(raw_data/images, row[image_path]) # 新图片路径可以重命名以避免冲突 new_img_name fimage_{idx:08d}.jpg dst_img_path os.path.join(output_dir, images, new_img_name) # 复制图片如果需要可以在这里进行图片处理 import shutil shutil.copy2(src_img_path, dst_img_path) # 构建标准格式的条目 item { image: new_img_name, # 相对路径 caption: str(row[caption]).strip(), original_source: row.get(source, unknown) if source in row else unknown, quality_score: row.get(clip_score, 1.0) if clip_score in row else 1.0 } # 添加其他可能有用的字段 for col in df.columns: if col not in [image_path, caption, clip_score]: item[col] row[col] standard_data.append(item) except Exception as e: logger.warning(f处理条目 {idx} 失败: {str(e)}) continue # 保存为JSONL格式每行一个JSON对象 jsonl_path os.path.join(output_dir, dataset.jsonl) with open(jsonl_path, w, encodingutf-8) as f: for item in standard_data: f.write(json.dumps(item, ensure_asciiFalse) \n) # 同时保存为Parquet格式更高效 parquet_path os.path.join(output_dir, dataset.parquet) pd.DataFrame(standard_data).to_parquet(parquet_path) # 保存统计信息 stats { total_samples: len(standard_data), image_dir: images/, metadata_files: [dataset.jsonl, dataset.parquet], created_at: pd.Timestamp.now().isoformat(), columns: list(standard_data[0].keys()) if standard_data else [] } with open(os.path.join(output_dir, dataset_info.json), w) as f: json.dumps(stats, indent2, ensure_asciiFalse) logger.info(f格式转换完成:) logger.info(f 总样本数: {len(standard_data)}) logger.info(f 图片保存到: {os.path.join(output_dir, images)}) logger.info(f 元数据保存到: {jsonl_path}) logger.info(f 统计信息: {os.path.join(output_dir, dataset_info.json)}) return standard_data # 使用示例 standard_data convert_to_standard_format(df_high_quality, output_dirprocessed_data)6.2 创建数据集的README为了让后续使用更清晰最好创建一个说明文件。def create_dataset_readme(output_dirprocessed_data): 创建数据集的README文件 readme_content f# 清洗后的Ostrakon-VL-8B训练数据集 ## 数据集信息 - 创建时间: {pd.Timestamp.now().strftime(%Y-%m-%d %H:%M:%S)} - 总样本数: {len(standard_data) if standard_data in locals() else 请运行脚本获取} - 图片格式: JPEG - 文本编码: UTF-8 ## 文件结构{output_dir}/ ├── images/ # 所有图片文件 │ ├── image_00000001.jpg │ ├── image_00000002.jpg │ └── ... ├── dataset.jsonl # JSONL格式的元数据 ├── dataset.parquet # Parquet格式的元数据 └── dataset_info.json # 数据集统计信息## 数据格式 每个样本包含以下字段 ### JSONL格式dataset.jsonl 每行是一个JSON对象 json {{ image: image_00000001.jpg, # 图片文件名 caption: 一张猫在沙发上的照片, # 文本描述 original_source: coco, # 数据来源 quality_score: 0.95 # 质量评分如果有 }}Parquet格式dataset.parquet相同的字段但以列式存储读取更快。使用示例Python读取示例import json import pandas as pd from PIL import Image # 读取JSONL格式 data [] with open(dataset.jsonl, r, encodingutf-8) as f: for line in f: data.append(json.loads(line)) # 读取Parquet格式推荐更快 df pd.read_parquet(dataset.parquet) # 加载图片 for idx, row in df.iterrows(): img_path fimages/{row[image]} image Image.open(img_path) caption row[caption] # ... 处理数据清洗流程基础清洗移除损坏文件和无效文本去重处理基于图片哈希和文本相似度质量过滤检查图文相关性格式标准化转换为模型可接受的格式注意事项所有图片已统一为JPEG格式文本描述已进行基础清洗去空格、小写化等建议在训练前再进行一次数据增强可根据需要调整训练/验证/测试集划分生成脚本: data_clean.py 生成环境: Python 3.8 readme_path os.path.join(output_dir, README.md) with open(readme_path, w, encodingutf-8) as f: f.write(readme_content) logger.info(fREADME文件已创建: {readme_path}) return readme_path使用示例readme_path create_dataset_readme(processed_data)## 7. 完整的数据清洗流程 现在我们把所有步骤组合起来形成一个完整的清洗流程。 python def complete_data_cleaning_pipeline( metadata_path, image_dirraw_data/images, output_dirprocessed_data, similarity_threshold0.9, min_confidence0.6 ): 完整的数据清洗流程 logger.info( * 50) logger.info(开始完整的数据清洗流程) logger.info( * 50) # 步骤1: 加载数据 logger.info(\n步骤1: 加载和检查数据) df load_and_inspect_data(metadata_path) # 步骤2: 基础清洗 logger.info(\n步骤2: 基础清洗) df_clean basic_cleaning(df, image_dir) # 步骤3: 图片去重 logger.info(\n步骤3: 图片去重) df_deduped_img, img_groups find_similar_images( df_clean, image_dir, similarity_thresholdsimilarity_threshold ) # 步骤4: 文本去重 logger.info(\n步骤4: 文本去重) df_deduped_text, text_groups find_similar_texts( df_deduped_img, similarity_threshold0.85 ) # 步骤5: 质量过滤 logger.info(\n步骤5: 质量过滤) df_high_quality, quality_info quality_filtering( df_deduped_text, image_dir, min_confidencemin_confidence ) # 步骤6: 格式转换 logger.info(\n步骤6: 格式转换) standard_data convert_to_standard_format(df_high_quality, output_dir) # 步骤7: 创建文档 logger.info(\n步骤7: 创建文档) create_dataset_readme(output_dir) # 最终统计 logger.info(\n * 50) logger.info(数据清洗流程完成!) logger.info( * 50) stats { original_samples: len(df), after_basic_cleaning: len(df_clean), after_image_dedup: len(df_deduped_img), after_text_dedup: len(df_deduped_text), after_quality_filter: len(df_high_quality), final_samples: len(standard_data), removal_rate: f{((len(df) - len(standard_data)) / len(df) * 100):.1f}% } logger.info(清洗统计:) for stage, count in stats.items(): logger.info(f {stage}: {count}) # 保存统计信息 stats_path os.path.join(output_dir, cleaning_stats.json) with open(stats_path, w) as f: json.dump(stats, f, indent2) logger.info(f\n清洗后的数据保存在: {output_dir}) logger.info(f统计信息: {stats_path}) return standard_data, stats # 使用示例 if __name__ __main__: # 运行完整流程 cleaned_data, statistics complete_data_cleaning_pipeline( metadata_pathraw_data/metadata.csv, image_dirraw_data/images, output_dirprocessed_data, similarity_threshold0.95, min_confidence0.7 ) print(f\n清洗完成! 最终保留 {statistics[final_samples]} 个样本) print(f数据清洗率: {statistics[removal_rate]})8. 实际应用中的注意事项走完整个流程你应该能得到一个干净、高质量的数据集。但在实际项目中还有一些细节需要注意。首先是参数调整。我在代码里用了一些默认参数比如图片相似度阈值设为0.95文本相似度阈值设为0.85。这些值不是绝对的你需要根据自己的数据特点来调整。如果你的数据来源单一重复率可能很高可以把阈值调严格一些如果数据本身就很稀缺可以适当放宽标准。然后是计算资源。处理大规模数据时内存和计算时间都是问题。如果数据量特别大比如上百万你可能需要分批处理或者使用更高效的数据结构。我建议先从一小部分数据开始把整个流程跑通确认效果后再扩展到全量数据。还有一个常见问题是类别平衡。如果你的数据里某些类别特别多某些类别特别少模型可能会偏向多数类。你可以在清洗完成后检查一下数据分布必要时进行采样平衡。最后是迭代优化。数据清洗很少能一次到位。你可能需要多次运行根据中间结果调整策略。比如如果发现某种类型的错误特别多可以专门写规则来处理。清洗日志就是代码里生成的logs/目录下的文件是你的好朋友多看看里面记录了什么问题。9. 总结数据清洗确实是个体力活没什么炫酷的技术但它的重要性怎么强调都不为过。一个干净的数据集能让后续的模型训练事半功倍。这套流程我们从基础清洗开始去掉明显的问题数据然后做去重避免数据重复影响模型接着检查图文匹配质量确保数据“名副其实”最后转换成标准格式方便直接用于训练。每一步都有对应的代码你可以直接拿来用也可以根据自己的需求修改。实际用下来这套方法能处理大部分常见的数据质量问题。当然不同的数据集可能有特殊的情况这时候就需要你灵活调整了。比如如果你的数据有很多专业领域的术语文本清洗的部分就要加强如果图片风格很统一去重的策略也要相应调整。最后给个建议数据清洗不要追求一步到位。先跑一遍基础流程看看效果再针对发现的问题进行优化。清洗过程中多保存中间结果和日志这样出了问题也好排查。有了干净的数据你的Ostrakon-VL-8B微调之路就成功了一半。获取更多AI镜像想探索更多AI镜像和应用场景访问 CSDN星图镜像广场提供丰富的预置镜像覆盖大模型推理、图像生成、视频生成、模型微调等多个领域支持一键部署。