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

前言

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

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

数据收集

数据源选择

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)

总结

数据集构建需要完整的流程：

1. 数据收集：选择合适的数据源
2. 数据清洗：去除噪声和低质量数据
3. 数据标注：添加标签和注释
4. 质量检查：确保数据质量
5. 格式转换：转换为标准格式

关键要点：

• 数据质量是模型性能的关键
• 需要多个标注员保证一致性
• 定期检查数据质量
• 使用标准格式便于后续处理