一 整体方案

整个数据处理的实现思路，其实可以概括成一条很清晰的数据流水线：

1. 读取原始数据：从 JSON 文件中加载样本。

2. 统一样本格式：兼容 messages、conversations、prompt-response、instruction-output 等不同字段风格。

3. 渲染对话模板：把结构化消息列表转成带角色前缀的纯文本。

4. Tokenizer 编码：把文本转成 input_ids。

5. 构造监督标签：根据任务类型生成 labels。

6. 屏蔽无效损失位置：把不该参与训练的位置置为 -100。

   也就是说，这份代码并不是“模型层”，而是一个 训练数据适配层。

二 具体实现

2.1 Tokenizer

从头训练一个高质量的中文分词器（Tokenizer）不仅需要海量、比例均衡的清洗语料，还要反复调试词表大小（Vocab Size）和合并规则。如果处理不好，很容易遇到生僻字切碎、中英文字符编码冲突等问题。直接“白嫖”开源社区已经打磨好的分词器（比如 Qwen），往往是性价比最高的选择。

分词器的核心逻辑：BPE (Byte-Pair Encoding)

现代大模型（包括你选用的基线）几乎全部采用基于 BPE 的分词算法。它的逻辑并不复杂，核心就是找高频词缀：

1. 字节层面开始：一开始，它把所有中文文本拆成最基础的字节（Byte）或单字。

2. 高频合并：它会在语料库里统计哪些组合最常挨在一起出现。比如它发现“人”和“工”经常连着，“智”和“能”经常连着，就会把它们分别合并成“人工”和“智能”。

3. 最终成型：合并到预设的词表大小（比如 vocab_size = 64000）就停止。

最终，词表里既有单字（比如“的”、“了”），也有高频词组（比如“人工智能”、“测试”）。当输入一段文本时，Tokenizer 会像查字典一样，把文本切分成词表里最长的片段，然后输出这些片段对应的行号（整数 ID）。

通过以下几行代码直观感受文本是如何变成数字的：

from transformers import AutoTokenizer

# 1. 加载现成的分词器 (这里以一个开源路径为例)
tokenizer = AutoTokenizer.from_pretrained("Qwen/Qwen1.5-0.5B", trust_remote_code=True)

# 2. 准备一段预训练语料 (比如从 MNBVC 里抽出来的一句话)
text = "人工智能正在改变世界。在此背景下，如何实现模型轻量化？"

# 3. 编码 (Encode)：将文本转化为 ID 序列
# add_special_tokens=True 会在序列末尾自动加上 <eos> (结束符) 等特殊 Token
input_ids = tokenizer.encode(text, add_special_tokens=True)

print("切分后的文本:", [tokenizer.decode([id]) for id in input_ids])
print("模型吃进去的 ID:", input_ids)

# 4. 解码 (Decode)：将 ID 还原成文本，用于模型生成结果后向人类展示
recovered_text = tokenizer.decode(input_ids)
print("还原后的文本:", recovered_text)

它的输出大概会是这样：

• 切分后的文本: ['人工', '智能', '正在', '改变', '世界', '。', '在此', '背景', '下', '，', '如何', '实现', '模型', '轻', '量', '化', '？', '<|endoftext|>']

• 模型吃进去的 ID: [104106, 102434, 101835, 102144, 100780, 11319, 107052, 105658, 100067, 11302, 101185, 101569, 103239, 101673, 100412, 100345, 11438, 151643]

可以看到，这串由数字组成的列表，就是刚才封装的 PocketLLM 类里 forward(input_ids) 所需要的输入格式！

特殊 Token 的作用

在接下来的数据拼接和训练中，有几个特殊的 ID 你必须牢记在心（具体的 ID 数字取决于你选用的分词器）：

• <eos> (End of Sequence)：句子结束符。告诉模型“这句话到此为止”。在预训练把多篇文章拼在一起时，全靠这个符号来隔开。

• <pad> (Padding)：填充符。矩阵计算必须是对齐的（比如固定的 max_seq_len = 1024）。如果一句话只有 100 个 Token，剩下的 924 个位置全都要填上 <pad> 的 ID，并在算 Loss 的时候把这些位置忽略掉。

