AI辅助数据标注与Synthetic Data完全教程

高质量数据是AI模型的命脉。本教程系统讲解如何利用AI辅助完成数据标注、生成合成数据，以及在模型微调中实际应用这些技术，帮助团队以更低的成本获得更高质量的训练数据。

目录

1. AI数据标注技术演进
2. Label Studio + LLM自动标注
3. GPT-4/Claude批量标注实战
4. Synthetic Data生成技术
5. 数据质量评估与过滤
6. 合成数据在模型微调中的应用
7. 数据增强技术
8. 多模态数据合成
9. 数据标注成本优化策略
10. 合成数据法律与伦理考量

1. AI数据标注技术演进

1.1 三代标注范式

数据标注经历了从纯人工到AI驱动的三个阶段：

1.2 核心挑战

即使在AI辅助时代，数据标注仍面临几个核心挑战：

• 一致性：不同标注员（或不同批次AI标注）对同一样本的理解可能不同
• 边界模糊：很多真实场景的分类边界是模糊的（如"中性"与"略微正面"）
• 长尾分布：稀有类别的样本少，AI标注准确率低
• 领域专业性：医疗、法律等专业领域需要专家知识

1.3 AI辅助标注的整体架构

原始数据 → 预处理 → LLM自动标注 → 置信度过滤 → 人工复核 → 高质量数据集
                                    ↓
                              低置信度样本 → 专家标注 → 标注指南更新

2. Label Studio + LLM自动标注

2.1 Label Studio 简介

Label Studio 是最流行的开源数据标注工具，支持文本、图像、音频、视频等多种数据类型。其强大的后端API和插件系统使其成为AI辅助标注的理想平台。

2.2 环境搭建

# 安装 Label Studio
pip install label-studio

# 启动服务
label-studio start --port 8080

# 安装 ML 后端 SDK
pip install label-studio-ml

2.3 自定义 LLM 后端

创建一个连接 OpenAI API 的 ML 后端，实现自动预标注：

from label_studio_ml.model import LabelStudioMLBase
from openai import OpenAI
import json

client = OpenAI()

class LLMAnnotator(LabelStudioMLBase):
    """使用LLM进行自动标注的Label Studio后端"""

    def __init__(self, **kwargs):
        super().__init__(**kwargs)
        self.parsed_label_config = self.parsed_label_config
        # 从Label Studio配置中提取标签
        self.labels = self._get_labels()

    def _get_labels(self):
        """从标注配置中提取所有标签"""
        labels = []
        for tag_name, tag_info in self.parsed_label_config.items():
            if 'labels' in tag_info:
                labels.extend(tag_info['labels'])
        return labels

    def predict(self, tasks, **kwargs):
        """对输入任务进行预测"""
        predictions = []

        for task in tasks:
            text = task['data']['text']

            # 调用LLM进行标注
            result = self._annotate_with_llm(text)

            predictions.append({
                'result': result,
                'score': result[0].get('score', 0.8) if result else 0,
                'model_version': 'gpt-4o-mini'
            })

        return predictions

    def _annotate_with_llm(self, text):
        """调用LLM进行文本分类标注"""
        prompt = f"""你是一个专业的数据标注专家。请对以下文本进行情感分类。

可选标签：{', '.join(self.labels)}

请严格按照以下JSON格式返回结果：
{{"label": "选择的标签", "confidence": 0.0到1.0之间的置信度, "reasoning": "简短的判断理由"}}

待标注文本：
{text}

请只返回JSON，不要其他内容。"""

        try:
            response = client.chat.completions.create(
                model="gpt-4o-mini",
                messages=[{"role": "user", "content": prompt}],
                temperature=0,
                response_format={"type": "json_object"}
            )

            data = json.loads(response.choices[0].message.content)

            return [{
                'from_name': 'label',
                'to_name': 'text',
                'type': 'choices',
                'value': {
                    'start': 0,
                    'end': len(text),
                    'text': text,
                    'choices': [data['label']]
                },
                'score': data.get('confidence', 0.8)
            }]

        except Exception as e:
            print(f"LLM标注失败: {e}")
            return []

    def fit(self, completions, workdir=None, **kwargs):
        """使用人工修正的结果进行微调（可选）"""
        # 这里可以收集人工修正的数据，用于后续优化prompt或微调模型
        return {}

2.4 启动 ML 后端

# 创建后端目录
mkdir llm-backend && cd llm-backend

# 创建启动脚本 _wsgi.py
cat > _wsgi.py << 'EOF'
from label_studio_ml.api import init_app
from model import LLMAnnotator

app = init_app(model_class=LLMAnnotator)

if __name__ == '__main__':
    app.run(host='0.0.0.0', port=9090)
EOF

# 启动
python _wsgi.py

2.5 Label Studio 配置示例

<!-- 情感分析标注模板 -->
<View>
  <Header value="请对以下文本进行情感分类" />
  <Text name="text" value="$text" />

  <Choices name="label" toName="text" choice="single" showInline="true">
    <Choice value="正面" />
    <Choice value="负面" />
    <Choice value="中性" />
  </Choices>

  <TextArea name="reasoning"
            toName="text"
            placeholder="标注理由（可选）"
            rows="2" />
</View>

3. GPT-4/Claude批量标注实战

3.1 批量标注框架设计

