mixle.stats.latent.heterogeneous_pcfg module

Probabilistic context-free grammar with heterogeneous terminal emissions.

This module implements a small Chomsky-normal-form PCFG whose terminal rules emit observations through arbitrary mixle.stats distributions. A terminal distribution can be scalar-valued, tuple-valued, set-valued, sequence-valued, or any other sequence-encodable distribution; each terminal rule keeps its own encoder.

Data type: an observation is a finite sequence of terminal observations. The grammar supports binary rules A -> B C and terminal rules A -> emission. Epsilon productions and unary nonterminal productions are intentionally not modeled.

class HeterogeneousPCFGDistribution(binary_rules, terminal_rules, start=None, nonterminals=None, name=None)[source]

Bases: SequenceEncodableProbabilityDistribution

CNF PCFG whose terminal productions emit arbitrary distributions.

Parameters:

  • binary_rules (BinaryRuleInput | None) – Either {parent: [(left, right, prob), ...]} or a flat sequence of (parent, left, right, prob) rules.

  • terminal_rules (TerminalRuleInput) – Either {parent: [(emission_dist, prob), ...]} or a flat sequence of (parent, emission_dist, prob) rules.

  • start (Any | None) – Start nonterminal. If omitted, the first nonterminal is used.

  • nonterminals (Sequence[Any] | None) – Optional explicit nonterminal order.

  • name (str | None) – Optional model name.

Rule probabilities are normalized over all rules sharing a parent.

density(x)[source]

Return the probability density or mass at a single observation.

Parameters:

x (Sequence[Any])

Return type:

float

log_density(x)[source]

Return the log-density or log-mass at a single observation.

Parameters:

x (Sequence[Any])

Return type:

float

seq_log_density(x)[source]

Return vectorized log-density values for sequence-encoded observations.

Parameters:

x (tuple[ndarray, tuple[Any, ...]])

Return type:

ndarray

compute_capabilities()[source]

Engine readiness intersected from the terminal emission distributions.

The CKY inside dynamic program is expressed in ComputeEngine ops (see backend_seq_log_density), so the grammar is engine-ready on whatever engines all of its terminal emission leaves support (numpy plus, e.g., torch for autograd/GPU).

backend_seq_log_density(x, engine)[source]

Engine-routed CKY inside scoring (numpy + torch).

Terminal log-densities come from each emission leaf’s own engine backend score, and the inside dynamic program runs in ComputeEngine ops. Per-sequence charts are still a Python loop (variable-length parses), so this trades the tuned NumPy _inside for engine portability and differentiability rather than raw speed.

Parameters:

x (tuple[ndarray, tuple[Any, ...]])

Return type:

Any

to_fisher(**kwargs)[source]

Inside-outside Fisher view for the PCFG.

sampler(seed=None)[source]

Return a sampler for drawing observations from this distribution.

Parameters:

seed (int | None)

Return type:

HeterogeneousPCFGSampler

enumerator()[source]

Return an exact support enumerator for acyclic enumerable PCFGs.

Return type:

HeterogeneousPCFGEnumerator

quantized_index(max_bits, bin_width_bits=1.0)[source]

Build a bounded index with an inside-style DP over quantized derivation costs.

Terminal emission distributions must themselves support quantized_index. The DP counts grammar derivations by additive quantized bit cost and lazily unranks those derivations into emitted terminal sequences. For unambiguous grammars this is an index over distinct sequences; for ambiguous grammars the same sequence can be represented by multiple derivations, and the returned log probability is still the exact PCFG log_density of that sequence.

Parameters:
Return type:

LazyQuantizedEnumerationIndex

estimator(pseudo_count=None)[source]

Return an estimator for fitting this distribution from data.

Parameters:

pseudo_count (float | None)

Return type:

HeterogeneousPCFGEstimator

dist_to_encoder()[source]

Return the data encoder used by this distribution for vectorized methods.

Return type:

HeterogeneousPCFGDataEncoder

class HeterogeneousPCFGEnumerator(dist)[source]

Bases: DistributionEnumerator

Exact best-first enumerator for acyclic heterogeneous PCFGs.

Parameters:

dist (HeterogeneousPCFGDistribution)

class HeterogeneousPCFGSampler(dist, seed=None, max_depth=100, max_steps=10000)[source]

Bases: DistributionSampler

Sampler for HeterogeneousPCFGDistribution.

Parameters:

  • dist (HeterogeneousPCFGDistribution)

  • seed (int | None)

  • max_depth (int)

  • max_steps (int)

sample(size=None)[source]

Draw observations.

Combinator samplers (mixture/sequence/…) accept batched. With batched=True (the default) each child stream is drawn in one vectorized call instead of a per-draw Python loop – far faster. Because every child sampler owns an independent RandomState, batching consumes each stream in the same order as the loop, so the draws are identical to the legacy path. batched=False forces that legacy per-draw loop as a guaranteed- stable reference. Leaf samplers are already vectorized and ignore the flag.

Parameters:

size (int | None)

Return type:

list[Any] | list[list[Any]]

class HeterogeneousPCFGAccumulator(emission_accumulators, num_binary_rules, terminal_parents, binary_parents, keys=(None, None))[source]

Bases: SequenceEncodableStatisticAccumulator

Inside-outside sufficient-statistic accumulator.

Parameters:

  • emission_accumulators (Sequence[SequenceEncodableStatisticAccumulator])

  • num_binary_rules (int)

  • terminal_parents (Sequence[int])

  • binary_parents (Sequence[int])

  • keys (tuple[str | None, str | None] | None)

update(x, weight, estimate)[source]
Parameters:
Return type:

None

initialize(x, weight, rng)[source]
Parameters:
Return type:

None

seq_initialize(x, weights, rng)[source]
Parameters:
Return type:

None

seq_update(x, weights, estimate)[source]
Parameters:
Return type:

None

combine(suff_stat)[source]
Parameters:

suff_stat (tuple[ndarray, ndarray, Sequence[Any]])

Return type:

HeterogeneousPCFGAccumulator

value()[source]
Return type:

tuple[ndarray, ndarray, tuple[Any, …]]

from_value(x)[source]
Parameters:

x (tuple[ndarray, ndarray, Sequence[Any]])

Return type:

HeterogeneousPCFGAccumulator

key_merge(stats_dict)[source]

Pool this accumulator’s statistics into stats_dict under its merge key.

The structural default implements the common single-key pattern: store the accumulator under self.keys the first time the key is seen, else combine into the one already there. Accumulators with several named keys (e.g. an HMM’s init/trans/state keys) or a non-accumulator stats payload override this. A keys of None (the default) is a no-op.

Parameters:

stats_dict (dict[str, Any])

Return type:

None

key_replace(stats_dict)[source]

Replace this accumulator’s statistics from the pooled stats_dict entry (see key_merge).

Parameters:

stats_dict (dict[str, Any])

Return type:

None

acc_to_encoder()[source]
Return type:

HeterogeneousPCFGDataEncoder

class HeterogeneousPCFGAccumulatorFactory(factories, num_binary_rules, terminal_parents, binary_parents, keys=(None, None))[source]

Bases: StatisticAccumulatorFactory

Factory for HeterogeneousPCFGAccumulator.

Parameters:

  • factories (Sequence[StatisticAccumulatorFactory])

  • num_binary_rules (int)

  • terminal_parents (Sequence[int])

  • binary_parents (Sequence[int])

  • keys (tuple[str | None, str | None] | None)

make()[source]
Return type:

HeterogeneousPCFGAccumulator

class HeterogeneousPCFGEstimator(binary_rules, terminal_rules, start=None, nonterminals=None, pseudo_count=None, name=None, keys=(None, None))[source]

Bases: ParameterEstimator