2.2 训练数据适配器

这个适配器的逻辑较为简单，主要目标就是把训练数据能够转成训练时的tensor张量，__getitem__的主要流程如下：

• 取出 sample['text']
• tokenizer 分词
• 截断到 max_length - 2
• 手动加 bos/eos
• 补 pad
• 转成 torch.tensor
• 构造 labels
• 把 padding 位置设成 -100，避免参与 loss

class PretrainDataset(Dataset):
    def __init__(self, data_path, tokenizer, max_len=512):
        super().__init__()
        self.tokenizer = tokenizer
        self.max_length = max_len
        self.samples = load_dataset('json', data_files=data_path, split='train')

    def __len__(self):
        return len(self.samples)

    def __getitem__(self, index):
        sample = self.samples[index]
        tokens = self.tokenizer(
            str(sample['text']),
            add_special_tokens=False,
            truncation=True,
            max_length=self.max_length - 2, # -2 for [CLS] and [SEP]
        )
        tokens = [self.tokenizer.bos_token_id] + tokens + [self.tokenizer.eos_token_id]  # 手动添加开始与结束符号
        input_ids = tokens + [self.tokenizer.pad_token_id] * (self.max_length - len(tokens)) # 填充
        input_ids = torch.tensor(input_ids, dtype=torch.long) # 转换成张量
        labels = input_ids.clone() # 通过clone保持与原序列的一致
        labels[input_ids == self.tokenizer.pad_token_id] = -100 # 将填充位置的标签设为-100
        return input_ids, labels

2.3 SFT数据训练适配器

通常原始的数据格式是

{
  "conversations": [
    {"role": "user", "content": "..."},
    {"role": "assistant", "content": "..."}
  ]
}

但训练循环和模型真正要用的是这样：

• input_ids: 一整条 token 序列
• labels: 只在该学习的位置保留 token id，其他位置全是 -100

所以我们需要的就是把“原始多轮对话数据”适配成“模型可直接做 SFT 训练的 (input_ids, labels) 样本”。

我们先把结构化对话，转成模型实际训练时看到的文本prompt

def create_chat_prompt(self, conversations):
    """
    把结构化对话，转成模型实际训练时看到的文本prompt,
    类似于：
        "<bos>user\n你好\n<eos>\n<bos>assistant\n你好，有什么可以帮你？<eos>\n"
    """
    messages = []
    tools = None
    for message in conversations:
        message = dict(message)
        if message.get("role") == "system" and message.get("tools"):
            tools = json.loads(message["tools"]) if isinstance(message["tools"], str) else message["tools"]
        if message.get("tool_calls") and isinstance(message["tool_calls"], str):
            message["tool_calls"] = json.loads(message["tool_calls"])
        messages.append(message)
    return self.tokenizer.apply_chat_template(
        messages,
        tokenize=False,
        add_generation_prompt=False,
        tools=tools
    )

接下来就是很重要的一个步骤，SFT / 对话微调 场景，核心思想是：

• 不让整段输入都参与 loss
• 只让某些目标片段参与 loss
• 这些目标片段通常是 assistant 的回答部分

所以我们需要根据 bos_id 和 eos_id 标记出“哪些 token 需要参与训练损失”，其余位置全部设为 -100 忽略掉。

f generate_labels(self, input_ids):
    labels = [-100] * len(input_ids) # 统一设置为-100
    i = 0
    while i < len(input_ids):
        if input_ids[i:i + len(self.bos_id)] == self.bos_id: # 找到开始符号
            start = i + len(self.bos_id) 
            end = start 
            while end < len(input_ids): # 找到结束符号
                if input_ids[end:end + len(self.eos_id)] == self.eos_id:
                    break
                end += 1
            for j in range(start, min(end + len(self.eos_id), self.max_length)): # 将中间的文本标记为标签
                labels[j] = input_ids[j]
            i = end + len(self.eos_id) if end < len(input_ids) else len(input_ids) # 跳转到下一个开始符号
        else:
            i += 1
    return labels

接下来就行编排总流程