import asyncio
import json
import csv
from dataclasses import dataclass, field
from typing import Optional
from openai import AsyncOpenAI
from anthropic import AsyncAnthropic
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

@dataclass
class AnnotationTask:
    id: str
    text: str
    metadata: dict = field(default_factory=dict)

@dataclass
class AnnotationResult:
    task_id: str
    label: str
    confidence: float
    reasoning: str
    model: str
    raw_response: str = ""

class BatchAnnotator:
    """批量标注引擎，支持多模型、并发控制、断点续标"""

    def __init__(
        self,
        model: str = "gpt-4o-mini",
        concurrency: int = 10,
        retry_limit: int = 3,
        checkpoint_file: str = "checkpoint.json"
    ):
        self.model = model
        self.concurrency = concurrency
        self.retry_limit = retry_limit
        self.checkpoint_file = checkpoint_file
        self.semaphore = asyncio.Semaphore(concurrency)

        # 根据模型选择客户端
        if "gpt" in model or "o1" in model:
            self.client = AsyncOpenAI()
            self.provider = "openai"
        elif "claude" in model:
            self.client = AsyncAnthropic()
            self.provider = "anthropic"
        else:
            raise ValueError(f"不支持的模型: {model}")

        # 加载检查点（断点续标）
        self.completed: set[str] = set()
        self._load_checkpoint()

    def _load_checkpoint(self):
        try:
            with open(self.checkpoint_file) as f:
                data = json.load(f)
                self.completed = set(data.get("completed", []))
            logger.info(f"已加载检查点，已完成 {len(self.completed)} 条")
        except FileNotFoundError:
            pass

    def _save_checkpoint(self, task_id: str):
        self.completed.add(task_id)
        with open(self.checkpoint_file, 'w') as f:
            json.dump({"completed": list(self.completed)}, f)

    async def annotate_batch(
        self,
        tasks: list[AnnotationTask],
        prompt_template: str,
        labels: list[str]
    ) -> list[AnnotationResult]:
        """批量标注"""
        results = []
        pending = [t for t in tasks if t.id not in self.completed]
        logger.info(f"总任务: {len(tasks)}, 待处理: {len(pending)}")

        async def process_task(task: AnnotationTask) -> Optional[AnnotationResult]:
            async with self.semaphore:
                for attempt in range(self.retry_limit):
                    try:
                        result = await self._annotate_single(task, prompt_template, labels)
                        self._save_checkpoint(task.id)
                        return result
                    except Exception as e:
                        logger.warning(f"任务 {task.id} 第 {attempt+1} 次失败: {e}")
                        if attempt < self.retry_limit - 1:
                            await asyncio.sleep(2 ** attempt)
                return None

        # 并发执行
        coros = [process_task(task) for task in pending]
        batch_results = await asyncio.gather(*coros)

        results = [r for r in batch_results if r is not None]
        logger.info(f"标注完成: {len(results)}/{len(pending)}")
        return results

    async def _annotate_single(
        self,
        task: AnnotationTask,
        prompt_template: str,
        labels: list[str]
    ) -> AnnotationResult:
        """单条标注"""
        prompt = prompt_template.format(
            text=task.text,
            labels=", ".join(labels)
        )

        if self.provider == "openai":
            response = await self.client.chat.completions.create(
                model=self.model,
                messages=[{"role": "user", "content": prompt}],
                temperature=0,
                response_format={"type": "json_object"}
            )
            content = response.choices[0].message.content
        else:
            response = await self.client.messages.create(
                model=self.model,
                max_tokens=500,
                messages=[{"role": "user", "content": prompt}],
                temperature=0
            )
            content = response.content[0].text

        data = json.loads(content)
        return AnnotationResult(
            task_id=task.id,
            label=data.get("label", "unknown"),
            confidence=float(data.get("confidence", 0.5)),
            reasoning=data.get("reasoning", ""),
            model=self.model,
            raw_response=content
        )


# ===== 使用示例 =====

async def main():
    # 准备数据
    tasks = []
    with open("raw_data.csv", encoding="utf-8") as f:
        reader = csv.DictReader(f)
        for i, row in enumerate(reader):
            tasks.append(AnnotationTask(
                id=f"task_{i:05d}",
                text=row["text"],
                metadata={"source": row.get("source", "unknown")}
            ))

    # 定义标注prompt
    prompt = """你是一个专业的数据标注专家。请对以下文本进行情感分析。

可选标签：{labels}

请严格按以下JSON格式返回：
{{"label": "选择的标签", "confidence": 0-1的置信度, "reasoning": "判断理由"}}

文本：
{text}"""

    labels = ["正面", "负面", "中性"]

    # 创建标注器并执行
    annotator = BatchAnnotator(
        model="gpt-4o-mini",
        concurrency=20,
        checkpoint_file="sentiment_checkpoint.json"
    )

    results = await annotator.annotate_batch(tasks, prompt, labels)

    # 保存结果
    with open("annotations.jsonl", "w", encoding="utf-8") as f:
        for r in results:
            f.write(json.dumps({
                "task_id": r.task_id,
                "label": r.label,
                "confidence": r.confidence,
                "reasoning": r.reasoning,
                "model": r.model
            }, ensure_ascii=False) + "\n")

    # 统计
    from collections import Counter
    label_counts = Counter(r.label for r in results)
    avg_confidence = sum(r.confidence for r in results) / len(results)
    print(f"\n标注统计:")
    print(f"  总数: {len(results)}")
    print(f"  标签分布: {dict(label_counts)}")
    print(f"  平均置信度: {avg_confidence:.3f}")

