数据构建深度解析：LLM 训练数据的艺术与科学

1
┌─────────────────────────────────────────────────────────────────┐
2
│                    LLM 训练数据的三种类型                         │
3
├─────────────────────────────────────────────────────────────────┤
4
│                                                                  │
5
│  类型 1: 预训练数据 (Pre-training Data)                         │
6
│    用途: 学习语言模型的基础能力（语法、语义、世界知识）           │
7
│    规模: trillions of tokens（数万亿）                          │
8
│    来源: 网页、书籍、代码、论文、百科等                          │
9
│    构建: 爬取 → 去重 → 质量过滤 → 格式转换                      │
10
│                                                                  │
11
│  类型 2: SFT 数据 (Supervised Fine-Tuning Data)                 │
12
│    用途: 学习指令遵循、对话格式、特定任务能力                     │
13
│    规模: thousands to millions（数千到数百万）                  │
14
│    来源: 人类标注、LLM 生成、公开数据集                           │
15
│    构建: 指令设计 → 回答生成 → 质量筛选                         │
16
│                                                                  │
17
│  类型 3: 偏好数据 (Preference Data)                              │
18
│    用途: 学习人类偏好（RLHF/DPO/KTO 对齐）                       │
19
│    规模: thousands to hundreds of thousands（数千到数十万）       │
20
│    来源: 人类标注、LLM 评判                                     │
21
│    构建: 候选生成 → 偏好标注 → 质量过滤                         │
22
│                                                                  │
23
└─────────────────────────────────────────────────────────────────┘

1
class PretrainingDataSources:
2
    """
3
    预训练数据来源及其特点
4
    """
5

6
    SOURCES = {
7
        # 网页数据：最大、最多样、但噪声最多
8
        "web": {
9
            "examples": ["CommonCrawl", "WebText", "CCNet", "C4"],
10
            "size": "~trillions of tokens",
11
            "pros": ["多样性高", "覆盖广", "成本相对低"],
12
            "cons": ["噪声多", "隐私问题", "质量参差不齐"],
13
        },
14

15
        # 书籍：高质量、连贯、但覆盖有限
16
        "books": {
17
            "examples": ["BookCorpus", "Books3", "Gutenberg", "Libgen"],
18
            "size": "~hundreds of billions",
19
            "pros": ["质量高", "长程依赖", "结构化"],
20
            "cons": ["版权问题", "覆盖有限", "可能过时"],
21
        },
22

23
        # 代码：GPT-4/Codex 能力的关键
24
        "code": {
25
            "examples": ["GitHub", "The Stack", "Python data"],
26
            "size": "~hundreds of billions",
27
            "pros": ["结构化", "逻辑性强", "可执行验证"],
28
            "cons": ["语言有限", "注释质量差", "版权问题"],
29
        },
30

31
        # 学术论文：知识和推理能力
32
        "papers": {
33
            "examples": ["arXiv", "Semantic Scholar"],
34
            "size": "~hundreds of billions",
35
            "pros": ["知识密集", "推理严谨", "引用丰富"],
36
            "cons": ["格式复杂", "数学公式多", "需要处理"],
37
        },
38

39
        # 百科：结构化知识
40
        "encyclopedia": {
41
            "examples": ["Wikipedia", "Wikidata"],
42
            "size": "~billions",
43
            "pros": ["知识准确", "结构化", "引用丰富"],
44
            "cons": ["覆盖面有限", "可能有偏见"],
45
        },
46
    }

1
class CCNetPipeline:
2
    """
3
    CCNet: CommonCrawl Net
4

5
    处理 CommonCrawl 网页数据的完整流水线
6
    论文: Wenzek et al., 2020
7
    """
8

9
    def __init__(self):
10
        self.language = "en"
11
        self.min_quality_score = 0.7
12
        self.min_length = 100
13

14
    def process_common_crawl(self, raw_warc_files):
