training_code/model/dpo_data.py

"""
DPO data pipeline: loads UltraFeedback preference pairs.

Each example has a prompt + chosen response + rejected response.
We tokenize both (prompt+chosen) and (prompt+rejected), apply the same
chat template, and return them as pairs for DPO training.
"""

import torch
from torch.utils.data import Dataset, DataLoader
from datasets import load_dataset


CHAT_TEMPLATE = {
    "user_start": "<|user|>\n",
    "assistant_start": "<|assistant|>\n",
    "turn_end": "\n<|end|>\n",
}


def format_preference_pair(prompt, chosen_msgs, rejected_msgs):
    """Build chat-templated strings for chosen and rejected."""
    def build(messages):
        text = CHAT_TEMPLATE["user_start"] + prompt.strip() + CHAT_TEMPLATE["turn_end"]
        for msg in messages:
            role = msg.get("role", "assistant")
            content = msg.get("content", "").strip()
            if role == "assistant":
                text += CHAT_TEMPLATE["assistant_start"] + content + CHAT_TEMPLATE["turn_end"]
            elif role == "user":
                text += CHAT_TEMPLATE["user_start"] + content + CHAT_TEMPLATE["turn_end"]
        return text

    return build(chosen_msgs), build(rejected_msgs)


class DPODataset(Dataset):
    """
    Loads UltraFeedback preference pairs and tokenizes them.
    Returns (prompt_ids, chosen_ids, rejected_ids) with proper shifting.
    """

    def __init__(self, tokenizer, max_seq_len=2048, split="train",
                 cache_dir=None, max_samples=None):
        self.tokenizer = tokenizer
        self.max_seq_len = max_seq_len

        special_tokens = ["<|user|>", "<|assistant|>", "<|end|>"]
        vocab = tokenizer.get_vocab()
        new_tokens = [t for t in special_tokens if t not in vocab]
        if new_tokens:
            tokenizer.add_tokens(new_tokens, special_tokens=True)

        self.assistant_token_id = tokenizer.encode("<|assistant|>", add_special_tokens=False)[0]
        self.end_token_id = tokenizer.encode("<|end|>", add_special_tokens=False)[0]
        self.user_token_id = tokenizer.encode("<|user|>", add_special_tokens=False)[0]

        print(f"[DPO Data] Loading UltraFeedback preferences ({split})...")
        ds = load_dataset(
            "argilla/ultrafeedback-binarized-preferences-cleaned",
            split=split,
            cache_dir=cache_dir,
        )
        if max_samples:
            ds = ds.select(range(min(max_samples, len(ds))))
        print(f"[DPO Data] {len(ds)} preference pairs loaded")

        self.examples = []
        skipped = 0
        for i, row in enumerate(ds):
            prompt = row.get("prompt", "")
            chosen = row.get("chosen", [])
            rejected = row.get("rejected", [])

            if not prompt or not chosen or not rejected:
                skipped += 1
                continue

            chosen_text, rejected_text = format_preference_pair(prompt, chosen, rejected)

            chosen_ids = tokenizer.encode(chosen_text, add_special_tokens=False)
            rejected_ids = tokenizer.encode(rejected_text, add_special_tokens=False)

            # Truncate if needed
            if len(chosen_ids) > max_seq_len + 1:
                chosen_ids = chosen_ids[:max_seq_len + 1]
            if len(rejected_ids) > max_seq_len + 1:
                rejected_ids = rejected_ids[:max_seq_len + 1]

            if len(chosen_ids) < 10 or len(rejected_ids) < 10:
                skipped += 1
                continue

            # Find where the prompt ends (first <|assistant|> token)
            prompt_end = 0
            for j, tid in enumerate(chosen_ids):
                if tid == self.assistant_token_id:
                    prompt_end = j + 2  # skip <|assistant|> and \n
                    break

            self.examples.append({
                "chosen_ids": chosen_ids,
                "rejected_ids": rejected_ids,
                "prompt_len": prompt_end,
            })

            if (i + 1) % 20000 == 0:
                print(f"  Processed {i+1} pairs...")

        print(f"[DPO Data] {len(self.examples)} pairs ready, {skipped} skipped")

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

    def __getitem__(self, idx):
        ex = self.examples[idx]
        return {
            "chosen_ids": torch.tensor(ex["chosen_ids"], dtype=torch.long),
            "rejected_ids": torch.tensor(ex["rejected_ids"], dtype=torch.long),
            "prompt_len": ex["prompt_len"],
        }


def dpo_collate_fn(batch, pad_id=0):
    """Pad chosen and rejected sequences separately."""
    max_chosen = max(b["chosen_ids"].size(0) for b in batch)
    max_rejected = max(b["rejected_ids"].size(0) for b in batch)

    chosen_padded = []
    rejected_padded = []
    prompt_lens = []

    for b in batch:
        c_pad = max_chosen - b["chosen_ids"].size(0)
        r_pad = max_rejected - b["rejected_ids"].size(0)
        chosen_padded.append(torch.cat([b["chosen_ids"], torch.full((c_pad,), pad_id, dtype=torch.long)]))
        rejected_padded.append(torch.cat([b["rejected_ids"], torch.full((r_pad,), pad_id, dtype=torch.long)]))
        prompt_lens.append(b["prompt_len"])

    return {
        "chosen_ids": torch.stack(chosen_padded),
        "rejected_ids": torch.stack(rejected_padded),
        "prompt_lens": torch.tensor(prompt_lens, dtype=torch.long),
    }