asyncio.run(main())

3.2 复杂标注任务：NER命名实体识别

NER_PROMPT = """你是一个专业的命名实体识别(NER)标注专家。

请从以下文本中提取所有命名实体，并按类型分类。

实体类型：
- PERSON: 人名
- ORG: 组织/公司名
- LOC: 地点/地址
- DATE: 日期/时间
- MONEY: 金额
- PRODUCT: 产品名

请严格按以下JSON格式返回：
{{"entities": [{{"text": "实体文本", "type": "实体类型", "start": 起始位置, "end": 结束位置}}]}}

文本：
{text}"""

async def annotate_ner(annotator: BatchAnnotator, texts: list[str]):
    tasks = [AnnotationTask(id=f"ner_{i}", text=t) for i, t in enumerate(texts)]
    return await annotator.annotate_batch(
        tasks,
        NER_PROMPT,
        ["PERSON", "ORG", "LOC", "DATE", "MONEY", "PRODUCT"]
    )

3.3 多模型交叉验证

为提高标注质量，可以使用多个模型进行标注并取共识：

async def cross_validate(
    tasks: list[AnnotationTask],
    prompt: str,
    labels: list[str],
    models: list[str] = ["gpt-4o-mini", "claude-3-5-haiku-20241022"]
) -> list[dict]:
    """多模型交叉验证标注"""
    all_results = {}

    for model in models:
        annotator = BatchAnnotator(model=model, concurrency=10)
        results = await annotator.annotate_batch(tasks, prompt, labels)
        for r in results:
            if r.task_id not in all_results:
                all_results[r.task_id] = []
            all_results[r.task_id].append(r)

    # 投票决定最终标签
    final_results = []
    for task_id, results in all_results.items():
        labels_list = [r.label for r in results]
        from collections import Counter
        vote = Counter(labels_list)
        most_common_label, count = vote.most_common(1)[0]
        agreement = count / len(labels_list)

        final_results.append({
            "task_id": task_id,
            "final_label": most_common_label,
            "agreement": agreement,
            "model_labels": {r.model: r.label for r in results},
            "needs_review": agreement < 1.0  # 不一致的需要人工审核
        })

    return final_results

4. Synthetic Data生成技术

4.1 Evol-Instruct 技术

Evol-Instruct 是 WizardLM 提出的方法，通过逐步演化指令来增加复杂度：

import json
from openai import OpenAI

client = OpenAI()

def evol_instruct(seed_instruction: str, evolution_rounds: int = 3) -> list[dict]:
    """指令演化：从简单指令逐步生成复杂指令"""
    instructions = [seed_instruction]
    current = seed_instruction

    evolution_strategies = [
        ("增加约束", "请在以下指令基础上增加更多约束条件，使其更具体、更有挑战性：\n{instruction}"),
        ("深化推理", "请将以下指令改写为需要多步推理才能完成的复杂指令：\n{instruction}"),
        ("增加上下文", "请为以下指令添加具体的场景和上下文，使其更贴近真实应用场景：\n{instruction}"),
        ("多任务融合", "请将以下指令与一个相关但不同的任务融合，形成一个复合指令：\n{instruction}"),
    ]

    for round_num in range(evolution_rounds):
        strategy_name, strategy_prompt = evolution_strategies[round_num % len(evolution_strategies)]

        response = client.chat.completions.create(
            model="gpt-4o",
            messages=[{
                "role": "user",
                "content": strategy_prompt.format(instruction=current)
            }],
            temperature=0.7
        )

        current = response.choices[0].message.content.strip()
        instructions.append(current)
        print(f"Round {round_num + 1} ({strategy_name}): {current[:80]}...")

    return [{"round": i, "instruction": inst} for i, inst in enumerate(instructions)]


def generate_response(instruction: str) -> str:
    """为演化后的指令生成高质量回答"""
    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[{
            "role": "system",
            "content": "你是一个专业的AI助手，请详细、准确地回答用户的问题。回答应该结构清晰，包含具体的例子和解释。"
        }, {
            "role": "user",
            "content": instruction
        }],
        temperature=0.3
    )
    return response.choices[0].message.content


# 使用示例
seed = "请解释什么是机器学习"
evolved = evol_instruct(seed, evolution_rounds=4)

# 为每个演化后的指令生成回答
dataset = []
for item in evolved:
    answer = generate_response(item["instruction"])
    dataset.append({
        "instruction": item["instruction"],
        "response": answer,
        "evolution_round": item["round"]
    })

# 保存数据集
with open("evol_instruct_dataset.json", "w", encoding="utf-8") as f:
    json.dump(dataset, f, ensure_ascii=False, indent=2)

4.2 Self-Instruct 技术

Self-Instruct 让模型自己生成指令、输入和输出：

SEED_TASKS = [
    {"instruction": "将以下句子翻译成英文", "input": "今天天气很好。", "output": "The weather is nice today."},
    {"instruction": "总结以下文章的要点", "input": "（文章内容）", "output": "要点：..."},
    {"instruction": "解释以下概念", "input": "量子计算", "output": "量子计算是..."},
]

