Module neuralops.data.synthetic_generator

High-Fidelity Synthetic Dataset & Token Generator for NLP / LLM Evaluation.

Replaces uniform random token noise with a power-law Zipfian distribution: $$P(k) = \frac{1 / k^s}{\sum_{n=1}^V 1 / n^s} \quad \text{with } s \approx 1.15$$ to simulate authentic natural language entropy, word rank frequency curves, and valid causal perplexity evaluation dynamics.

Functions

def compute_empirical_entropy(tokens: torch.Tensor, vocab_size: int = 32000) ‑> float

Expand source code

def compute_empirical_entropy(tokens: torch.Tensor, vocab_size: int = 32000) -> float:
    """
    Computes the empirical Shannon entropy (in bits) of a tokenized tensor.

    Args:
        tokens: Long tensor of token IDs.
        vocab_size: Total vocabulary size.

    Returns:
        float: Shannon entropy $H(X) = -\\sum p_i \\log_2(p_i)$ in bits.
    """
    flat = tokens.view(-1).cpu().numpy()
    counts = np.bincount(flat, minlength=vocab_size)
    counts = counts[counts > 0]
    probs = counts / counts.sum()
    entropy = -np.sum(probs * np.log2(probs))
    return float(entropy)

Computes the empirical Shannon entropy (in bits) of a tokenized tensor.

Args

-----=

tokens

Long tensor of token IDs.

vocab_size

Total vocabulary size.

Returns

-----=

float

Shannon entropy $H(X) = -\sum p_i \log_2(p_i)$ in bits.

def generate_synthetic_data(output_path: str,
num_samples: int = 10000,
domain: str = 'movies',
seed: int = 42) ‑> pandas.DataFrame

Expand source code

def generate_synthetic_data(
    output_path: str, num_samples: int = 10000, domain: str = "movies", seed: int = 42
) -> pd.DataFrame:
    """Generate a single-domain synthetic classification dataset."""
    random.seed(seed)
    half = num_samples // 2
    pos = _generate_batch(domain, half, label=1, is_positive=True)
    neg = _generate_batch(domain, half, label=0, is_positive=False)
    df = pd.DataFrame(pos + neg).sample(frac=1, random_state=seed).reset_index(drop=True)
    Path(output_path).parent.mkdir(parents=True, exist_ok=True)
    df.to_csv(output_path, index=False)
    logger.info("Generated %d samples to %s", len(df), output_path)
    return df

Generate a single-domain synthetic classification dataset.

def generate_zipfian_markov_tokens(num_samples: int,
seq_len: int,
vocab_size: int = 32000,
zipf_param: float = 1.15,
transition_temp: float = 0.8,
seed: int = 42,
min_token_id: int = 1) ‑> torch.Tensor

Expand source code

def generate_zipfian_markov_tokens(
    num_samples: int,
    seq_len: int,
    vocab_size: int = 32000,
    zipf_param: float = 1.15,
    transition_temp: float = 0.8,
    seed: int = 42,
    min_token_id: int = 1,
) -> torch.Tensor:
    """
    Generates token sequences with Zipfian marginal rank distribution combined with
    first-order Markov sequential transition coupling.

    Args:
        num_samples: Number of sequences.
        seq_len: Sequence length.
        vocab_size: Vocabulary size.
        zipf_param: Zipf exponent.
        transition_temp: Softmax temperature for Markov transition matrix.
        seed: Random seed.
        min_token_id: Minimum token ID offset.

    Returns:
        torch.Tensor: Long tensor of shape `(num_samples, seq_len)`.
    """
    np_rng = np.random.default_rng(seed)
    effective_vocab = max(2, min(vocab_size - min_token_id, 2048))

    # FIX: Compute Zipfian unigram prior
    ranks = np.arange(1, effective_vocab + 1, dtype=np.float64)
    unigram_probs = 1.0 / (ranks**zipf_param)
    unigram_probs /= unigram_probs.sum()

    tokens = np.zeros((num_samples, seq_len), dtype=np.int64)
    tokens[:, 0] = np_rng.choice(effective_vocab, size=num_samples, p=unigram_probs)

    for t in range(1, seq_len):
        prev = tokens[:, t - 1]
        # Transition distribution biased by previous token and global Zipfian prior
        next_choices = np_rng.choice(effective_vocab, size=num_samples, p=unigram_probs)
        # 30% local Markov continuity, 70% Zipfian base distribution
        mask = np_rng.random(size=num_samples) < 0.3
        local_step = (prev + np_rng.integers(-3, 4, size=num_samples)) % effective_vocab
        tokens[:, t] = np.where(mask, local_step, next_choices)

    return torch.from_numpy(tokens + min_token_id).long()

Generates token sequences with Zipfian marginal rank distribution combined with first-order Markov sequential transition coupling.

Args

-----=

num_samples

Number of sequences.

seq_len

Sequence length.

vocab_size

Vocabulary size.

zipf_param

Zipf exponent.