Estimator for a fixed-topology heterogeneous PCFG.

Parameters:

  • binary_rules (BinaryRuleInput | None)

  • terminal_rules (TerminalRuleInput)

  • start (Any | None)

  • nonterminals (Sequence[Any] | None)

  • pseudo_count (float | None)

  • name (str | None)

  • keys (tuple[str | None, str | None] | None)

accumulator_factory()[source]
Return type:

HeterogeneousPCFGAccumulatorFactory

estimate(nobs, suff_stat)[source]
Parameters:
Return type:

HeterogeneousPCFGDistribution

class InducedHeterogeneousPCFGEstimator(max_nonterminals, terminal_estimators, start='S', nonterminal_prefix='NT', terminal_rule_mass=0.5, rule_pseudo_count=1.0e-3, prune_threshold=0.0, min_rule_prob=0.0, name=None, keys=(None, None))[source]

Bases: ParameterEstimator

Overcomplete sparse PCFG structure learner.

This estimator builds a finite grammar skeleton automatically:

  • K nonterminals named start, prefix1, ..., prefixK-1.

  • every binary rule A -> B C.

  • every terminal rule A -> terminal_family_j.

EM learns rule probabilities and terminal emission parameters. Rules whose expected count is below prune_threshold or whose normalized probability is below min_rule_prob are assigned probability zero, but the rule layout is kept stable so repeated calls to seq_estimate remain compatible.

Parameters:

  • max_nonterminals (int)

  • terminal_estimators (Sequence[ParameterEstimator])

  • start (Any)

  • nonterminal_prefix (str)

  • terminal_rule_mass (float)

  • rule_pseudo_count (float | None)

  • prune_threshold (float)

  • min_rule_prob (float)

  • name (str | None)

  • keys (tuple[str | None, str | None] | None)

accumulator_factory()[source]
Return type:

HeterogeneousPCFGAccumulatorFactory

initial_model(terminal_distributions, rng=None, jitter=0.0)[source]

Create an overcomplete starting grammar without hand-written rules.

terminal_distributions may contain one distribution per terminal family, in which case each nonterminal gets its own copy of the same family list, or one distribution per generated terminal rule.

Parameters:
Return type:

HeterogeneousPCFGDistribution

estimate(nobs, suff_stat)[source]
Parameters:
Return type:

HeterogeneousPCFGDistribution

class HeterogeneousPCFGDataEncoder(terminal_encoders)[source]

Bases: DataSequenceEncoder

Encode iid PCFG observations by flattening terminal observations.

Parameters:

terminal_encoders (Sequence[DataSequenceEncoder])

seq_encode(x)[source]

Encode the iid observation sequence x for vectorized evaluation.

Parameters:

x (Sequence[Sequence[Any]])

Return type:

tuple[ndarray, tuple[Any, …]]

class HeterogeneousPCFGFisherView(dist)[source]

Bases: FixedFisherView

Inside-outside Fisher view for heterogeneous PCFGs.

Coordinates are expected complete-data rule counts followed by terminal emission sufficient statistics gated by the posterior probability of the corresponding terminal rule at each token. For finite enumerable PCFGs, the model Fisher is the exact observed Fisher covariance of these posterior-expected complete-data statistics under the model distribution. Recursive or infinite-support grammars should use observed_fisher_* on data.

Parameters:

dist (Any)

fisher_information(stats=None, diagonal=False, ridge=1.0e-8, **kwargs)[source]

Empirical Fisher approximation from per-observation statistic vectors.

Parameters:
Return type:

ndarray

fisher_vectors(stats=None, metric='diagonal', center=None, fisher=None, ridge=1.0e-8, **kwargs)[source]

Return centered/whitened sufficient-statistic vectors.

metric=’identity’ returns centered statistics, ‘diagonal’ divides by per-coordinate Fisher standard deviations, and ‘full’ applies an empirical full-matrix whitening transform.

Parameters:
Return type:

ndarray