GENERATION_PROMPT = """你是一个指令数据生成器。基于以下已有的任务示例，生成5个新的、不同的任务。每个任务包含：
1. instruction: 清晰的任务描述
2. input: 任务的输入（可以为空字符串）
3. output: 期望的输出

已有任务示例：
{existing_tasks}

要求：
- 任务应该多样化，涵盖不同的能力（分类、生成、翻译、推理、总结等）
- 每个任务应该清晰明确，有唯一正确的答案
- 不要重复已有任务

请以JSON数组格式返回：
[{{"instruction": "...", "input": "...", "output": "..."}}]"""


def self_instruct(seed_tasks: list[dict], num_iterations: int = 10, batch_size: int = 5) -> list[dict]:
    """Self-Instruct: 自动生成指令数据集"""
    all_tasks = seed_tasks.copy()

    for iteration in range(num_iterations):
        # 随机选择已有任务作为示例
        import random
        examples = random.sample(all_tasks, min(4, len(all_tasks)))
        examples_text = "\n".join([
            f"- 指令: {t['instruction']}\n  输入: {t['input']}\n  输出: {t['output']}"
            for t in examples
        ])

        response = client.chat.completions.create(
            model="gpt-4o",
            messages=[{"role": "user", "content": GENERATION_PROMPT.format(
                existing_tasks=examples_text
            )}],
            temperature=0.8,
            response_format={"type": "json_object"}
        )

        new_tasks = json.loads(response.choices[0].message.content)
        if isinstance(new_tasks, dict) and "tasks" in new_tasks:
            new_tasks = new_tasks["tasks"]

        # 去重和质量过滤
        for task in new_tasks:
            if is_valid_task(task) and not is_duplicate(task, all_tasks):
                all_tasks.append(task)

        print(f"Iteration {iteration + 1}: 新增 {len(new_tasks)} 个任务，总计 {len(all_tasks)} 个")

    return all_tasks


def is_valid_task(task: dict) -> bool:
    """验证任务格式和质量"""
    if not all(k in task for k in ["instruction", "output"]):
        return False
    if len(task["instruction"]) < 10:
        return False
    if len(task["output"]) < 10:
        return False
    return True


def is_duplicate(task: dict, existing: list[dict]) -> bool:
    """简单的去重检查"""
    for existing_task in existing:
        if task["instruction"].lower().strip() == existing_task["instruction"].lower().strip():
            return True
    return False

4.3 基于种子数据的扩增

def augment_from_seeds(
    seeds: list[dict],
    target_count: int = 1000,
    diversity_prompt: str = None
) -> list[dict]:
    """从种子数据扩增生成更多训练数据"""
    dataset = []

    diversity_prompt = diversity_prompt or """基于以下种子数据，生成一个类似但不同的训练样本。
保持相同的数据格式和标签体系，但内容应该不同。

种子数据：
{seed}

请生成一个新的样本，JSON格式：
{{"instruction": "...", "input": "...", "output": "...", "category": "..."}}"""

    while len(dataset) < target_count:
        # 随机选择种子
        seed = random.choice(seeds)

        response = client.chat.completions.create(
            model="gpt-4o-mini",  # 用更便宜的模型批量生成
            messages=[{"role": "user", "content": diversity_prompt.format(
                seed=json.dumps(seed, ensure_ascii=False)
            )}],
            temperature=0.9,  # 高温度增加多样性
            response_format={"type": "json_object"}
        )

        try:
            new_sample = json.loads(response.choices[0].message.content)
            if is_valid_task(new_sample):
                dataset.append(new_sample)
        except json.JSONDecodeError:
            continue

        if len(dataset) % 100 == 0:
            print(f"已生成 {len(dataset)}/{target_count} 条")

    return dataset

5. 数据质量评估与过滤

5.1 多维度质量评估

from dataclasses import dataclass
import re

@dataclass
class QualityScore:
    relevance: float      # 相关性
    completeness: float   # 完整性
    consistency: float    # 一致性
    complexity: float     # 复杂度
    overall: float        # 综合分

class DataQualityEvaluator:
    """数据质量评估器"""

    def evaluate(self, sample: dict) -> QualityScore:
        scores = {
            "relevance": self._check_relevance(sample),
            "completeness": self._check_completeness(sample),
            "consistency": self._check_consistency(sample),
            "complexity": self._check_complexity(sample),
        }
        scores["overall"] = sum(scores.values()) / len(scores)
        return QualityScore(**scores)

    def _check_relevance(self, sample: dict) -> float:
        """检查instruction和output的相关性"""
        instruction = sample.get("instruction", "")
        output = sample.get("output", "")

        # 简单的关键词重叠检查
        inst_words = set(instruction.lower().split())
        out_words = set(output.lower().split())
        if not inst_words:
            return 0.0
        overlap = len(inst_words & out_words) / len(inst_words)
        return min(overlap * 2, 1.0)  # 归一化到0-1

    def _check_completeness(self, sample: dict) -> float:
        """检查数据完整性"""
        score = 0.0
        if sample.get("instruction") and len(sample["instruction"]) > 10:
            score += 0.4
        if sample.get("output") and len(sample["output"]) > 20:
            score += 0.4
        if sample.get("input"):
            score += 0.2
        return score

    def _check_consistency(self, sample: dict) -> float:
        """检查格式一致性"""
        output = sample.get("output", "")
        # 检查是否有明显格式问题
        if not output:
            return 0.0
        # 检查是否以"抱歉"、"我不能"等开头（可能表示模型拒绝回答）
        refusal_patterns = ["抱歉", "我不能", "I cannot", "I'm sorry", "I can't"]
        for pattern in refusal_patterns:
            if output.startswith(pattern):
                return 0.2
        return 0.9

    def _check_complexity(self, sample: dict) -> float:
        """评估任务复杂度"""
        instruction = sample.get("instruction", "")
        output = sample.get("output", "")

        # 指令长度和输出长度的综合评估
        inst_len = len(instruction)
        out_len = len(output)

        # 太短或太长都扣分
        inst_score = 1.0 if 20 < inst_len < 500 else 0.5
        out_score = 1.0 if 50 < out_len < 2000 else 0.5

        return (inst_score + out_score) / 2