def __getitem__(self, index):
    sample = self.samples[index]
    conversations = pre_processing_chat(sample['conversations']) # 添加system提示
    prompt = self.create_chat_prompt(conversations) # 创建prompt
    prompt = post_processing_chat(prompt) # 移除空思考
    input_ids = self.tokenizer(prompt).input_ids[:self.max_length] # 截取长度
    input_ids += [self.tokenizer.pad_token_id] * (self.max_length - len(input_ids)) # 填充
    labels = self.generate_labels(input_ids) # 生成labels
    return torch.tensor(input_ids, dtype=torch.long), torch.tensor(labels, dtype=torch.long)

这里我们还用到了两个辅助工具，为对话添加system角色以及移除空的思考，代码如下：

def pre_processing_chat(conversations, add_system_ratio=0.2):
    if any(conv.get('tools') for conv in conversations): return conversations

    SYSTEM_PROMPTS = [
        "你是一个知识丰富的AI，尽力为用户提供准确的信息。",
        "你是PocketLLM，一个小巧但有用的语言模型。",
        "你是一个专业的AI助手，请提供有价值的回答。",
        "你是PocketLLM，请尽力帮助用户解决问题。",
        "你是一个可靠的AI，请给出准确的回答。",
        "You are a helpful AI assistant.",
        "You are PocketLLM, a lightweight intelligent assistant.",
        "You are a friendly chatbot. Please answer the user's questions carefully.",
        "You are a knowledgeable AI. Try your best to provide accurate information.",
        "You are PocketLLM, a small but useful language model."
    ]
    # 概率性添加system
    if conversations[0].get('role') != 'system':
        if random.random() < add_system_ratio:
            return [{'role': 'system', 'content': random.choice(SYSTEM_PROMPTS)}] + conversations
    return conversations

def post_processing_chat(prompt_content, empty_think_ratio=0.2):
    # 以80%概率移除空思考标签
    if '<think>\n\n</think>\n\n' in prompt_content and random.random() > empty_think_ratio:
        prompt_content = prompt_content.replace('<think>\n\n</think>\n\n', '')
    return prompt_content

2.4 DPO数据训练适配器

DPODataset 做了三件关键事：

• 用 apply_chat_template 把 chosen/rejected 渲染成训练文本
• 用 generate_loss_mask() 只让 assistant 回复部分参与偏好损失
• 返回 x_chosen / y_chosen / mask_chosen 和 x_rejected / y_rejected / mask_rejected

大致的逻辑思想其实与SFT数据训练师适配器大致相同，只是一些细节实现不同，这里也在赘述，代码如下：