15
        """
16
        完整流水线：
17
        1. 下载 CommonCrawl WARC 文件
18
        2. 提取文本（Text Extraction）
19
        3. 语言识别（Language Identification）
20
        4. 质量过滤（Quality Filtering）
21
        5. 去重（Deduplication）
22
        """
23
        all_documents = []
24

25
        for warc_file in tqdm(raw_warc_files):
26
            # Step 1: 下载并解压 WARC 文件
27
            warc_records = self.parse_warc(warc_file)
28

29
            # Step 2: 文本提取
30
            for record in warc_records:
31
                text = self.extract_text(record)
32

33
                # Step 3: 语言识别
34
                if self.is_target_language(text, self.language):
35
                    all_documents.append(text)
36

37
        # Step 4: 质量过滤
38
        filtered = self.quality_filter(all_documents)
39

40
        # Step 5: 去重
41
        deduped = self.deduplicate(filtered)
42

43
        return deduped
44

45
    def extract_text(self, warc_record):
46
        """
47
        从 HTML 中提取文本
48

49
        使用 jusText 或类似的启发式提取器
50
        移除导航栏、页脚、广告等噪音内容
51
        """
52
        html = warc_record.content
53

54
        # 方法 1: jusText
55
        # 基于马尔可夫链的文本提取
56
        # 区分"正文内容"和"模板内容"
57

58
        # 方法 2: trafilatura
59
        # 更准确的提取，保留表格和列表
60

61
        # 方法 3: boilerplate removal
62
        # 基于规则的噪音移除
63

64
        text = trafilatura.extract(html)
65

66
        return text
67

68
    def is_target_language(self, text, target_lang):
69
        """
70
        语言识别
71

72
        使用 fastText 的 LID (Language Identification) 模型
73
        在 Wikipedia 上预训练，支持 176 种语言
74
        """
75
        # 加载预训练的 fastText 模型
76
        lid_model = fasttext.load_model('lid.176.bin')
77

78
        # 预测语言和置信度
79
        predictions = lid_model.predict(text, k=1)
80

81
        lang = predictions[0][0].replace('__label__', '')
82
        confidence = predictions[1][0]
83

84
        return lang == target_lang and confidence > 0.8
85

86
    def quality_filter(self, documents):
87
        """
88
        质量过滤
89

90
        使用多个质量指标组合：
91
        1. 语言模型困惑度（LM Perplexity）
92
        2. 单词/字符比率
93
        3. 特殊字符密度
94
        4. 停用词比率
95
        """
96
        scored_docs = []
97

98
        for doc in documents:
99
            score = 0.0
100

101
            # 指标 1: 句子完整度
102
            # 好的文档应该有完整的句子
103
            sentences = sent_tokenize(doc)
104
            complete_ratio = sum(1 for s in sentences if s.endswith(('.','!','?'))) / len(sentences)
105
            score += 0.3 * complete_ratio
106

107
            # 指标 2: 词/字符比率
108
            # 合理范围：4-10 chars/word
109
            words = doc.split()
110
            if len(words) > 0:
111
                avg_word_len = len(doc) / len(words)
112
                if 3 < avg_word_len < 15:
113
                    score += 0.2
114
                else:
115
                    score += 0.1 * (10 - abs(avg_word_len - 6)) / 10
116

117
            # 指标 3: 停用词比率
118
            stopwords = set(stopwords_list)
119
            stopword_ratio = sum(1 for w in words if w.lower() in stopwords) / len(words)
120
            if 0.3 < stopword_ratio < 0.7:
121
                score += 0.2
122

123
            # 指标 4: 重复度
124
            # 高重复度可能是模板/垃圾内容
125
            repeat_ratio = self.compute_repeat_ratio(doc)
126
            if repeat_ratio < 0.2:
127
                score += 0.3 * (1 - repeat_ratio)
128

129
            if score >= self.min_quality_score:
130
                scored_docs.append((score, doc))
131

132
        # 按分数排序
133
        scored_docs.sort(reverse=True)
134

135
        return [doc for _, doc in scored_docs]
136

137
    def compute_repeat_ratio(self, text, n=5):