def filter_dataset(dataset: list[dict], min_score: float = 0.6) -> tuple[list[dict], list[dict]]:
    """过滤数据集，返回(通过, 未通过)"""
    evaluator = DataQualityEvaluator()
    passed, failed = [], []

    for sample in dataset:
        score = evaluator.evaluate(sample)
        sample["quality_score"] = score.overall
        if score.overall >= min_score:
            passed.append(sample)
        else:
            failed.append(sample)

    print(f"质量过滤结果: {len(passed)} 通过, {len(failed)} 未通过")
    return passed, failed

5.2 LLM辅助质量评估

async def llm_quality_check(sample: dict, model: str = "gpt-4o-mini") -> dict:
    """使用LLM评估单条数据的质量"""
    prompt = f"""请评估以下训练数据的质量，从1-5打分。

指令: {sample['instruction']}
输入: {sample.get('input', '(无)')}
输出: {sample['output']}

评估维度：
1. 准确性(output是否正确回答了instruction)
2. 完整性(output是否充分回答了问题)
3. 清晰度(instruction是否清晰明确)
4. 自然度(output语言是否自然流畅)

请返回JSON格式：
{{"accuracy": 1-5, "completeness": 1-5, "clarity": 1-5, "naturalness": 1-5, "overall": 1-5, "issues": ["问题描述"]}}"""

    client = AsyncOpenAI()
    response = await client.chat.completions.create(
        model=model,
        messages=[{"role": "user", "content": prompt}],
        temperature=0,
        response_format={"type": "json_object"}
    )
    return json.loads(response.choices[0].message.content)


async def batch_quality_check(dataset: list[dict], sample_rate: float = 0.1) -> dict:
    """对数据集进行抽样质量检查"""
    import random
    sample_size = max(int(len(dataset) * sample_rate), 10)
    samples = random.sample(dataset, min(sample_size, len(dataset)))

    results = await asyncio.gather(*[llm_quality_check(s) for s in samples])

    # 统计
    avg_scores = {}
    for dim in ["accuracy", "completeness", "clarity", "naturalness", "overall"]:
        scores = [r[dim] for r in results if dim in r]
        avg_scores[dim] = sum(scores) / len(scores) if scores else 0

    issues = []
    for r in results:
        issues.extend(r.get("issues", []))

    from collections import Counter
    common_issues = Counter(issues).most_common(5)

    return {
        "sample_size": sample_size,
        "average_scores": avg_scores,
        "common_issues": common_issues,
        "recommendation": "PASS" if avg_scores["overall"] >= 3.5 else "NEEDS_IMPROVEMENT"
    }

6. 合成数据在模型微调中的应用

6.1 数据格式准备

不同微调框架需要不同的数据格式：

def convert_to_alpaca_format(dataset: list[dict]) -> list[dict]:
    """转换为Alpaca格式（适用于LLaMA-Factory等）"""
    return [{
        "instruction": item["instruction"],
        "input": item.get("input", ""),
        "output": item["output"]
    } for item in dataset]


def convert_to_sharegpt_format(dataset: list[dict]) -> list[dict]:
    """转换为ShareGPT格式（适用于多数微调框架）"""
    return [{
        "conversations": [
            {"from": "human", "value": item["instruction"] + (f"\n{item['input']}" if item.get("input") else "")},
            {"from": "gpt", "value": item["output"]}
        ]
    } for item in dataset]


def convert_to_openai_format(dataset: list[dict]) -> list[dict]:
    """转换为OpenAI微调API格式"""
    return [{
        "messages": [
            {"role": "system", "content": "你是一个专业的AI助手。"},
            {"role": "user", "content": item["instruction"] + (f"\n{item['input']}" if item.get("input") else "")},
            {"role": "assistant", "content": item["output"]}
        ]
    } for item in dataset]


# 保存为JSONL
def save_jsonl(data: list[dict], filename: str):
    with open(filename, "w", encoding="utf-8") as f:
        for item in data:
            f.write(json.dumps(item, ensure_ascii=False) + "\n")
    print(f"已保存 {len(data)} 条到 {filename}")

6.2 数据配比策略

def create_balanced_dataset(
    real_data: list[dict],
    synthetic_data: list[dict],
    real_ratio: float = 0.3,
    max_total: int = 10000
) -> list[dict]:
    """创建真实数据与合成数据的平衡混合数据集"""

    # 计算各类数据的数量
    real_count = int(max_total * real_ratio)
    synth_count = max_total - real_count

    # 采样
    import random
    real_sample = random.sample(real_data, min(real_count, len(real_data)))
    synth_sample = random.sample(synthetic_data, min(synth_count, len(synthetic_data)))

    # 合并并打乱
    combined = real_sample + synth_sample
    random.shuffle(combined)

    # 添加数据来源标记
    for item in combined:
        if item not in synthetic_data:
            item["source"] = "real"
        else:
            item["source"] = "synthetic"

    print(f"混合数据集: {len(real_sample)} 真实 + {len(synth_sample)} 合成 = {len(combined)} 总计")
    return combined