class DPODataset(Dataset):
    def __init__(self, file_path, tokenizer, max_length=4096):
        super().__init__()
        self.tokenizer = tokenizer
        self.max_length = max_length
        self.padding = tokenizer.pad_token_id if tokenizer.pad_token_id is not None else 0
        self.bos_id = tokenizer(f'{tokenizer.bos_token}assistant\n', add_special_tokens=False).input_ids
        self.eos_id = tokenizer(f'{tokenizer.eos_token}\n', add_special_tokens=False).input_ids
        self.samples = load_dataset('json', data_files=file_path, split='train')

    def __len__(self):
        return len(self.samples)

    def __getitem__(self, index):
        sample = self.samples[index]
        chosen = sample['chosen']  # 是一个 list，里面包含若干 {role, content}
        rejected = sample['rejected']  # 同上
        chosen_prompt = self.tokenizer.apply_chat_template(
            chosen, tokenize=False, add_generation_prompt=False
        )
        chosen_prompt = post_processing_chat(chosen_prompt) # 构建chosen prompt

        rejected_prompt = self.tokenizer.apply_chat_template(
            rejected, tokenize=False, add_generation_prompt=False
        )
        rejected_prompt = post_processing_chat(rejected_prompt) # 构建chosen prompt
        chosen_encoding = self.tokenizer(
            chosen_prompt, truncation=True, max_length=self.max_length, padding='max_length'
        ) # 进行tokenize
        rejected_encoding = self.tokenizer(
            rejected_prompt, truncation=True, max_length=self.max_length, padding='max_length'
        )

        chosen_input_ids = chosen_encoding['input_ids']
        chosen_loss_mask = self.generate_loss_mask(chosen_input_ids) # 生成loss mask，只让assistant参与训练

        rejected_input_ids = rejected_encoding['input_ids']
        rejected_loss_mask = self.generate_loss_mask(rejected_input_ids) # 同上
        x_chosen = torch.tensor(chosen_input_ids[:-1], dtype=torch.long)
        y_chosen = torch.tensor(chosen_input_ids[1:], dtype=torch.long)
        mask_chosen = torch.tensor(chosen_loss_mask[1:], dtype=torch.long)
        x_rejected = torch.tensor(rejected_input_ids[:-1], dtype=torch.long)
        y_rejected = torch.tensor(rejected_input_ids[1:], dtype=torch.long)
        mask_rejected = torch.tensor(rejected_loss_mask[1:], dtype=torch.long)

        return {
            'x_chosen': x_chosen,
            'y_chosen': y_chosen,
            'mask_chosen': mask_chosen,
            'x_rejected': x_rejected,
            'y_rejected': y_rejected,
            'mask_rejected': mask_rejected
        }

    def generate_loss_mask(self, input_ids):
        loss_mask = [0] * len(input_ids)
        i = 0
        while i < len(input_ids):
            if input_ids[i:i + len(self.bos_id)] == self.bos_id:
                start = i + len(self.bos_id)
                end = start
                while end < len(input_ids):
                    if input_ids[end:end + len(self.eos_id)] == self.eos_id:
                        break
                    end += 1
                for j in range(start, min(end + len(self.eos_id), self.max_length)):
                    loss_mask[j] = 1
                i = end + len(self.eos_id) if end < len(input_ids) else len(input_ids)
            else:
                i += 1
        return loss_mask

2.5 RLAIF数据训练适配器

RLAIF的输入数据还是 conversations 格式，但 RLAIFDataset 不会把最后这条 assistant 当监督标签喂给模型，而是：

• 把前面的上下文拼成 prompt
• 留出最后该由模型自己生成的位置
• 返回给 rollout 阶段

所以它的输出不是 (input_ids, labels)，而是：

{
    "prompt": "...",
    "answer": ""
}

这个 answer 基本只是个占位符。具体代码如下：

class RLAIFDataset(Dataset):
    def __init__(self, jsonl_path, tokenizer, max_length=1024, thinking_ratio=0.5):
        super().__init__()
        self.tokenizer = tokenizer
        self.max_length = max_length
        self.thinking_ratio = thinking_ratio  # 按概率开启 thinking
        self.samples = load_dataset('json', data_files=jsonl_path, split='train')
        self.bos_id = tokenizer(f'{tokenizer.bos_token}assistant', add_special_tokens=False).input_ids
        self.eos_id = tokenizer(f'{tokenizer.eos_token}', add_special_tokens=False).input_ids

    def __len__(self):
        return len(self.samples)

    def create_chat_prompt(self, conversations):
        conversations = pre_processing_chat(conversations)
        use_thinking = random.random() < self.thinking_ratio
        return self.tokenizer.apply_chat_template(
            conversations[:-1],
            tokenize=False,
            open_thinking=use_thinking,
            add_generation_prompt=True
        )
    def __getitem__(self, index):
        sample = self.samples[index]
        prompt = self.create_chat_prompt(sample['conversations'])

        return {
            'prompt': prompt,
            'answer': ""
        }

三 SFT、DPO、RLAIF区别

用通俗的话来总结：

• SFT：教模型“标准答案应该怎么说”
• DPO：教模型“两个答案里哪个更好”
• RLAIF：教模型“在试错中靠奖励自己逼近更优策略”

3.1 SFT(Supervised Fine-Tuning)，监督微调

SFT的数据长这样：

{
  "conversations": [
    {"role": "user", "content": "什么是机器学习？"},
    {"role": "assistant", "content": "机器学习是让模型从数据中学习规律的方法。"}
  ]
}

训练目标很直接：

• 给模型看用户输入
• 让它学习输出人工或高质量 teacher 给出的答案

所以它学的是：“给定这个问题，标准回答应该长成什么样。”

但是SFT同样也有缺点：

