sft_dataset.py

"""
Batch iterator for packed SFT data (mirrors ``data_loader.data_loader.get_batch_iterator``
but carries the per-token loss mask alongside the tokens).

The HDF5 file has two aligned datasets, ``tokens`` and ``loss_mask``, both shape
(N, context_length), produced by ``scripts/prepare_sft_data.py``.
"""

from __future__ import annotations

from typing import Iterator

import h5py
import numpy as np
import torch


def get_sft_batch_iterator(
    data_path: str,
    batch_size: int,
    device: str = "cpu",
    *,
    rank: int = 0,
    world_size: int = 1,
    shuffle: bool = True,
    infinite: bool = True,
) -> Iterator[tuple[torch.Tensor, torch.Tensor, int]]:
    """
    Yield ``(tokens, loss_mask, epoch)`` batches from a packed SFT HDF5 file.

    Rows are sharded across ranks (each rank sees a disjoint stride) so DDP data-parallel
    training covers the dataset once per epoch. Set ``infinite=False`` to stop after one
    pass (used for evaluation).
    """
    with h5py.File(data_path, "r") as f:
        tokens = f["tokens"]
        masks = f["loss_mask"]
        n = tokens.shape[0]
        idxs = np.arange(rank, n, world_size)
        epoch = 0
        rng = np.random.default_rng(1234 + rank)
        while True:
            if shuffle:
                rng.shuffle(idxs)
            for start in range(0, len(idxs) - batch_size + 1, batch_size):
                batch = np.sort(idxs[start: start + batch_size])  # sorted for h5py fancy-index
                tk = torch.tensor(np.asarray(tokens[batch]), dtype=torch.long, device=device)
                mk = torch.tensor(np.asarray(masks[batch]), dtype=torch.long, device=device)
                yield tk, mk, epoch
            epoch += 1
            if not infinite:
                return