6.3 微调效果验证

async def evaluate_model_quality(
    model_name: str,
    test_set: list[dict],
    baseline_model: str = "gpt-4o-mini"
) -> dict:
    """评估微调后模型的质量"""
    client = OpenAI()
    correct = 0
    total = len(test_set)

    for sample in test_set:
        # 用微调模型生成回答
        response = client.chat.completions.create(
            model=model_name,
            messages=[{"role": "user", "content": sample["instruction"]}],
            temperature=0
        )
        model_output = response.choices[0].message.content

        # 用GPT-4评估回答质量
        eval_prompt = f"""评估以下回答是否正确且完整。

问题: {sample['instruction']}
参考答案: {sample['output']}
模型回答: {model_output}

回答是否正确且完整？只回答 "YES" 或 "NO"。"""

        eval_response = client.chat.completions.create(
            model=baseline_model,
            messages=[{"role": "user", "content": eval_prompt}],
            temperature=0
        )

        if "YES" in eval_response.choices[0].message.content.upper():
            correct += 1

    accuracy = correct / total
    return {
        "model": model_name,
        "accuracy": accuracy,
        "correct": correct,
        "total": total,
        "verdict": "PASS" if accuracy >= 0.8 else "NEEDS_IMPROVEMENT"
    }

7. 数据增强技术

7.1 回译增强（Back Translation）

def back_translate(text: str, pivot_language: str = "en") -> list[str]:
    """回译增强：翻译成中间语言再翻译回来"""
    results = []

    # 中文 → 英文 → 中文
    en_text = translate(text, "zh", pivot_language)
    zh_back = translate(en_text, pivot_language, "zh")
    results.append(zh_back)

    # 中文 → 日文 → 中文（增加多样性）
    ja_text = translate(text, "zh", "ja")
    zh_back2 = translate(ja_text, "ja", "zh")
    results.append(zh_back2)

    return results


def translate(text: str, source_lang: str, target_lang: str) -> str:
    """使用LLM进行翻译"""
    lang_map = {"zh": "中文", "en": "英文", "ja": "日文", "ko": "韩文"}

    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{
            "role": "user",
            "content": f"请将以下{lang_map[source_lang]}文本翻译成{lang_map[target_lang]}，只返回翻译结果：\n{text}"
        }],
        temperature=0.3
    )
    return response.choices[0].message.content.strip()

7.2 改写增强（Paraphrasing）

def paraphrase(text: str, styles: list[str] = None) -> list[str]:
    """多风格改写增强"""
    if styles is None:
        styles = ["正式", "口语化", "简洁", "详细"]

    results = []
    for style in styles:
        prompt = f"""请用{style}的风格改写以下文本，保持原意不变，只返回改写后的文本：

{text}"""

        response = client.chat.completions.create(
            model="gpt-4o-mini",
            messages=[{"role": "user", "content": prompt}],
            temperature=0.7
        )
        results.append(response.choices[0].message.content.strip())

    return results


def expand_with_examples(text: str) -> str:
    """为文本添加具体示例"""
    prompt = f"""请为以下解释添加2-3个具体、贴近生活的例子，使其更易于理解：

{text}"""

    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": prompt}],
        temperature=0.5
    )
    return response.choices[0].message.content.strip()

7.3 难度级别变换

def adjust_difficulty(text: str, target_level: str) -> str:
    """调整文本难度级别"""
    level_prompts = {
        "beginner": "请将以下内容改写为适合初学者理解的版本，使用简单的词汇和短句",
        "intermediate": "请将以下内容改写为中等难度版本",
        "advanced": "请将以下内容改写为专业/高级版本，使用更精确的术语和复杂的句式"
    }

    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{
            "role": "user",
            "content": f"{level_prompts[target_level]}：\n\n{text}"
        }],
        temperature=0.3
    )
    return response.choices[0].message.content.strip()

8. 多模态数据合成

8.1 图文配对数据生成

def generate_image_caption_pairs(descriptions: list[str]) -> list[dict]:
    """为给定描述生成图片说明对"""
    pairs = []

    for desc in descriptions:
        # 生成不同风格的图片说明
        styles = [
            f"请为以下场景写一段简洁的图片说明（30字以内）：{desc}",
            f"请为以下场景写一段详细的图片描述（100字左右）：{desc}",
            f"请为以下场景写一段适合社交媒体的图片说明：{desc}",
        ]

        for style_prompt in styles:
            response = client.chat.completions.create(
                model="gpt-4o-mini",
                messages=[{"role": "user", "content": style_prompt}],
                temperature=0.7
            )
            pairs.append({
                "scene": desc,
                "caption": response.choices[0].message.content.strip()
            })

    return pairs

8.2 代码-注释对生成