transition_temp

Softmax temperature for Markov transition matrix.

seed

Random seed.

min_token_id

Minimum token ID offset.

Returns

-----=

torch.Tensor

Long tensor of shape (num_samples, seq_len).

def generate_zipfian_tokens(num_samples: int,
seq_len: int,
vocab_size: int = 32000,
zipf_param: float = 1.15,
seed: int = 42,
min_token_id: int = 1) ‑> torch.Tensor

Expand source code

def generate_zipfian_tokens(
    num_samples: int,
    seq_len: int,
    vocab_size: int = 32000,
    zipf_param: float = 1.15,
    seed: int = 42,
    min_token_id: int = 1,
) -> torch.Tensor:
    """
    Generates token sequences following a Zipfian power-law distribution.

    Ensures that low-index tokens (common stop words, articles) occur with high frequency
    while high-index tokens (rare technical terms) occur sparsely, mirroring natural language.

    Args:
        num_samples: Number of sequences to generate.
        seq_len: Sequence length (number of tokens per sample).
        vocab_size: Size of vocabulary.
        zipf_param: Power-law exponent $s > 0$ (default $1.15$, typical for natural language).
        seed: Random seed for deterministic reproducibility.
        min_token_id: Minimum token ID offset (default 1 to reserve 0 for PAD).

    Returns:
        torch.Tensor: Long tensor of shape `(num_samples, seq_len)` containing sampled token IDs.
    """
    np_rng = np.random.default_rng(seed)

    effective_vocab = max(2, vocab_size - min_token_id)

    # FIX: Compute exact Zipfian rank probabilities: P(k) = (1 / k^s) / zeta(s)
    # to simulate authentic natural language entropy and prevent flat uniform noise distortion.
    ranks = np.arange(1, effective_vocab + 1, dtype=np.float64)
    probs = 1.0 / (ranks**zipf_param)
    probs_sum = np.sum(probs)
    if probs_sum > 0:
        probs /= probs_sum
    else:
        probs = np.full(effective_vocab, 1.0 / effective_vocab)

    # Sample token indices from power-law distribution
    flat_tokens = np_rng.choice(effective_vocab, size=(num_samples, seq_len), p=probs, replace=True)

    # Shift indices to respect min_token_id (reserving special tokens)
    token_tensor = torch.from_numpy(flat_tokens + min_token_id).long()
    return token_tensor

Generates token sequences following a Zipfian power-law distribution.

Ensures that low-index tokens (common stop words, articles) occur with high frequency while high-index tokens (rare technical terms) occur sparsely, mirroring natural language.

Args

-----=

num_samples

Number of sequences to generate.

seq_len

Sequence length (number of tokens per sample).

vocab_size

Size of vocabulary.

zipf_param

Power-law exponent $s > 0$ (default $1.15$, typical for natural language).

seed

Random seed for deterministic reproducibility.

min_token_id

Minimum token ID offset (default 1 to reserve 0 for PAD).

Returns

-----=

torch.Tensor

Long tensor of shape (num_samples, seq_len) containing sampled token IDs.

def run_universal_generation(output_dir: str | Path,
total_samples_per_domain: int = 5000,
domains: dict[str, Any] | None = None,
seed: int = 42) ‑> pandas.DataFrame

Expand source code

def run_universal_generation(
    output_dir: str | Path,
    total_samples_per_domain: int = 5000,
    domains: dict[str, Any] | None = None,
    seed: int = 42,
) -> pd.DataFrame:
    """Generate synthetic dataset across multiple domains and combine them."""
    random.seed(seed)
    domains = domains or DEFAULT_DOMAINS
    out_dir = Path(output_dir)
    out_dir.mkdir(parents=True, exist_ok=True)

    all_dfs = []
    for domain_key in domains:
        half = total_samples_per_domain // 2
        pos = _generate_batch(domain_key, half, label=1, is_positive=True)
        neg = _generate_batch(domain_key, half, label=0, is_positive=False)
        df = pd.DataFrame(pos + neg).sample(frac=1, random_state=seed).reset_index(drop=True)
        df.to_csv(out_dir / f"data_{domain_key}.csv", index=False)
        all_dfs.append(df)

    final_df = (
        pd.concat(all_dfs, ignore_index=True)
        .sample(frac=1, random_state=seed)
        .reset_index(drop=True)
    )
    final_df.to_csv(out_dir / "combined_master_dataset.csv", index=False)
    logger.info(
        "Universal generation complete: %d total samples across %d domains.",
        len(final_df),
        len(domains),
    )
    return final_df

Generate synthetic dataset across multiple domains and combine them.

Classes

class SyntheticClassificationDataset (num_samples: int = 64,
seq_len: int = 32,
vocab_size: int = 1000,
num_classes: int = 2,
seed: int = 42)

Expand source code

