#!/usr/bin/env python3 """ MLP Gate (gϕ) for EvoRM Two-Stage Inference Controller ======================================================== Paper: Section III-E, "MLP gating" Implements a lightweight MLP gate that decides whether ambiguous (Survival) pairs require full LLM inference. Architecture: Input: concat[entity_name_sim, rule_type_dist, conf_stats, topo_features] = 64 dims Hidden: 256 → 128 → 64 Output: gating probability ∈ [0,1] Training: Self-supervised - Positive: pairs where Stage 2 LLM returned match - Negative: pairs where Stage 2 LLM returned non-match - N_warmup = 500 trajectories before training """ import numpy as np import torch import torch.nn as nn import torch.optim as optim from typing import Dict, List, Tuple, Optional import os import json class MLPGate(nn.Module): """Lightweight MLP gate for Stage 1→Stage 2 routing.""" def __init__(self, input_dim: int = 64, hidden_dims: List[int] = None, theta_gate: float = 0.5, n_warmup: int = 500, device: str = 'cuda'): super().__init__() if hidden_dims is None: hidden_dims = [256, 128, 64] self.theta_gate = theta_gate self.n_warmup = n_warmup self.device = device # Build MLP layers layers = [] prev_dim = input_dim for h_dim in hidden_dims: layers.append(nn.Linear(prev_dim, h_dim)) layers.append(nn.ReLU()) layers.append(nn.Dropout(0.2)) prev_dim = h_dim layers.append(nn.Linear(prev_dim, 1)) layers.append(nn.Sigmoid()) self.net = nn.Sequential(*layers) self.to(device) # Training state self.is_trained = False self.samples_features = [] # list of np arrays self.samples_labels = [] # list of floats self.trajectory_count = 0 # Optimizer (created when training starts) self.optimizer = None self.criterion = nn.BCELoss() # Statistics self.total_predictions = 0 self.llm_skipped = 0 def extract_features(self, es_context: Dict, et_context: Dict, triggered_rules: List) -> np.ndarray: """Extract 64-dim feature vector from entity pair context and triggered rules.""" features = [] # (i) Entity name similarity features (16 dims) name1 = str(es_context.get('entity_name', '')).lower() name2 = str(et_context.get('entity_name', '')).lower() tokens1 = set(name1.split()) tokens2 = set(name2.split()) jaccard = len(tokens1 & tokens2) / max(1, len(tokens1 | tokens2)) chars1 = set(name1) chars2 = set(name2) char_jaccard = len(chars1 & chars2) / max(1, len(chars1 | chars2)) len_ratio = min(len(name1), len(name2)) / max(1, max(len(name1), len(name2))) len_diff = abs(len(name1) - len(name2)) / max(1, max(len(name1), len(name2))) name_feats = [jaccard, char_jaccard, len_ratio, len_diff, float(len(name1) > 0), float(len(name2) > 0), float(len(name1) > 10), float(len(name2) > 10)] while len(name_feats) < 16: name_feats.append(name_feats[len(name_feats) % 8] * 0.5) features.extend(name_feats[:16]) # (ii) Rule-type distribution profile (16 dims) atom_type_counts = {'SameValue': 0, 'DifferValue': 0, 'ShareNeighbor': 0, 'DifferNeighbor': 0, 'SemanticEquiv': 0, 'SemanticConflict': 0} total_atoms = 0 for rule in triggered_rules: for atom in rule.atoms: atype = getattr(atom, 'atom_type', 'Unknown') if atype in atom_type_counts: atom_type_counts[atype] += 1 total_atoms += 1 rule_feats = [] for atype in sorted(atom_type_counts.keys()): rule_feats.append(atom_type_counts[atype] / max(1, total_atoms)) n_match = sum(1 for r in triggered_rules if r.conclusion == 1) n_nonmatch = len(triggered_rules) - n_match rule_feats.extend([ n_match / max(1, len(triggered_rules)), n_nonmatch / max(1, len(triggered_rules)), float(len(triggered_rules)), min(1.0, len(triggered_rules) / 10.0), ]) while len(rule_feats) < 16: rule_feats.append(0.0) features.extend(rule_feats[:16]) # (iii) Max historical confidence + trigger stats (16 dims) conf_feats = [] if triggered_rules: confs = [r.conf for r in triggered_rules] triggers = [r.trigger_count for r in triggered_rules] conf_feats = [ max(confs), min(confs), sum(confs) / len(confs), float(np.std(confs)) if len(confs) > 1 else 0.0, max(triggers) / max(1, max(triggers)), min(triggers) / max(1, max(triggers)), sum(triggers) / max(1, sum(triggers) + len(triggers)), float(len(triggered_rules)), ] while len(conf_feats) < 16: conf_feats.append(0.0) features.extend(conf_feats[:16]) # (iv) Topological features (16 dims) topo_feats = [] for ctx in [es_context, et_context]: n_neighbor_keys = sum(1 for k in ctx if k.startswith('neighbors_')) total_n = sum(len(v) if isinstance(v, set) else 1 for k, v in ctx.items() if k.startswith('neighbors_')) topo_feats.append(float(n_neighbor_keys)) topo_feats.append(float(total_n) / max(1, total_n)) es_rels = set(k.replace('neighbors_', '') for k in es_context if k.startswith('neighbors_')) et_rels = set(k.replace('neighbors_', '') for k in et_context if k.startswith('neighbors_')) shared_rels = es_rels & et_rels topo_feats.extend([ float(len(shared_rels)), len(shared_rels) / max(1, len(es_rels | et_rels)), ]) while len(topo_feats) < 16: topo_feats.append(0.0) features.extend(topo_feats[:16]) return np.array(features, dtype=np.float32) def predict_proba(self, features: np.ndarray) -> float: """Predict gating probability.""" self.eval() with torch.no_grad(): x = torch.from_numpy(features).float().unsqueeze(0).to(self.device) prob = self.net(x).item() self.total_predictions += 1 return prob def should_invoke_llm(self, features: np.ndarray) -> bool: """Decide whether to invoke LLM based on gating probability.""" if not self.is_trained: return True # During warmup, always invoke LLM prob = self.predict_proba(features) if prob < self.theta_gate: self.llm_skipped += 1 return False return True def collect_sample(self, features: np.ndarray, label: float): """Collect a training sample during warmup phase.""" if len(self.samples_features) < self.n_warmup * 2: self.samples_features.append(features) self.samples_labels.append(label) self.trajectory_count += 1 if self.trajectory_count >= self.n_warmup and not self.is_trained: self.fit_model() def fit_model(self, epochs: int = 50, batch_size: int = 32, verbose: bool = True): """Self-supervised training on collected samples.""" if len(self.samples_features) < 10: if verbose: print(f"[MLPGate] Not enough samples ({len(self.samples_features)})") return # Set module to training mode super().train() X = torch.from_numpy(np.stack(self.samples_features)).float().to(self.device) y = torch.tensor(self.samples_labels).float().to(self.device) n_train = int(0.8 * len(self.samples_features)) indices = torch.randperm(len(self.samples_features)) X_train, y_train = X[indices[:n_train]], y[indices[:n_train]] X_val, y_val = X[indices[n_train:]], y[indices[n_train:]] if self.optimizer is None: self.optimizer = optim.Adam(self.parameters(), lr=1e-3, weight_decay=1e-5) for epoch in range(epochs): super().train() # training mode total_loss = 0.0 for i in range(0, len(X_train), batch_size): batch_X = X_train[i:i+batch_size] batch_y = y_train[i:i+batch_size] self.optimizer.zero_grad() pred = self.net(batch_X).squeeze() loss = self.criterion(pred, batch_y) loss.backward() self.optimizer.step() total_loss += loss.item() self.eval() with torch.no_grad(): val_pred = self.net(X_val).squeeze() val_loss = self.criterion(val_pred, y_val).item() val_acc = ((val_pred > 0.5) == y_val).float().mean().item() if verbose and epoch % 10 == 0: print(f"[MLPGate] Epoch {epoch}: loss={total_loss/max(1,len(X_train)):.4f}, " f"val_loss={val_loss:.4f}, val_acc={val_acc:.4f}") self.is_trained = True self.eval() if verbose: with torch.no_grad(): final_pred = self.net(X).squeeze() final_acc = ((final_pred > 0.5) == y).float().mean().item() n_pos = int((y == 1).sum().item()) n_neg = int((y == 0).sum().item()) print(f"[MLPGate] Trained: {len(self.samples_features)} samples, " f"acc={final_acc:.4f}, pos={n_pos}, neg={n_neg}") def save(self, path: str): """Save MLP gate state.""" state = { 'model_state': self.state_dict(), 'optimizer_state': self.optimizer.state_dict() if self.optimizer else None, 'is_trained': self.is_trained, 'trajectory_count': self.trajectory_count, 'n_samples': len(self.samples_features), 'theta_gate': self.theta_gate, 'n_warmup': self.n_warmup, 'total_predictions': self.total_predictions, 'llm_skipped': self.llm_skipped, } os.makedirs(os.path.dirname(path) if os.path.dirname(path) else '.', exist_ok=True) torch.save(state, path) print(f"[MLPGate] Saved to {path}") def load(self, path: str) -> bool: """Load MLP gate state.""" if not os.path.exists(path): return False state = torch.load(path, map_location=self.device) self.load_state_dict(state['model_state']) if state.get('optimizer_state'): if self.optimizer is None: self.optimizer = optim.Adam(self.parameters(), lr=1e-3) self.optimizer.load_state_dict(state['optimizer_state']) self.is_trained = state.get('is_trained', False) self.trajectory_count = state.get('trajectory_count', 0) self.theta_gate = state.get('theta_gate', 0.5) self.n_warmup = state.get('n_warmup', 500) self.total_predictions = state.get('total_predictions', 0) self.llm_skipped = state.get('llm_skipped', 0) print(f"[MLPGate] Loaded: trained={self.is_trained}, samples={state.get('n_samples', 0)}") return True def get_stats(self) -> Dict: return { 'is_trained': self.is_trained, 'trajectory_count': self.trajectory_count, 'n_samples': len(self.samples_features), 'total_predictions': self.total_predictions, 'llm_skipped': self.llm_skipped, 'skip_rate': self.llm_skipped / max(1, self.total_predictions), } # ============================================================================== # Test # ============================================================================== if __name__ == "__main__": print("MLP Gate - Self Test") print("=" * 60) gate = MLPGate(input_dim=64, n_warmup=20, device='cpu') print(f"Architecture:\n{gate.net}") print(f"Params: {sum(p.numel() for p in gate.parameters())}") es_ctx = {'entity_name': 'Test Entity A', 'neighbors_rel1': {'B', 'C'}} et_ctx = {'entity_name': 'Test Entity B', 'neighbors_rel1': {'C', 'D'}} class MockAtom: def __init__(self, atom_type, attr): self.atom_type = atom_type self.attr = attr class MockRule: def __init__(self, conclusion, conf, trigger_count): self.conclusion = conclusion self.conf = conf self.trigger_count = trigger_count self.atoms = [MockAtom('SameValue', 'name')] triggered = [MockRule(1, 0.8, 5), MockRule(0, 0.3, 2)] features = gate.extract_features(es_ctx, et_ctx, triggered) print(f"Features: shape={features.shape}, range=[{features.min():.3f}, {features.max():.3f}]") prob = gate.predict_proba(features) print(f"Prob (before training): {prob:.4f}") print(f"Should invoke: {gate.should_invoke_llm(features)}") print("\nCollecting samples...") for i in range(30): label = 1.0 if i < 15 else 0.0 gate.collect_sample(features + np.random.normal(0, 0.1, 64), label) stats = gate.get_stats() print(f"Stats: {stats}") prob2 = gate.predict_proba(features) print(f"Prob (after training): {prob2:.4f}") print(f"Should invoke: {gate.should_invoke_llm(features)}") print("\n✅ MLP Gate test complete!")