def generate_code_documentation(code_snippets: list[str]) -> list[dict]:
    """为代码片段生成文档"""
    results = []

    for code in code_snippets:
        # 生成不同详细程度的文档
        docs = {}

        # 简短注释
        response = client.chat.completions.create(
            model="gpt-4o",
            messages=[{
                "role": "user",
                "content": f"请为以下代码写一行简洁的注释（20字以内）：\n{code}"
            }],
            temperature=0
        )
        docs["brief"] = response.choices[0].message.content.strip()

        # 详细文档
        response = client.chat.completions.create(
            model="gpt-4o",
            messages=[{
                "role": "user",
                "content": f"""请为以下代码写详细的文档，包括：
1. 功能说明
2. 参数说明
3. 返回值说明
4. 使用示例

代码：
{code}"""
            }],
            temperature=0
        )
        docs["detailed"] = response.choices[0].message.content.strip()

        results.append({"code": code, "documentation": docs})

    return results

9. 数据标注成本优化策略

9.1 成本对比分析

9.2 分层标注策略

class TieredAnnotationPipeline:
    """分层标注流水线：根据任务难度选择不同成本的标注方式"""

    def __init__(self):
        self.easy_model = "gpt-4o-mini"      # 简单任务
        self.medium_model = "gpt-4o-mini"     # 中等任务
        self.hard_model = "gpt-4o"            # 困难任务
        self.client = OpenAI()

    async def annotate(self, tasks: list[dict]) -> list[dict]:
        results = []

        for task in tasks:
            # 第一层：本地规则快速判断
            rule_result = self._rule_based_check(task)
            if rule_result:
                results.append({**task, "label": rule_result, "method": "rule", "cost": 0})
                continue

            # 第二层：小模型初筛
            confidence = await self._quick_classify(task)

            if confidence > 0.9:
                # 高置信度，直接使用
                label = await self._annotate_with_model(task, self.easy_model)
                results.append({**task, "label": label, "method": "easy_model", "cost": 0.001})
            elif confidence > 0.6:
                # 中等置信度，使用中等模型
                label = await self._annotate_with_model(task, self.medium_model)
                results.append({**task, "label": label, "method": "medium_model", "cost": 0.005})
            else:
                # 低置信度，使用最强模型
                label = await self._annotate_with_model(task, self.hard_model)
                results.append({**task, "label": label, "method": "hard_model", "cost": 0.05})

        return results

    def _rule_based_check(self, task: dict) -> str | None:
        """基于规则的快速判断"""
        text = task.get("text", "").lower()
        # 简单的关键词规则
        positive_keywords = ["好", "棒", "优秀", "喜欢", "推荐"]
        negative_keywords = ["差", "烂", "讨厌", "失望", "退货"]

        pos_count = sum(1 for kw in positive_keywords if kw in text)
        neg_count = sum(1 for kw in negative_keywords if kw in text)

        if pos_count > 0 and neg_count == 0:
            return "正面"
        if neg_count > 0 and pos_count == 0:
            return "负面"
        return None

    async def _quick_classify(self, task: dict) -> float:
        """快速分类并返回置信度"""
        response = await self.client.chat.completions.create(
            model=self.easy_model,
            messages=[{
                "role": "user",
                "content": f"对以下文本进行情感分类，返回JSON：{{\"label\": \"正面/负面/中性\", \"confidence\": 0-1}}\n\n{task['text']}"
            }],
            temperature=0,
            response_format={"type": "json_object"}
        )
        result = json.loads(response.choices[0].message.content)
        return result.get("confidence", 0.5)

    async def _annotate_with_model(self, task: dict, model: str) -> str:
        response = await self.client.chat.completions.create(
            model=model,
            messages=[{"role": "user", "content": f"情感分类：{task['text']}"}],
            temperature=0
        )
        return response.choices[0].message.content.strip()

9.3 预算控制

class BudgetController:
    """标注预算控制器"""

    def __init__(self, daily_budget: float = 100.0):  # 美元
        self.daily_budget = daily_budget
        self.spent_today = 0.0
        self.model_prices = {
            "gpt-4o": {"input": 2.5, "output": 10.0},       # per 1M tokens
            "gpt-4o-mini": {"input": 0.15, "output": 0.6},
            "claude-3-5-haiku": {"input": 0.8, "output": 4.0},
        }

    def estimate_cost(self, model: str, input_tokens: int, output_tokens: int) -> float:
        """估算单次调用成本"""
        prices = self.model_prices.get(model, {"input": 1.0, "output": 3.0})
        cost = (input_tokens * prices["input"] + output_tokens * prices["output"]) / 1_000_000
        return cost

    def can_afford(self, estimated_cost: float) -> bool:
        """检查预算是否充足"""
        return (self.spent_today + estimated_cost) <= self.daily_budget

    def record_spend(self, cost: float):
        """记录支出"""
        self.spent_today += cost
        remaining = self.daily_budget - self.spent_today
        if remaining < self.daily_budget * 0.1:
            print(f"⚠️ 预算警告: 今日已花费 ${self.spent_today:.2f}, 仅剩 ${remaining:.2f}")

    def choose_model(self, task_complexity: str) -> str:
        """根据任务复杂度和预算选择模型"""
        remaining = self.daily_budget - self.spent_today

        if task_complexity == "easy":
            return "gpt-4o-mini"
        elif task_complexity == "medium":
            return "gpt-4o-mini" if remaining < 20 else "gpt-4o"
        else:  # hard
            if remaining < 10:
                return "gpt-4o-mini"  # 预算紧张，降级模型
            elif remaining < 50:
                return "gpt-4o-mini"
            else:
                return "gpt-4o"