138
        """
139
        计算 n-gram 重复比率
140

141
        重复比率 = 1 - (唯一 n-gram 数 / 总 n-gram 数)
142
        """
143
        words = text.split()
144
        if len(words) < n:
145
            return 0.0
146

147
        ngrams = [tuple(words[i:i+n]) for i in range(len(words)-n+1)]
148
        unique_ngrams = len(set(ngrams))
149
        total_ngrams = len(ngrams)
150

151
        return 1 - unique_ngrams / total_ngrams

1
class RefineWebPipeline:
2
    """
3
    RefineWeb: Refined Web Dataset
4

5
    相比 CCNet 的改进：
6
    1. 更激进的模板过滤
7
    2. 使用更强的质量分类器
8
    3. 保留文档级去重
9
    """
10

11
    def __init__(self):
12
        self.quality_classifier = None  # 用高质量数据训练的分类器
13

14
    def build_quality_classifier(self, positive_samples, negative_samples):
15
        """
16
        训练质量分类器
17

18
        positive: Wikipedia, Books, 人工筛选的高质量网页
19
        negative: 广告、导航栏、低质量网页
20
        """
21
        from sklearn.linear_model import LogisticRegression
22

23
        def extract_features(text):
24
            """提取质量相关特征"""
25
            features = {}
26

27
            # 词汇丰富度
28
            words = text.split()
29
            unique_words = set(words)
30
            features['vocab_richness'] = len(unique_words) / (len(words) + 1)
31

32
            # 平均句子长度
33
            sentences = sent_tokenize(text)
34
            features['avg_sent_len'] = np.mean([len(s.split()) for s in sentences])
35

36
            # 标点符号密度
37
            punct_count = sum(1 for c in text if c in '.,!?;:')
38
            features['punct_density'] = punct_count / (len(text) + 1)
39

40
            # 大写字母比率
41
            upper_count = sum(1 for c in text if c.isupper())
42
            features['upper_ratio'] = upper_count / (len(text) + 1)
43

44
            # URL 密度（低质量内容 URL 多）
45
            url_count = len(re.findall(r'http[s]?://', text))
46
            features['url_density'] = url_count / (len(text) + 1)
47

48
            return list(features.values())
49

50
        # 训练
51
        X_pos = [extract_features(s) for s in positive_samples]
52
        X_neg = [extract_features(s) for s in negative_samples]
53

54
        X = X_pos + X_neg
55
        y = [1] * len(X_pos) + [0] * len(X_neg)
56

57
        self.quality_classifier = LogisticRegression().fit(X, y)
58

59
    def heuristic_filters(self, doc):
60
        """
61
        启发式过滤器
62
        RefineWeb 使用的 10 个启发式规则
63
        """
64
        words = doc.split()
65

66
        # 规则 1: 最小长度
67
        if len(words) < 50:
68
            return False
69

70
        # 规则 2: 最大长度（避免文档过长）
71
        if len(words) > 100000:
72
            return False
73

74
        # 规则 3: 最小平均词长
75
        avg_word_len = len(doc) / len(words)
76
        if avg_word_len < 2.5 or avg_word_len > 15:
77
            return False
78

79
        # 规则 4: 最大特殊字符密度
80
        special_chars = sum(1 for c in doc if not c.isalnum() and not c.isspace())
81
        if special_chars / len(doc) > 0.3:
82
            return False
83

84
        # 规则 5: 最小句子数
85
        sentences = sent_tokenize(doc)
86
        if len(sentences) < 3:
87
            return False
88

89
        # 规则 6: 检查"段落"结构
90
        # 好的文档应该有多个段落
91
        paragraphs = doc.split('\n\n')
92
        if len(paragraphs) < 2:
93
            return False
94

95
        # 规则 7: 检查"锚文本"密度
96
        # 锚文本（链接文字）过多可能是导航/列表页
97
        anchor_ratio = doc.count('<a ') / (len(words) + 1)
98
        if anchor_ratio > 0.3:
99
            return False
100

101
        # 规则 8: 检查"call to action"
102
        cta_patterns = ['click here', 'sign up', 'subscribe', 'buy now', 'learn more']
103
        if any(p in doc.lower() for p in cta_patterns):
104
            return False
105

106
        # 规则 9: 检查"contact us"
107
        if 'contact' in doc.lower() and 'email' in doc.lower():
108
            return False  # 可能是联系页面
109

110
        # 规则 10: 检查"terms of service"
111
        if 'terms of service' in doc.lower() or 'privacy policy' in doc.lower():
112
            return False  # 法律页面
113

114
        return True

1
class DeduplicationStrategies:
2
    """
3
    多层次去重策略
4
    """
5

6
    def exact_deduplication(self, documents):
7
        """
8
        精确匹配去重（MinHash）
9

10
        时间复杂度: O(n)
11
        空间复杂度: O(n)
12
        """
13
        from datasketch import MinHash, MinHashEncoder
14

15
        minhashes = []
16
        for doc in documents:
17
            m = MinHash(num_perm=128)
18
            for word in doc.split():
19
                m.update(word.encode('utf8'))
20
            minhashes.append(m)
21

22
        # 编码并去重
23
        encoder = MinHashEncoder(num_perm=128)
24
        encoded = [encoder.encode(m) for m in minhashes]
25

26
        unique_idx = []
27
        seen = set()
28
        for i, e in enumerate(encoded):
29
            if e not in seen:
30
                seen.add(e)
31
                unique_idx.append(i)
32

33
        return [documents[i] for i in unique_idx]
34

35
    def near_deduplication(self, documents, threshold=0.8):
36
        """
37
        近似去重（SimHash 或 MinHash Jaccard）
38

39
        用于检测高度相似的文档
40
        threshold: Jaccard 相似度阈值
41
        """
42
        from datasketch import MinHash, MinHashLSH
43

44
        # 构建 MinHash 索引
45
        lsh = MinHashLSH(threshold=threshold, num_perm=128)
46

47
        for i, doc in enumerate(documents):
48
            m = MinHash(num_perm=128)
49
            for word in doc.split():
50
                m.update(word.encode('utf8'))
51
            lsh.insert(f"doc_{i}", m)
52

53
        # 查找并移除重复
54
        unique_docs = []
55
        duplicate_indices = set()
56

57
        for i, doc in enumerate(documents):
58
            if i in duplicate_indices:
59
                continue
60

61
            unique_docs.append(doc)
62

63
            # 查找相似文档
64
            m = MinHash(num_perm=128)
65
            for word in doc.split():
66
                m.update(word.encode('utf8'))
67

68
            duplicates = lsh.query(m)
69
            for dup_id in duplicates:
70
                dup_idx = int(dup_id.split('_')[1])
71
                if dup_idx != i:
72
                    duplicate_indices.add(dup_idx)
73

74
        return unique_docs
75

76
    def sentence_deduplication(self, text, min_occurrence=10):
77
        """
78
        句子级去重
79

80
        移除在整个语料中出现次数过多的句子
81
        这些通常是模板文本或版权声明
82
        """
83
        from collections import Counter
84

85
        sentences = sent_tokenize(text)
86

87
        # 统计句子出现次数
88
        sentence_counts = Counter(sentences)
89

90
        # 过滤出现过多的句子
91
        filtered = [
92
            s for s in sentences
93
            if sentence_counts[s] <= min_occurrence
94
        ]
95

96
        return ' '.join(filtered)
97

98
    def document_level_deduplication(self, documents):
99
        """
100
        文档级去重（基于内容哈希）
101

102
        方法：
103
        1. 提取文档的 k-mer
104
        2. 计算 MinHash
105
        3. 用 LSH 找相似文档
106
        4. 保留最长/质量最高的版本
107
        """
108
        from datasketch import MinHash, MinHashLSH
109

110
        lsh = MinHashLSH(threshold=0.9, num_perm=256)
111

112
        # 构建索引