• 它只会模仿答案
• 不直接理解“为什么这个答案比另一个更好”
• 如果数据质量一般，模型会学到平均化回答

3.2 DPO(Direct Preference Optimization)，直接偏好优化

它不再只给一个标准答案，而是给一对答案：

• chosen：更好的
• rejected：更差的

比如：

{
  "chosen": [...一个更好的对话...],
  "rejected": [...一个更差的对话...]
}

训练目标不是“复读 chosen”，而是：让模型更倾向 chosen，而不是 rejected。也就是说，DPO 学的是偏序关系：

• 这个回答更有帮助
• 哪个回答更啰嗦

所以DPO 不是告诉学生标准答案，而是让学生反复比较：这两个答案，哪个更像好助手。

但是这也意味着DPO：

• 需要高质量 chosen/rejected 配对数据
• 偏好标注贵

3.3 RLAIF（Reinforcement Learning from AI Feedback）AI 反馈的强化学习

RLAIF 的核心不是“监督答案”，也不是“偏好对比较”，而是模型先自己生成，再根据奖励信号更新策略。

RLAIF 常见流程可以压成 4 步：

1. 准备 prompt
2. 让当前模型对每个 prompt 生成多个候选回答
3. 让评审 AI 按规则/原则/标准答案对这些候选打分或排序
4. 用这些分数做强化学习或偏好优化

这种训练方式带来的优点：

• 不依赖人工逐条写标准答案
• 很适合可验证任务：数学、代码、工具调用、Agent
• 可以在线探索，比 SFT/DPO 更主动

同样的也有缺点：

• 最不稳定
• 很容易 reward hacking
• 奖励设计差，模型就会朝错方向猛冲

3.4 技术选择

2.2章节是通常最基础的预训练数据适配器，通常我们使用这个数据适配器来完成一次底座大语言模型的训练，这个时候的模型通常只学习到了：

• 词法、语法、上下文续写能力
• 通用知识记忆
• 一定的模式归纳和推理潜力

但是不知道怎么“听指令”，不知道什么时候该答、怎么按聊天格式答，所以预训练模型更像“会说话的原始大脑”，不是“可用助手”。

3.4.1 SFT

这时候通常我们就需要进行SFT微调，SFT 做的不是补知识，而是补“行为格式”，它把模型从“会续写文本”变成“会按指令回答”。通常训练过后，模型具备了

• 看到 user 指令后，应该输出 assistant 回答
• 多轮对话如何接续
• 什么叫任务完成
• 工具调用、思维标签、结构化回答这些格式怎么写
• 什么叫符合你定义的输出风格

这些能力，所以大多数真正可用的聊天模型，SFT 基本都是必经阶段。

但是这时候的模型的能力还是大而全，但是对于某一个细分的领域还达不到精，这时候通常我们会使用LoRA进行低成本的领域/风格/任务适配，可以让模型成为一个小而精的垂直领域模型。

3.4.2 DPO

DPO跳过了传统 RLHF 需要训练独立 Reward Model（奖励模型）的步骤，直接用数学推导将强化学习问题转化为分类问题，在人类偏好数据上优化策略模型。通常是在SFT 之后的 Alignment（对齐阶段），教模型“怎样的回答更好”。

当你有了大量优质的(Prompt, Chosen_Response, Rejected_Response)数据对的时候，你就可以在SFT或者LoRA之后进行再一次的微调，在“已经会答”的基础上，把回答往更偏好的方向推。

3.4.3 RLAIF

近期因 DeepSeek-R1 受到广泛关注。它省去了传统 PPO 中的 Critic 模型（价值模型），通过对同一个 Prompt 采样一组输出，并在组内计算相对优势（Advantage）来进行策略梯度更新。

RLAIF主要用于激发模型的“深层推理能力”，特别是数学推导、代码编写或复杂的多 Agent 协作规划，不需要昂贵的人类标注偏好，任务的对错可以通过编译器（代码是否运行通过）、校验器（最终答案是否符合特定结果）自动生成 Reward。

所以通常RLAIF也是建立在SFT或者LoRA之后的，如果基座模型本省能力就很差，那么生成质量太差，reward 基本全是噪声，并且探索空间太大，训练极不稳定。