当前位置：首页 > news >正文

大模型数据集构建方法：从数据收集到质量保证

news 2026/5/29 4:27:41

大模型数据集构建方法：从数据收集到质量保证

前言

高质量的数据集是训练优秀大模型的基础。数据集的质量直接影响模型的性能和泛化能力。构建一个好的数据集需要从数据收集、清洗、标注到质量保证的完整流程。

我在项目中参与过多个数据集的构建工作，对数据处理流程有深入理解。今天分享一些实用的数据集构建方法。

数据收集

数据源选择

class DataCollector: """数据收集器""" def __init__(self): self.sources = [] def add_source(self, name: str, url: str, format: str): """添加数据源""" self.sources.append({ "name": name, "url": url, "format": format }) def collect(self, output_dir: str): """收集数据""" for source in self.sources: print(f"收集 {source['name']}...") if source["format"] == "json": self._download_json(source["url"], output_dir) elif source["format"] == "csv": self._download_csv(source["url"], output_dir) def _download_json(self, url: str, output_dir: str): """下载 JSON 数据""" import requests response = requests.get(url) with open(f"{output_dir}/data.json", "w") as f: f.write(response.text)

数据清洗

class DataCleaner: """数据清洗器""" def __init__(self): self.filters = [] def add_filter(self, filter_func): """添加过滤器""" self.filters.append(filter_func) def clean(self, data: list) -> list: """清洗数据""" cleaned = [] for item in data: item = self._remove_empty(item) item = self._normalize_text(item) # 应用自定义过滤器 for filter_func in self.filters: if not filter_func(item): break else: cleaned.append(item) return cleaned def _remove_empty(self, item: dict) -> dict: """移除空字段""" return {k: v for k, v in item.items() if v} def _normalize_text(self, item: dict) -> dict: """标准化文本""" if "text" in item: item["text"] = item["text"].strip() return item

数据标注

标注流程

class DataAnnotator: """数据标注器""" def __init__(self): self.annotators = [] def add_annotator(self, name: str): """添加标注员""" self.annotators.append(name) def annotate(self, data: list, task_type: str) -> list: """标注数据""" annotated = [] for item in data: annotations = [] for annotator in self.annotators: annotation = self._get_annotation(item, annotator, task_type) annotations.append(annotation) # 多数投票 final_label = self._majority_vote(annotations) item["label"] = final_label annotated.append(item) return annotated def _majority_vote(self, annotations: list) -> str: """多数投票""" from collections import Counter counts = Counter(annotations) return counts.most_common(1)[0][0]

质量检查

class QualityChecker: """质量检查器""" def __init__(self): self.threshold = 0.8 def check_quality(self, data: list) -> tuple: """检查数据质量""" total = len(data) valid = 0 for item in data: if self._is_valid(item): valid += 1 quality = valid / total if quality < self.threshold: print(f"警告：数据质量低于阈值 ({quality:.2%})") return quality, valid, total def _is_valid(self, item: dict) -> bool: """检查单条数据""" # 检查必要字段 required_fields = ["text", "label"] for field in required_fields: if field not in item or not item[field]: return False # 检查文本长度 if len(item["text"]) < 10: return False return True

数据集格式

标准格式

class DatasetFormatter: """数据集格式化器""" def __init__(self, format_type: str = "jsonl"): self.format_type = format_type def format(self, data: list, output_path: str): """格式化数据集""" if self.format_type == "jsonl": self._to_jsonl(data, output_path) elif self.format_type == "hf": self._to_hf_format(data, output_path) def _to_jsonl(self, data: list, output_path: str): """转换为 JSONL 格式""" with open(output_path, "w") as f: for item in data: import json f.write(json.dumps(item, ensure_ascii=False) + "\n") def _to_hf_format(self, data: list, output_path: str): """转换为 HuggingFace 格式""" from datasets import Dataset dataset = Dataset.from_list(data) dataset.save_to_disk(output_path)