113
        indexed = []
114
        for i, doc in enumerate(documents):
115
            m = self._compute_minhash(doc)
116
            key = f"doc_{i}"
117
            lsh.insert(key, m)
118
            indexed.append((i, doc, m))
119

120
        # 聚类并保留代表
121
        clusters = defaultdict(list)
122

123
        for i, doc, m in indexed:
124
            # 查找相似文档
125
            candidates = lsh.query(m)
126
            if candidates:
127
                cluster_id = int(candidates[0].split('_')[1])
128
                clusters[cluster_id].append((i, doc))
129
            else:
130
                clusters[i].append((i, doc))
131

132
        # 从每个 cluster 选代表（最长/最高质量）
133
        representatives = []
134
        for cluster in clusters.values():
135
            best = max(cluster, key=lambda x: len(x[1].split()))
136
            representatives.append(best[1])
137

138
        return representatives
139

140
    def _compute_minhash(self, text, k=5):
141
        """
142
        计算文本的 MinHash
143
        """
144
        from datasketch import MinHash
145

146
        words = text.split()
147
        m = MinHash(num_perm=256)
148

149
        for i in range(len(words) - k + 1):
150
            kmer = ' '.join(words[i:i+k])
151
            m.update(kmer.encode('utf8'))
152

153
        return m

1
class SFTDataCore:
2
    """
3
    SFT 数据的核心要素
4
    """
5

6
    @dataclass
7
    class SFTSample:
8
        """
9
        SFT 样本的三个核心要素
10
        """
11
        instruction: str   # 指令/问题/任务描述
12
        input: str        # 可选的输入上下文
13
        output: str       # 期望的回答
14

15
        def format(self, template="chatml"):
16
            """
17
            格式化为训练格式
18
            """
19
            if template == "chatml":
20
                return f"<|im_start|>system\nYou are a helpful assistant.<|im_end|>\n<|im_start|>user\n{self.instruction}\n{self.input}<|im_end|>\n<|im_start|>assistant\n{self.output}<|im_end|>"
21
            elif template == "alpaca":
22
                return f"Below is an instruction that describes a task, paired with an input that provides further context. Write a response that appropriately completes the request.\n\n### Instruction:\n{self.instruction}\n\n### Input:\n{self.input}\n\n### Response:\n{self.output}"
23
            elif template == "llama3":
24
                return f"<|begin_of_text|><|start_header_id|>system<|end_header_id|>\n\nYou are a helpful assistant.<|eot_id|>\n<|start_header_id|>user<|end_header_id|>\n\n{self.instruction}\n{self.input}<|eot_id|>\n<|start_header_id|>assistant<|end_header_id|>\n\n{self.output}<|eot_id|>"

1
class InstructionTemplates:
2
    """
3
    指令模板设计
4
    """
5

6
    TEMPLATES = {
7
        # 开放式问答
8
        "qa_open": [
9
            "{instruction}",
10
            "请回答以下问题：{instruction}",
11
            "Question: {instruction}\nAnswer:",
12
        ],
13

14
        # 带上下文的问答
15
        "qa_context": [
16
            "根据以下信息回答问题：\n{input}\n\n问题：{instruction}",
17
            "Context: {input}\n\nQuestion: {instruction}\n\nAnswer:",
18
        ],
19

20
        # 总结任务
21
        "summarize": [
22
            "请总结以下文本：\n{input}",
23
            "Summarize this: {input}",
24
        ],
25

26
        # 翻译任务
27
        "translate": [
28
            "请将以下文本翻译成{target_lang}：\n{input}",
29
            "Translate to {target_lang}: {input}",
30
        ],
31

32
        # 代码任务
33
        "code": [
34
            "请用{language}编写代码：\n{instruction}",
35
            "Write a {language} function that {instruction}",
36
        ],
37

38
        # 数学任务
39
        "math": [
40
            "求解：{instruction}",
41
            "Calculate: {instruction}",
42
            "Given that {input}, {instruction}",
43
        ],
44

45
        # 分析任务
46
        "analysis": [
47
            "分析以下内容：\n{input}",
48
            "请分析{instruction}，参考：\n{input}",
49
        ],
50
    }
