微调数据准备最佳实践

class DataCollector:
    def __init__(self):
        self.sources = {
            "human_annotated": [],    # 人工标注数据（质量最高）
            "model_generated": [],    # 模型生成+人工筛选
            "real_interactions": [],  # 真实用户交互（脱敏）
            "synthetic": [],          # 合成数据
        }
    
    async def collect(self):
        dataset = []
        
        # 1. 人工标注数据
        for item in self.sources["human_annotated"]:
            dataset.append({
                **item,
                "source": "human",
                "quality": "high"
            })
        
        # 2. 模型生成数据（需要筛选）
        for item in self.sources["model_generated"]:
            if await self.quality_check(item):
                dataset.append({
                    **item,
                    "source": "model_generated",
                    "quality": "medium"
                })
        
        # 3. 真实交互数据（脱敏处理）
        for item in self.sources["real_interactions"]:
            cleaned = self.desensitize(item)
            if cleaned:
                dataset.append({
                    **cleaned,
                    "source": "real",
                    "quality": "high"
                })
        
        return dataset

class DataFormatter:
    """统一数据格式为对话格式"""
    
    def format_instruction(self, instruction, input_text=None, output=None):
        return {
            "messages": [
                {"role": "system", "content": "你是一个专业助手。"},
                {"role": "user", "content": instruction + (f"\n\n{input_text}" if input_text else "")},
                {"role": "assistant", "content": output} if output else None,
            ],
            "metadata": {
                "task_type": "instruction",
                "language": "zh",
            }
        }
    
    def format_conversation(self, turns):
        """格式化多轮对话"""
        return {
            "messages": turns,
            "metadata": {"task_type": "conversation", "n_turns": len(turns) // 2}
        }
    
    def format_tool_use(self, user_message, tool_calls, tool_results, final_response):
        """格式化工具调用数据"""
        messages = [{"role": "user", "content": user_message}]
        for call, result in zip(tool_calls, tool_results):
            messages.append({"role": "assistant", "tool_calls": [call]})
            messages.append({"role": "tool", "content": json.dumps(result)})
        messages.append({"role": "assistant", "content": final_response})
        
        return {"messages": messages, "metadata": {"task_type": "tool_use"}}

class DataQualityChecker:
    def __init__(self):
        self.checks = [
            self.check_length,
            self.check_encoding,
            self.check_repetition,
            self.check_toxicity,
            self.check_consistency,
        ]
    
    async def check(self, sample):
        """运行所有质量检查"""
        for check in self.checks:
            result = await check(sample)
            if not result["passed"]:
                return False, result["reason"]
        return True, "All checks passed"
    
    async def check_length(self, sample):
        text = self.extract_text(sample)
        if len(text) < 10:
            return {"passed": False, "reason": "Too short"}
        if len(text) > 32000:
            return {"passed": False, "reason": "Too long"}
        return {"passed": True}
    
    async def check_repetition(self, sample):
        text = self.extract_text(sample)
        # 检查n-gram重复
        words = text.split()
        if len(words) > 10:
            bigrams = [' '.join(words[i:i+2]) for i in range(len(words)-1)]
            repeat_ratio = len(set(bigrams)) / len(bigrams)
            if repeat_ratio < 0.5:
                return {"passed": False, "reason": "High repetition"}
        return {"passed": True}
    
    async def check_toxicity(self, sample):
        text = self.extract_text(sample)
        toxic_words = ["暴力", "色情", "毒品"]  # 简化示例
        if any(word in text for word in toxic_words):
            return {"passed": False, "reason": "Toxic content"}
        return {"passed": True}

class DataDeduplicator:
    def __init__(self, similarity_threshold=0.9):
        self.threshold = similarity_threshold
        self.embeddings = []
        self.model = SentenceTransformer('BAAI/bge-small-zh-v1.5')
    
    def deduplicate(self, dataset):
        """基于语义相似度去重"""
        texts = [self.extract_text(d) for d in dataset]
        embeddings = self.model.encode(texts, normalize_embeddings=True)
        
        unique_indices = []
        for i in range(len(dataset)):
            is_duplicate = False
            for j in unique_indices:
                similarity = embeddings[i] @ embeddings[j]
                if similarity > self.threshold:
                    is_duplicate = True
                    break
            if not is_duplicate:
                unique_indices.append(i)
        
        return [dataset[i] for i in unique_indices]

class DataAugmentor:
    def __init__(self, llm):
        self.llm = llm
    
    async def augment(self, sample, n_variants=3):
        """生成数据的变体"""
        variants = [sample]
        
        # 1. 改写用户问题
        rewritten = await self.rewrite_query(sample)
        variants.append(rewritten)
        
        # 2. 添加噪声（错别字等）
        noisy = self.add_typo_noise(sample)
        variants.append(noisy)
        
        # 3. 改变语气/风格
        restyled = await self.restyle(sample)
        variants.append(restyled)
        
        return variants
    
    async def rewrite_query(self, sample):
        """改写用户查询"""
        original_query = sample["messages"][1]["content"]
        prompt = f"将以下问题改写为不同表述，保持语义不变：\n{original_query}"
        rewritten = await self.llm.generate(prompt)
        
        new_sample = copy.deepcopy(sample)
        new_sample["messages"][1]["content"] = rewritten
        new_sample["metadata"]["augmented"] = "rewritten"
        return new_sample

def split_dataset(dataset, train_ratio=0.9, val_ratio=0.05, test_ratio=0.05):
    """按任务类型分层划分"""
    from sklearn.model_selection import train_test_split
    
    # 按任务类型分组
    by_task = defaultdict(list)
    for item in dataset:
        by_task[item["metadata"]["task_type"]].append(item)
    
    train, val, test = [], [], []
    for task_type, items in by_task.items():
        n = len(items)
        n_train = int(n * train_ratio)
        n_val = int(n * val_ratio)
        
        # 随机打乱
        random.shuffle(items)
        train.extend(items[:n_train])
        val.extend(items[n_train:n_train+n_val])
        test.extend(items[n_train+n_val:])
    
    return train, val, test

class DatasetAnalyzer:
    def analyze(self, dataset):
        stats = {
            "total_samples": len(dataset),
            "task_distribution": Counter(d["metadata"]["task_type"] for d in dataset),
            "avg_turns": np.mean([len(d["messages"]) // 2 for d in dataset]),
            "avg_length": np.mean([len(self.extract_text(d)) for d in dataset]),
            "length_distribution": self.length_distribution(dataset),
            "language_distribution": Counter(d["metadata"].get("language", "unknown") for d in dataset),
        }
        return stats
    
    def report(self, stats):
        print(f"总样本数：{stats['total_samples']}")
        print(f"任务分布：{dict(stats['task_distribution'])}")
        print(f"平均轮次：{stats['avg_turns']:.1f}")
        print(f"平均长度：{stats['avg_length']:.0f}字符")