| import torch |
| import torch.nn as nn |
| import torch.nn.functional as F |
|
|
| from config import DROPOUT |
|
|
|
|
| class HypothesisValidator(nn.Module): |
| """ |
| Scores hypothesis-vs-option compatibility and emits a guidance |
| potential for the reasoner. |
| |
| On the "Born rule" here |
| ----------------------- |
| The distribution computed below weights the guidance potential. With |
| real, non-negative amplitudes it is a temperature-scaled softmax and |
| nothing more -- that is stated rather than dressed up. The place |
| where amplitudes are genuinely complex, and where the Born rule |
| therefore does real work, is models/interference.py; this module |
| feeds it (through `validator_energy`) but does not itself perform a |
| measurement. |
| |
| Fixed in the v44 audit, kept |
| ---------------------------- |
| * Question conditioning. The question embedding was never computed |
| anywhere in the pipeline, so the validator judged options using |
| hypotheses alone -- a third of which were template fallbacks. |
| Measured with a plain MLP on cached ARC embeddings: options only |
| 0.3585, +question 0.4849. |
| * L2 (not L1) normalization of the amplitude. L1 is not the Born |
| convention and gave this module half the collapse controller's |
| effective temperature at the same nominal T. |
| * Padded options are excluded from every mean over N. |
| |
| New in v45 |
| ---------- |
| * `align` / `polarity` features, so the validator can distinguish |
| "the support for option n" from "the objection to option n" |
| instead of averaging them together. |
| """ |
|
|
| def __init__(self, dim, use_question=True): |
|
|
| super().__init__() |
|
|
| self.use_question = use_question |
|
|
| |
| |
| n_vec = 6 if use_question else 4 |
|
|
| n_feat = dim * n_vec + 3 |
|
|
| self.encoder = nn.Sequential( |
| nn.Linear(n_feat, dim * 2), |
| nn.GELU(), |
| nn.LayerNorm(dim * 2), |
| nn.Dropout(DROPOUT), |
| nn.Linear(dim * 2, dim), |
| nn.GELU(), |
| ) |
|
|
| self.causal = nn.Linear(dim, 1) |
| self.diversity = nn.Linear(dim, 1) |
| self.specificity = nn.Linear(dim, 1) |
| self.relevance = nn.Linear(dim, 1) |
|
|
| self.observable_gate = nn.Sequential( |
| nn.Linear(dim, dim), |
| nn.GELU(), |
| nn.Linear(dim, 4), |
| ) |
|
|
| self.reliability = nn.Sequential( |
| nn.Linear(dim, dim), |
| nn.GELU(), |
| nn.Linear(dim, 1), |
| nn.Sigmoid(), |
| ) |
|
|
| self.potential = nn.Sequential( |
| nn.Linear(dim + 1, dim), |
| nn.GELU(), |
| nn.Dropout(DROPOUT), |
| nn.Linear(dim, dim), |
| ) |
|
|
| self.temperature = nn.Parameter(torch.tensor(1.0)) |
|
|
| def forward(self, H, O, Q=None, y=None, H_mask=None, O_mask=None, |
| align=None, polarity=None): |
|
|
| B, K, D = H.shape |
| _, N, _ = O.shape |
|
|
| H = F.normalize(H, dim=-1) |
| O = F.normalize(O, dim=-1) |
|
|
| H_exp = H.unsqueeze(2).expand(B, K, N, D) |
| O_exp = O.unsqueeze(1).expand(B, K, N, D) |
|
|
| parts = [H_exp, O_exp, H_exp - O_exp, H_exp * O_exp] |
|
|
| if self.use_question: |
|
|
| if Q is None: |
| raise ValueError( |
| "HypothesisValidator was built with use_question=True " |
| "but forward() got Q=None. Rebuild the cache so it " |
| "carries question embeddings, or construct the model " |
| "with use_question=False." |
| ) |
|
|
| Q_exp = F.normalize(Q, dim=-1).view(B, 1, 1, D).expand(B, K, N, D) |
|
|
| parts += [Q_exp * O_exp, Q_exp - O_exp] |
|
|
| if align is None: |
| align = torch.zeros(B, K, N, device=H.device, dtype=H.dtype) |
|
|
| if polarity is None: |
| polarity = torch.zeros(B, K, device=H.device, dtype=H.dtype) |
|
|
| a = align.unsqueeze(-1) |
| p = polarity.view(B, K, 1, 1).expand(B, K, N, 1) |
|
|
| parts += [a, p, a * p] |
|
|
| features = torch.cat(parts, dim=-1) |
|
|
| z = self.encoder(features) |
|
|
| causal = self.causal(z).squeeze(-1) |
| diversity = self.diversity(z).squeeze(-1) |
| specificity = self.specificity(z).squeeze(-1) |
| relevance = self.relevance(z).squeeze(-1) |
|
|
| target = None |
|
|
| if y is not None: |
| target = torch.zeros_like(relevance) |
| target[torch.arange(B, device=H.device), :, y] = 1.0 |
|
|
| |
| |
| |
|
|
| if O_mask is not None: |
| om = O_mask.view(B, 1, N, 1).to(z.dtype) |
| denom = om.sum(dim=2).clamp(min=1.0) |
| z_mean = (z * om).sum(dim=2) / denom |
| om2 = O_mask.view(B, 1, N).to(causal.dtype) |
| d2 = om2.sum(dim=2).clamp(min=1.0) |
| mean = lambda t: (t * om2).sum(dim=2) / d2 |
| else: |
| z_mean = z.mean(dim=2) |
| mean = lambda t: t.mean(dim=2) |
|
|
| gate = self.observable_gate(z_mean) |
|
|
| weights = F.softmax(gate, dim=-1) |
|
|
| energy = ( |
| weights[..., 0] * mean(causal) |
| + weights[..., 1] * mean(diversity) |
| + weights[..., 2] * mean(specificity) |
| + weights[..., 3] * mean(relevance) |
| ) |
|
|
| energy = -energy |
|
|
| |
| |
| |
| |
|
|
| temperature = 0.5 + F.softplus(self.temperature) |
|
|
| log_amp = -energy / (2.0 * temperature) |
|
|
| if H_mask is not None: |
| log_amp = log_amp.masked_fill(~H_mask, float("-inf")) |
|
|
| log_amp = log_amp - log_amp.max(dim=1, keepdim=True).values |
|
|
| amplitude = torch.exp(log_amp) |
|
|
| amplitude = amplitude / torch.sqrt( |
| (amplitude ** 2).sum(dim=1, keepdim=True) + 1e-8 |
| ) |
|
|
| probabilities = amplitude.pow(2) |
|
|
| probabilities = probabilities / (probabilities.sum(dim=1, keepdim=True) + 1e-8) |
|
|
| reliability = self.reliability(z_mean).squeeze(-1) |
|
|
| potential_input = torch.cat([z_mean, reliability.unsqueeze(-1)], dim=-1) |
|
|
| potential = self.potential(potential_input) |
|
|
| potential = potential * probabilities.unsqueeze(-1) |
|
|
| return { |
| "potential": potential, |
| "validator_energy": energy, |
| "validator_probabilities": probabilities, |
| "reliability": reliability, |
| "observable_weights": weights, |
| "causal": mean(causal), |
| "diversity": mean(diversity), |
| "specificity": mean(specificity), |
| "relevance": mean(relevance), |
| "relevance_logits": relevance, |
| "relevance_target": target, |
| } |
|
|