10. 合成数据法律与伦理考量

10.1 主要法律框架

10.2 合成数据的合规要点

隐私合规：

• 确保合成数据不包含真实个人信息（PII）
• 使用差分隐私技术增加保护
• 定期进行隐私审计（re-identification attack测试）

版权合规：

• 合成数据的"灵感来源"是否构成衍生作品
• 记录数据生成的完整pipeline
• 避免生成与版权作品高度相似的内容

质量合规：

• 合成数据需标注"synthetic"来源
• 建立质量标准和验收流程
• 保留生成过程的审计日志

10.3 PII检测与脱敏

import re

class PIIDetector:
    """个人信息检测器"""

    PATTERNS = {
        "phone": re.compile(r'1[3-9]\d{9}'),
        "id_card": re.compile(r'\d{17}[\dXx]'),
        "email": re.compile(r'[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}'),
        "bank_card": re.compile(r'\d{16,19}'),
        "name_prefix": re.compile(r'(?:姓名|名字|我叫|我是)\s*[\u4e00-\u9fa5]{2,4}'),
    }

    def detect(self, text: str) -> list[dict]:
        """检测文本中的PII"""
        findings = []
        for pii_type, pattern in self.PATTERNS.items():
            matches = pattern.finditer(text)
            for match in matches:
                findings.append({
                    "type": pii_type,
                    "value": match.group(),
                    "start": match.start(),
                    "end": match.end()
                })
        return findings

    def mask(self, text: str) -> str:
        """脱敏处理"""
        findings = self.detect(text)
        # 从后往前替换，避免位置偏移
        findings.sort(key=lambda x: x["start"], reverse=True)

        masked = text
        for finding in findings:
            if finding["type"] == "phone":
                replacement = finding["value"][:3] + "****" + finding["value"][-4:]
            elif finding["type"] == "id_card":
                replacement = finding["value"][:6] + "********" + finding["value"][-4:]
            elif finding["type"] == "email":
                parts = finding["value"].split("@")
                replacement = parts[0][:2] + "***@" + parts[1]
            else:
                replacement = "***"

            masked = masked[:finding["start"]] + replacement + masked[finding["end"]:]

        return masked


# 使用示例
detector = PIIDetector()
text = "我的电话是13812345678，邮箱是zhangsan@company.com"
print(detector.mask(text))
# 输出：我的电话是138****5678，邮箱是zh***@company.com

10.4 合成数据质量审计清单

def audit_synthetic_dataset(dataset: list[dict]) -> dict:
    """合成数据集质量审计"""
    report = {
        "total_samples": len(dataset),
        "checks": {}
    }

    # 1. PII检查
    detector = PIIDetector()
    pii_count = sum(1 for d in dataset if detector.detect(d.get("output", "")))
    report["checks"]["pii_leakage"] = {
        "status": "PASS" if pii_count == 0 else "FAIL",
        "affected_samples": pii_count
    }

    # 2. 多样性检查
    instructions = [d.get("instruction", "") for d in dataset]
    unique_ratio = len(set(instructions)) / len(instructions) if instructions else 0
    report["checks"]["diversity"] = {
        "status": "PASS" if unique_ratio > 0.8 else "WARNING",
        "unique_ratio": round(unique_ratio, 3)
    }

    # 3. 长度分布检查
    lengths = [len(d.get("output", "")) for d in dataset]
    avg_len = sum(lengths) / len(lengths) if lengths else 0
    report["checks"]["length_distribution"] = {
        "average_length": round(avg_len),
        "min_length": min(lengths) if lengths else 0,
        "max_length": max(lengths) if lengths else 0
    }

    # 4. 语言质量抽检
    sample_check = min(50, len(dataset))
    import random
    samples = random.sample(dataset, sample_check)
    # 简单的格式检查
    format_issues = sum(1 for s in samples if not s.get("output", "").strip())
    report["checks"]["format_quality"] = {
        "status": "PASS" if format_issues < sample_check * 0.05 else "WARNING",
        "empty_outputs": format_issues
    }

    # 总体评估
    all_pass = all(
        c.get("status") == "PASS"
        for c in report["checks"].values()
    )
    report["overall"] = "PASS" if all_pass else "NEEDS_REVIEW"

    return report

10.5 伦理最佳实践

1. 透明度：明确标注数据来源（真实/合成），不隐瞒AI参与
2. 公平性：检查合成数据是否引入或放大偏见
3. 可追溯：保留完整的生成日志和版本控制
4. 人工兜底：关键决策场景不完全依赖合成数据
5. 持续监测：模型上线后持续监控，发现合成数据相关问题及时修正

总结

AI辅助数据标注和合成数据生成正在重塑AI开发的数据准备环节。核心要点：

1. 工具选型：Label Studio + LLM后端是最灵活的方案
2. 批量标注：异步并发 + 断点续标 + 多模型验证是关键
3. 合成数据：Evol-Instruct和Self-Instruct是主流方法
4. 质量控制：自动化评估 + 人工抽检的组合最可靠
5. 成本优化：分层标注策略可以降低70%+的标注成本
6. 合规优先：PII检测、版权审查、审计日志缺一不可

合成数据不是银弹，但用对了，它是小团队做大模型的最有力武器。

延伸阅读：

• Label Studio 文档
• Self-Instruct 论文
• Evol-Instruct 论文
• 合成数据综述