51

52
    def generate_instruction_variants(self, base_instruction, num_variants=5):
53
        """
54
        生成指令变体
55

56
        使用 LLM 重写指令，增加多样性
57
        """
58
        prompt = f"""
59
        请将以下指令改写为 {num_variants} 种不同的表述方式。
60
        保持原意，但改变措辞和句式。
61

62
        原始指令：{base_instruction}
63

64
        改写要求：
65
        1. 使用不同的开头词
66
        2. 使用不同的句式结构
67
        3. 保持相同的任务要求
68
        4. 不要改变任务的核心目标
69
        """
70

71
        response = llm.generate(prompt)
72
        variants = extract_variants(response)
73

74
        return variants

1
class LLMDataGenerator:
2
    """
3
    使用 LLM 生成 SFT 数据
4
    """
5

6
    def __init__(self, generator_model, judge_model=None):
7
        self.generator = generator_model
8
        self.judge = judge_model
9

10
    def generate_from_prompts(self, prompts, task_type="general"):
11
        """
12
        从提示生成 SFT 数据
13
        """
14
        data = []
15

16
        for prompt in tqdm(prompts):
17
            response = self.generator.generate(prompt)
18

19
            # 质量检查
20
            if self.judge:
21
                quality = self.judge.score(prompt, response)
22
                if quality < 0.7:
23
                    continue
24

25
            data.append({
26
                "instruction": self.extract_instruction(prompt),
27
                "input": self.extract_input(prompt),
28
                "output": response,
29
                "source": "llm_generated",
30
            })
31

32
        return data
33

34
    def self_generate(self, model, instructions, num_samples_per_inst=1):
35
        """
36
        Self-Generation: 用模型自己生成回答
37

38
        典型流程：
39
        1. 准备多样化的指令集
40
        2. 用模型生成多个回答（不同 temperature）
41
        3. 过滤低质量回答
42
        4. 得到 (instruction, response) 对
43
        """
44
        data = []
45

46
        for instruction in tqdm(instructions):
47
            for _ in range(num_samples_per_inst):
48
                # 随机 temperature
49
                temp = random.choice([0.5, 0.7, 0.9, 1.1])
50

51
                response = model.generate(
52
                    instruction,
53
                    temperature=temp,
54
                    max_tokens=2048,
55
                    do_sample=True,
56
                )
57

58
                # 质量检查
59
                if self._passes_quality_check(response):
60
                    data.append({
61
                        "instruction": instruction,
62
                        "input": "",
63
                        "output": response,
64
                        "temperature": temp,
65
                    })
66

67
        return data
68

69
    def _passes_quality_check(self, response):
70
        """
71
        简单的质量检查
72
        """
73
        # 长度检查
74
        if len(response.split()) < 20:
75
            return False
76

77
        # 完整性检查
78
        if not response.strip().endswith(('.', '!', '?', '"', "'")):
79
            # 不完整可能表示截断
80
            if len(response) < 200:
81
                return False
82

83
        # 重复检查
84
        if self._has_excessive_repetition(response):
85
            return False
86

87
        return True
88

89
    def _has_excessive_repetition(self, text, n=3):
90
        """
91
        检查过度重复
92
        """
93
        words = text.lower().split()
94
        if len(words) < 10:
95
            return False
96

97
        # 检查连续 n 个词是否重复
98
        for i in range(len(words) - n * 2):
99
            window = ' '.join(words[i:i+n])
100
            next_window = ' '.join(words[i+n:i+2*n])
101
            if window == next_window:
102
                return True
103

104
        return False

1
class HumanAnnotationWorkflow:
2
    """
3
    人类标注工作流
4
    """
5

6
    def __init__(self, annotation_platform):
7
        self.platform = annotation_platform
8

9
    def setup_task(self, task_config):