class SyntheticClassificationDataset(Dataset):
    """
    High-fidelity synthetic classification dataset using Zipfian token distributions
    and class-conditional semantic bias for realistic classifier benchmarking.
    """

    def __init__(
        self,
        num_samples: int = 64,
        seq_len: int = 32,
        vocab_size: int = 1000,
        num_classes: int = 2,
        seed: int = 42,
    ) -> None:
        torch.manual_seed(seed)
        self.num_samples = num_samples
        self.seq_len = seq_len
        self.vocab_size = vocab_size
        self.num_classes = num_classes

        # FIX: Generate Zipfian base tokens
        self.input_ids = generate_zipfian_tokens(
            num_samples=num_samples,
            seq_len=seq_len,
            vocab_size=vocab_size,
            seed=seed,
        )
        self.attention_mask = torch.ones((num_samples, seq_len), dtype=torch.long)

        # Generate realistic balanced labels
        rng = np.random.default_rng(seed)
        self.labels = torch.from_numpy(rng.integers(0, num_classes, size=(num_samples,))).long()

        # Inject class-conditional salient tokens to simulate real sentiment / classification signal
        for i in range(num_samples):
            label_val = self.labels[i].item()
            salient_token = 10 + label_val * 5
            pos = rng.integers(1, seq_len - 1)
            self.input_ids[i, pos] = salient_token

    def __len__(self) -> int:
        return self.num_samples

    def __getitem__(self, idx: int) -> dict[str, torch.Tensor]:
        return {
            "input_ids": self.input_ids[idx],
            "attention_mask": self.attention_mask[idx],
            "labels": self.labels[idx],
        }

High-fidelity synthetic classification dataset using Zipfian token distributions and class-conditional semantic bias for realistic classifier benchmarking.

Ancestors

• torch.utils.data.dataset.Dataset
• typing.Generic

class SyntheticPerplexityDataset (num_samples: int = 128,
seq_len: int = 128,
vocab_size: int = 32000,
zipf_param: float = 1.15,
seed: int = 42)

Expand source code

class SyntheticPerplexityDataset(Dataset):
    """
    Synthetic Language Modeling Dataset using Zipfian power-law token distribution
    for authentic causal perplexity evaluation without empirical artifact distortion.
    """

    def __init__(
        self,
        num_samples: int = 128,
        seq_len: int = 128,
        vocab_size: int = 32000,
        zipf_param: float = 1.15,
        seed: int = 42,
    ) -> None:
        self.num_samples = num_samples
        self.seq_len = seq_len
        self.vocab_size = vocab_size

        # FIX: Generate Zipfian tokens to reflect natural language entropy
        self.input_ids = generate_zipfian_tokens(
            num_samples=num_samples,
            seq_len=seq_len,
            vocab_size=vocab_size,
            zipf_param=zipf_param,
            seed=seed,
        )
        self.attention_mask = torch.ones((num_samples, seq_len), dtype=torch.long)
        self.labels = self.input_ids.clone()

    def __len__(self) -> int:
        return self.num_samples

    def __getitem__(self, idx: int) -> dict[str, torch.Tensor]:
        return {
            "input_ids": self.input_ids[idx],
            "attention_mask": self.attention_mask[idx],
            "labels": self.labels[idx],
        }

Synthetic Language Modeling Dataset using Zipfian power-law token distribution for authentic causal perplexity evaluation without empirical artifact distortion.

Ancestors

• torch.utils.data.dataset.Dataset
• typing.Generic

class SyntheticLanguageModelingDataset (num_samples: int = 128,
seq_len: int = 128,
vocab_size: int = 32000,
zipf_param: float = 1.15,
seed: int = 42)

Expand source code

class SyntheticPerplexityDataset(Dataset):
    """
    Synthetic Language Modeling Dataset using Zipfian power-law token distribution
    for authentic causal perplexity evaluation without empirical artifact distortion.
    """

    def __init__(
        self,
        num_samples: int = 128,
        seq_len: int = 128,
        vocab_size: int = 32000,
        zipf_param: float = 1.15,
        seed: int = 42,
    ) -> None:
        self.num_samples = num_samples
        self.seq_len = seq_len
        self.vocab_size = vocab_size

        # FIX: Generate Zipfian tokens to reflect natural language entropy
        self.input_ids = generate_zipfian_tokens(
            num_samples=num_samples,
            seq_len=seq_len,
            vocab_size=vocab_size,
            zipf_param=zipf_param,
            seed=seed,
        )
        self.attention_mask = torch.ones((num_samples, seq_len), dtype=torch.long)
        self.labels = self.input_ids.clone()

    def __len__(self) -> int:
        return self.num_samples

    def __getitem__(self, idx: int) -> dict[str, torch.Tensor]:
        return {
            "input_ids": self.input_ids[idx],
            "attention_mask": self.attention_mask[idx],
            "labels": self.labels[idx],
        }

Synthetic Language Modeling Dataset using Zipfian power-law token distribution for authentic causal perplexity evaluation without empirical artifact distortion.

Ancestors

• torch.utils.data.dataset.Dataset
• typing.Generic