"""Character-level dataset for TinyGPT. The corpus is bundled in the package so every node derives an *identical* vocab (sorted unique characters) without any network round-trip — critical, because the server and all workers must agree on token ids and ``vocab_size`` for gradients to be meaningful. A worker can take a disjoint ``shard`` of the corpus (``--shard i/N``) to simulate federated, non-IID data spread across the swarm. """ from __future__ import annotations import os from dataclasses import dataclass from typing import Dict, List, Optional, Tuple import torch _CORPUS_PATH = os.path.join(os.path.dirname(__file__), "corpus.txt") def load_corpus(path: Optional[str] = None) -> str: """Read the training text. Defaults to the bundled corpus. ``SWARM_CORPUS`` env var or an explicit ``path`` overrides it (e.g. a larger corpus mounted on the server). The vocab is still derived from whatever text is loaded, so every node MUST use the same corpus. """ path = path or os.environ.get("SWARM_CORPUS") or _CORPUS_PATH with open(path, "r", encoding="utf-8") as f: return f.read() @dataclass class CharTokenizer: stoi: Dict[str, int] itos: Dict[int, str] @property def vocab_size(self) -> int: return len(self.stoi) @classmethod def from_text(cls, text: str) -> "CharTokenizer": # Sorted for determinism across machines/runs. chars = sorted(set(text)) stoi = {ch: i for i, ch in enumerate(chars)} itos = {i: ch for ch, i in stoi.items()} return cls(stoi=stoi, itos=itos) def encode(self, s: str) -> List[int]: return [self.stoi[c] for c in s if c in self.stoi] def decode(self, ids: List[int]) -> str: return "".join(self.itos[i] for i in ids) @dataclass class Dataset: data: torch.Tensor # 1-D LongTensor of token ids tokenizer: CharTokenizer block_size: int @property def vocab_size(self) -> int: return self.tokenizer.vocab_size def get_batch( self, batch_size: int, generator: Optional[torch.Generator] = None ) -> Tuple[torch.Tensor, torch.Tensor]: """Sample a random batch of (inputs, targets), each (batch_size, block_size).""" n = self.data.size(0) - self.block_size - 1 if n <= 0: raise ValueError( f"shard too small ({self.data.size(0)} tokens) for block_size {self.block_size}" ) ix = torch.randint(0, n, (batch_size,), generator=generator) x = torch.stack([self.data[i : i + self.block_size] for i in ix]) y = torch.stack([self.data[i + 1 : i + 1 + self.block_size] for i in ix]) return x, y def build_dataset( block_size: int, shard: int = 0, num_shards: int = 1, corpus_path: Optional[str] = None, ) -> Dataset: """Build a Dataset, optionally restricted to one shard of the corpus. The tokenizer/vocab is always derived from the *full* corpus so it is identical on every node; only the token slice used for sampling is sharded. """ text = load_corpus(corpus_path) tokenizer = CharTokenizer.from_text(text) ids = torch.tensor(tokenizer.encode(text), dtype=torch.long) if num_shards > 1: if not (0 <= shard < num_shards): raise ValueError(f"shard {shard} out of range for num_shards {num_shards}") shard_len = ids.size(0) // num_shards start = shard * shard_len end = start + shard_len if shard < num_shards - 1 else ids.size(0) ids = ids[start:end] return Dataset(data=ids, tokenizer=tokenizer, block_size=block_size)