10
        """
11
        配置标注任务
12
        """
13
        task = {
14
            "name": task_config["name"],
15
            "description": task_config["description"],
16

17
            # 输入字段
18
            "input_fields": [
19
                {"name": "instruction", "type": "text", "required": True},
20
                {"name": "context", "type": "text", "required": False},
21
            ],
22

23
            # 输出字段
24
            "output_fields": [
25
                {"name": "response", "type": "textarea", "required": True},
26
                {"name": "quality_score", "type": "rating", "min": 1, "max": 5},
27
                {"name": "issues", "type": "multiselect",
28
                 "options": ["incomplete", "incorrect", "harmful", "off-topic"]},
29
            ],
30

31
            # 质量控制
32
            "qualification_requirements": [
33
                {"type": "approval_rate", "threshold": 0.85},
34
                {"type": "completed_tasks", "threshold": 100},
35
            ],
36

37
            # 奖励设置
38
            "reward": task_config.get("reward_per_task", 0.10),
39
        }
40

41
        return self.platform.create_task(task)
42

43
    def annotate_instructions(self, instructions, num_workers=10):
44
        """
45
        标注指令-回答对
46

47
        典型流程：
48
        1. 分发任务给标注者
49
        2. 每个任务多人标注
50
        3. 收集并合并结果
51
        """
52
        # 分配任务
53
        tasks = self._distribute_tasks(instructions, num_workers)
54

55
        # 收集结果
56
        all_results = []
57

58
        for worker_id, worker_tasks in tasks.items():
59
            results = self.platform.get_worker_results(worker_id)
60
            all_results.extend(results)
61

62
        # 合并重复任务（多人标注同一任务）
63
        merged = self._merge_annotations(all_results)
64

65
        # 过滤低质量标注
66
        filtered = self._filter_by_agreement(merged)
67

68
        return filtered
69

70
    def _distribute_tasks(self, instructions, num_workers):
71
        """
72
        分发任务，确保每个任务被多人标注
73
        """
74
        import random
75

76
        tasks_per_worker = len(instructions) // num_workers
77

78
        tasks = {}
79
        for i in range(num_workers):
80
            start = i * tasks_per_worker
81
            end = start + tasks_per_worker if i < num_workers - 1 else len(instructions)
82
            tasks[f"worker_{i}"] = instructions[start:end]
83

84
        return tasks
85

86
    def _merge_annotations(self, annotations):
87
        """
88
        合并同一任务的多个标注
89

90
        策略：
91
        - 如果多数一致，选择多数结果
92
        - 如果一致，选择质量最高的
93
        """
94
        from collections import defaultdict
95

96
        by_instruction = defaultdict(list)
97

98
        for ann in annotations:
99
            key = ann["instruction"]
100
            by_instruction[key].append(ann)
101

102
        merged = []
103

104
        for instruction, anns in by_instruction.items():
105
            if len(anns) == 1:
106
                merged.append(anns[0])
107
                continue
108

109
            # 检查一致性
110
            responses = [a["response"] for a in anns]
111
            response_counts = Counter(responses)
112

113
            if response_counts.most_common(1)[0][1] >= len(anns) * 0.5:
114
                # 多数一致，选择多数结果
115
                best_response = response_counts.most_common(1)[0][0]
116
            else:
117
                # 不一致，选择质量最高的
118
                best_ann = max(anns, key=lambda a: a["quality_score"])
119
                best_response = best_ann["response"]
120

121
            merged.append({
122
                "instruction": instruction,
123
                "output": best_response,
124
                "agreement": response_counts.most_common(1)[0][1] / len(anns),
125
                "num_annotations": len(anns),
126
            })
127

128
        return merged
129

130
    def _filter_by_agreement(self, annotations, min_agreement=0.5):
131
        """
132
        按一致性过滤
133
        """
134
        return [a for a in annotations if a.get("agreement", 1.0) >= min_agreement]

1
class PreferenceDataFormat:
2
    """
3
    偏好数据的三种格式
4
    """
5

6
    @dataclass
7
    class PreferencePair:
8
        """
9
        标准偏好对
10
        """
11
        prompt: str
12
        chosen: str      # 被选中的回答
13
        rejected: str    # 被拒绝的回答
14

15
    @dataclass
16
    class PreferenceTriple:
17
        """
18
        偏好三元组（包含多个候选）
19
        """
20
        prompt: str
21
        responses: List[str]           # 候选回答列表
22
        ranking: List[int]            # 排序（0 = 最好）
23

24
    @dataclass
25
    class BinaryPreference:
26
        """
27
        二元偏好（正/负样本）
28
        """
29
        prompt: str
30
        response: str
31
        label: bool  # True = desirable, False = undesirable

1
class PreferenceAnnotationPipeline:
2
    """
3
    偏好标注流水线
4
    """
5

6
    def __init__(self, annotation_platform, judge_model=None):
7
        self.platform = annotation_platform
8
        self.judge = judge_model
9

10
    def generate_candidates(self, prompts, generator, num_candidates=4):
11
        """
12
        为每个 prompt 生成多个候选回答
13
        """
14
        candidates_data = []
15

16
        for prompt in tqdm(prompts):
17
            candidates = []
18

19
            for temp in [0.3, 0.5, 0.7, 0.9]:
20
                response = generator.generate(
21
                    prompt,
22
                    temperature=temp,
23
                    max_tokens=1024,
24
                )
25
                candidates.append(response)
26

27
            candidates_data.append({
28
                "prompt": prompt,
29
                "candidates": candidates,
30
            })
31

32
        return candidates_data
33

34
    def annotation_task(self, candidates_data, num_annotators=3):
35
        """
36
        偏好标注任务
37

38
        标注者需要：
39
        1. 阅读 prompt 和多个候选回答
40
        2. 选择最好的一个（或排序）
41
        3. 标记理由（可选）
42
        """
43
        task_config = {
44
            "name": "Preference Annotation",
45
            "instruction": """
46
            请阅读以下问题及其多个回答，选择最好的一个。
47

48
            评分标准：
49
            - 准确性：回答是否正确
50
            - 完整性：回答是否全面
51
            - 清晰度：回答是否易于理解
52
            - 有用性：回答是否有帮助
53

54
            请为每个回答打分（1-5）并选出最佳。
55
            """,
56
            "input_fields": [
57
                {"name": "prompt", "type": "display"},  # 只读
58
                {"name": "candidate_1", "type": "display"},
59
                {"name": "candidate_2", "type": "display"},
60
                {"name": "candidate_3", "type": "display"},
61
                {"name": "candidate_4", "type": "display"},
62
            ],
63
            "output_fields": [
64
                {"name": "scores", "type": "rating_group", "count": 4, "min": 1, "max": 5},
65
                {"name": "best", "type": "select", "options": ["1", "2", "3", "4"]},
66
                {"name": "reason", "type": "textarea"},
67
            ],
68
        }
69

70
        return task_config
71

72
    def build_preference_pairs(self, annotations):
73
        """
74
        从标注结果构建偏好对
75

76
        策略：两两比较构建 pair
77
        如果 A 得分 > B 得分，则 (prompt, A, B) 是一个正例
78
        """
79
        pairs = []
80

81
        for item in annotations:
82
            prompt = item["prompt"]
83
            candidates = item["candidates"]
84
            scores = item["scores"]  # [score_1, score_2, ...]
85

86
            # 构建所有两两比较
87
            for i in range(len(candidates)):
88
                for j in range(i+1, len(candidates)):
89
                    if scores[i] > scores[j]:
90
                        pairs.append({
91
                            "prompt": prompt,
92
                            "chosen": candidates[i],
93
                            "rejected": candidates[j],
94
                        })
95
                    elif scores[j] > scores[i]:
96
                        pairs.append({
97
                            "prompt": prompt,
98
                            "chosen": candidates[j],
99
                            "rejected": candidates[i],
100
                        })
101

102
        return pairs