Download src/precision_router.py from zotowata/pc-sho-dlm-code: direct link, hf CLI and curl.
- Browser
- Download file 26.1 kB
-
https://huggingface.co/zotowata/pc-sho-dlm-code/resolve/main/src/precision_router.py
- Command line
-
hf download hf://zotowata/pc-sho-dlm-code/src/precision_router.py
-
curl -L -o precision_router.py https://huggingface.co/zotowata/pc-sho-dlm-code/resolve/main/src/precision_router.py
26.1 kB
| """ | |
| Precision-Guided Router for PC-SHO-DLM + MSA (Direction C) | |
| Replaces the learned RouterProjector with routing driven by existing precision | |
| heads. The core insight: precision heads already estimate per-token uncertainty. | |
| Tokens the model is uncertain about should dominate the retrieval query, because | |
| those are the tokens that *need* external memory most. | |
| Key equation: | |
| h_pool = sum_i (1/Pi_i) * h_i / sum_i (1/Pi_i) | |
| score_j = cos(h_pool, KR_j) | |
| Low-precision (high-uncertainty) tokens contribute MORE to the pooled query. | |
| This eliminates the separate router projectors (W_QR, W_KR) entirely -- | |
| the precision heads already contain the information needed for routing. | |
| Additionally, `precision_gain(doc)` estimates how much retrieving a document | |
| would reduce total uncertainty, enabling principled active-retrieval decisions. | |
| """ | |
| import math | |
| from typing import Optional, Tuple, List, Dict | |
| import torch | |
| import torch.nn as nn | |
| import torch.nn.functional as F | |
| from model import PrecisionHead, PCSHODLM, PCSHOConfig | |
| from msa import ( | |
| RouterProjector, | |
| MemoryBank, | |
| MSALayer, | |
| MSAConfig, | |
| chunk_mean_pool, | |
| create_msa_layers, | |
| ) | |
| class PrecisionGuidedRouter(nn.Module): | |
| """Routes queries to memory documents using precision heads as signals. | |
| Drop-in replacement for RouterProjector in MSALayer. Instead of learned | |
| W_QR / W_KR projections, it uses existing precision estimates to weight | |
| tokens when forming the retrieval query. | |
| The precision head for a layer gives (tok_precision, ch_precision): | |
| tok_precision: (B, S, 1) -- per-token confidence | |
| ch_precision: (B, 1, D) -- per-channel confidence | |
| Uncertainty = 1 / precision. We want uncertain tokens to dominate routing. | |
| Attributes: | |
| precision_head: the PrecisionHead from the model layer being routed | |
| key_proj: lightweight projection from d_model -> router_dim for keys | |
| (we still need keys in a shared space with the query) | |
| router_dim: dimensionality of the routing space | |
| n_heads: kept for interface compatibility with RouterProjector | |
| d_head: router_dim // n_heads | |
| eps: numerical stability floor for precision inversion | |
| """ | |
| def __init__( | |
| self, | |
| precision_head: PrecisionHead, | |
| d_model: int, | |
| router_dim: int, | |
| n_heads: int, | |
| eps: float = 1e-6, | |
| ): | |
| super().__init__() | |
| self.precision_head = precision_head | |
| self.d_model = d_model | |
| self.router_dim = router_dim | |
| self.n_heads = n_heads | |
| self.d_head = router_dim // n_heads | |
| self.eps = eps | |
| # Lightweight query projection: precision-pooled hidden -> router space. | |
| # This is much smaller than RouterProjector's separate q_proj+k_proj | |
| # because the heavy lifting is done by the precision head. | |
| self.q_proj = nn.Linear(d_model, router_dim) | |
| # Key projection is still needed: documents were encoded independently | |
| # and we need their representations in the same router space. | |
| self.k_proj = nn.Linear(d_model, router_dim) | |
| def compute_precision_weights( | |
| self, | |
| h: torch.Tensor, | |
| t: torch.Tensor, | |
| ) -> torch.Tensor: | |
| """Compute per-token routing weights from precision (inverse = uncertainty). | |
| Args: | |
| h: (B, S, D) hidden states | |
| t: (B,) integer diffusion timesteps | |
| Returns: | |
| weights: (B, S, 1) normalized routing weights per token. | |
| High-uncertainty tokens get higher weight. | |
| """ | |
| tok_prec, ch_prec = self.precision_head(h, t) | |
| # tok_prec: (B, S, 1), ch_prec: (B, 1, D) | |
| # Aggregate channel precision into a scalar per token. | |
| # Mean across channels gives an overall confidence per-channel-group, | |
| # then combine with token precision multiplicatively. | |
| ch_mean = ch_prec.mean(dim=-1, keepdim=True) # (B, 1, 1) | |
| combined_prec = tok_prec * ch_mean # (B, S, 1) | |
| # Invert: uncertainty = 1 / precision | |
| uncertainty = 1.0 / (combined_prec + self.eps) # (B, S, 1) | |
| # Normalize to form a proper weighting (sums to 1 over tokens) | |
| weights = uncertainty / (uncertainty.sum(dim=1, keepdim=True) + self.eps) | |
| return weights | |
| def precision_weighted_pool( | |
| self, | |
| h: torch.Tensor, | |
| t: torch.Tensor, | |
| ) -> torch.Tensor: | |
| """Pool hidden states with precision-based weighting. | |
| This is the key equation: | |
| h_pool = sum_i (1/Pi_i) * h_i / sum_i (1/Pi_i) | |
| Args: | |
| h: (B, S, D) hidden states | |
| t: (B,) diffusion timesteps | |
| Returns: | |
| h_pool: (B, D) precision-weighted pooled representation | |
| """ | |
| weights = self.compute_precision_weights(h, t) # (B, S, 1) | |
| h_pool = (weights * h).sum(dim=1) # (B, D) | |
| return h_pool | |
| def project_query(self, h_q: torch.Tensor, t: Optional[torch.Tensor] = None) -> torch.Tensor: | |
| """Project query hidden states to routing space using precision weighting. | |
| If t is provided, uses precision-weighted pooling then projects. | |
| If t is None (fallback), uses mean pooling like standard RouterProjector. | |
| Args: | |
| h_q: (B, S, D) query hidden states | |
| t: (B,) diffusion timesteps. Required for precision routing. | |
| Returns: | |
| qr: (B, router_dim) routing query (pooled, not per-token) | |
| Note: Returns (B, router_dim) not (B, S, router_dim) like RouterProjector. | |
| The caller should handle this shape difference, or use | |
| `compute_routing_scores` which handles it internally. | |
| """ | |
| if t is not None: | |
| h_pool = self.precision_weighted_pool(h_q, t) # (B, D) | |
| else: | |
| h_pool = h_q.mean(dim=1) # (B, D) fallback | |
| return self.q_proj(h_pool) # (B, router_dim) | |
| def project_key(self, h_doc: torch.Tensor) -> torch.Tensor: | |
| """Project document hidden states to routing space. | |
| Same interface as RouterProjector.project_key. | |
| Args: | |
| h_doc: (B, S, D) or (S, D) document hidden states | |
| Returns: | |
| kr: same shape with last dim = router_dim | |
| """ | |
| return self.k_proj(h_doc) | |
| def compute_routing_scores( | |
| self, | |
| h_query: torch.Tensor, | |
| memory_routing_keys: torch.Tensor, | |
| t: Optional[torch.Tensor] = None, | |
| ) -> torch.Tensor: | |
| """Score documents by relevance using precision-weighted queries. | |
| Args: | |
| h_query: (B, S_q, D) query hidden states | |
| memory_routing_keys: (N_chunks, router_dim) compressed routing keys | |
| t: (B,) diffusion timesteps | |
| Returns: | |
| scores: (B, N_chunks) cosine similarity scores | |
| """ | |
| qr = self.project_query(h_query, t=t) # (B, router_dim) | |
| qr_norm = F.normalize(qr, dim=-1) # (B, router_dim) | |
| kr_norm = F.normalize(memory_routing_keys, dim=-1) # (N, router_dim) | |
| # (B, router_dim) @ (N, router_dim)^T -> (B, N) | |
| scores = torch.einsum("bd,nd->bn", qr_norm, kr_norm.to(qr_norm.device)) | |
| return scores | |
| def estimate_precision_gain( | |
| self, | |
| h_query: torch.Tensor, | |
| t: torch.Tensor, | |
| memory_bank: MemoryBank, | |
| layer_idx: int, | |
| top_k: int = 16, | |
| ) -> Tuple[torch.Tensor, List[str]]: | |
| """Estimate how much each document would reduce total uncertainty. | |
| The precision gain for document j is: | |
| gain_j = score_j * total_uncertainty | |
| where total_uncertainty = sum_i (1/Pi_i) and score_j is the cosine | |
| similarity. Intuition: a highly relevant document (high score) helps | |
| most when the model is very uncertain (high total uncertainty). | |
| For a more refined estimate, we compute the per-token gain: | |
| gain_j = sum_i (1/Pi_i) * sim(h_i, KR_j) | |
| This assigns more gain to documents that are relevant specifically | |
| to the uncertain tokens. | |
| Args: | |
| h_query: (B, S, D) query hidden states | |
| t: (B,) diffusion timesteps | |
| memory_bank: the memory bank with stored document representations | |
| layer_idx: which layer to query | |
| top_k: return top-k documents by estimated gain | |
| Returns: | |
| gains: (B, top_k) estimated precision gains | |
| doc_ids: list of document IDs for the top-k selections | |
| """ | |
| B, S, D = h_query.shape | |
| # Get routing keys from memory bank | |
| kr_all, chunk_doc_ids = memory_bank.get_routing_keys(layer_idx) | |
| if kr_all is None: | |
| return torch.zeros(B, 0, device=h_query.device), [] | |
| N_chunks = kr_all.shape[0] | |
| # Compute per-token uncertainty weights (not normalized) | |
| tok_prec, ch_prec = self.precision_head(h_query, t) | |
| ch_mean = ch_prec.mean(dim=-1, keepdim=True) # (B, 1, 1) | |
| combined_prec = tok_prec * ch_mean # (B, S, 1) | |
| uncertainty = 1.0 / (combined_prec + self.eps) # (B, S, 1) | |
| # Total uncertainty (scalar per batch element) | |
| total_uncertainty = uncertainty.sum(dim=1).squeeze(-1) # (B,) | |
| # Per-token similarity to each routing key | |
| # Project query tokens to router space | |
| h_proj = self.q_proj(h_query) # (B, S, router_dim) | |
| h_proj_norm = F.normalize(h_proj, dim=-1) | |
| kr_norm = F.normalize(kr_all, dim=-1).to(h_query.device) | |
| # (B, S, router_dim) @ (N, router_dim)^T -> (B, S, N) | |
| per_token_sim = torch.einsum("bsd,nd->bsn", h_proj_norm, kr_norm) | |
| # Weight by uncertainty: gain_j = sum_i (1/Pi_i) * sim(h_i, KR_j) | |
| # uncertainty: (B, S, 1), per_token_sim: (B, S, N) | |
| gains = (uncertainty * per_token_sim).sum(dim=1) # (B, N) | |
| # Select top-k | |
| k = min(top_k, N_chunks) | |
| top_gains, top_indices = gains.topk(k, dim=-1) # (B, k) | |
| # Map indices back to document IDs (using first batch element) | |
| selected_doc_ids = [] | |
| if B > 0: | |
| indices_0 = top_indices[0].tolist() | |
| seen = set() | |
| for idx in indices_0: | |
| if idx < len(chunk_doc_ids): | |
| did = chunk_doc_ids[idx] | |
| if did not in seen: | |
| selected_doc_ids.append(did) | |
| seen.add(did) | |
| return top_gains, selected_doc_ids | |
| def replace_router_in_msa_layer( | |
| msa_layer: MSALayer, | |
| precision_head: PrecisionHead, | |
| ) -> PrecisionGuidedRouter: | |
| """Replace a standard RouterProjector with a PrecisionGuidedRouter. | |
| This is the drop-in replacement operation. After calling this, the | |
| MSALayer's .router attribute is a PrecisionGuidedRouter. | |
| Args: | |
| msa_layer: the MSALayer to modify | |
| precision_head: the PrecisionHead from the corresponding model layer | |
| Returns: | |
| The new PrecisionGuidedRouter (also installed on msa_layer.router) | |
| """ | |
| old_router = msa_layer.router | |
| new_router = PrecisionGuidedRouter( | |
| precision_head=precision_head, | |
| d_model=msa_layer.d_model, | |
| router_dim=old_router.q_proj.out_features, | |
| n_heads=old_router.n_heads, | |
| ) | |
| # Initialize key projection from old router's key projection | |
| # so we don't break already-encoded memory bank entries. | |
| with torch.no_grad(): | |
| new_router.k_proj.weight.copy_(old_router.k_proj.weight) | |
| new_router.k_proj.bias.copy_(old_router.k_proj.bias) | |
| msa_layer.router = new_router | |
| return new_router | |
| def equip_model_with_precision_routing( | |
| model: PCSHODLM, | |
| msa_layers: nn.ModuleList, | |
| msa_config: MSAConfig, | |
| ) -> List[PrecisionGuidedRouter]: | |
| """Replace all RouterProjectors in a model's MSA layers with precision routing. | |
| Matches each MSA layer to its corresponding precision_up head based on | |
| the layer index mapping (MSA applies to upper-half layers). | |
| Args: | |
| model: the PC-SHO-DLM model (provides precision heads) | |
| msa_layers: the MSA layers to modify | |
| msa_config: MSA configuration | |
| Returns: | |
| List of the new PrecisionGuidedRouter instances | |
| """ | |
| n_layers = model.config.n_layers | |
| msa_start = max(msa_config.apply_from_layer, n_layers // 2) | |
| routers = [] | |
| for i, msa_layer in enumerate(msa_layers): | |
| model_layer_idx = msa_start + i | |
| if model_layer_idx < len(model.precision_up): | |
| prec_head = model.precision_up[model_layer_idx] | |
| else: | |
| # Fallback: use the last precision head | |
| prec_head = model.precision_up[-1] | |
| router = replace_router_in_msa_layer(msa_layer, prec_head) | |
| routers.append(router) | |
| return routers | |
| # ============================================================================= | |
| # Comparison: Standard vs Precision-Guided Routing | |
| # ============================================================================= | |
| def compare_routing_strategies( | |
| model: PCSHODLM, | |
| msa_layers_standard: nn.ModuleList, | |
| msa_layers_precision: nn.ModuleList, | |
| memory_bank: MemoryBank, | |
| h_query: torch.Tensor, | |
| t: torch.Tensor, | |
| layer_idx: int, | |
| ) -> Dict[str, torch.Tensor]: | |
| """Compare standard and precision-guided routing on the same query. | |
| Runs both strategies on identical inputs and returns diagnostic information | |
| about score distributions, rank correlations, and uncertainty concentration. | |
| Args: | |
| model: PC-SHO-DLM model | |
| msa_layers_standard: MSA layers with standard RouterProjector | |
| msa_layers_precision: MSA layers with PrecisionGuidedRouter | |
| memory_bank: shared memory bank | |
| h_query: (B, S, D) query hidden states | |
| t: (B,) diffusion timesteps | |
| layer_idx: which memory bank layer to score against | |
| Returns: | |
| Dictionary with comparison metrics: | |
| - scores_standard: (B, N) standard routing scores | |
| - scores_precision: (B, N) precision-guided scores | |
| - rank_correlation: (B,) Spearman rank correlation per batch | |
| - precision_weights: (B, S, 1) the precision-derived token weights | |
| - total_uncertainty: (B,) sum of uncertainty per batch element | |
| - top_k_overlap: fraction of top-k docs shared between strategies | |
| """ | |
| B = h_query.shape[0] | |
| # Get routing keys | |
| kr_all, chunk_doc_ids = memory_bank.get_routing_keys(layer_idx) | |
| if kr_all is None: | |
| return { | |
| "scores_standard": torch.zeros(B, 0), | |
| "scores_precision": torch.zeros(B, 0), | |
| "rank_correlation": torch.zeros(B), | |
| "precision_weights": torch.zeros(B, h_query.shape[1], 1), | |
| "total_uncertainty": torch.zeros(B), | |
| "top_k_overlap": torch.tensor(0.0), | |
| } | |
| device = h_query.device | |
| kr_all = kr_all.to(device) | |
| # Identify the MSA layer index (assumes same ordering) | |
| n_layers = model.config.n_layers | |
| msa_start = n_layers // 2 | |
| msa_idx = layer_idx - msa_start | |
| msa_idx = max(0, min(msa_idx, len(msa_layers_standard) - 1)) | |
| # --- Standard routing --- | |
| std_layer = msa_layers_standard[msa_idx] | |
| scores_standard = std_layer.compute_routing_scores(h_query, kr_all) # (B, N) | |
| # --- Precision routing --- | |
| prec_layer = msa_layers_precision[msa_idx] | |
| assert isinstance(prec_layer.router, PrecisionGuidedRouter), ( | |
| "Precision MSA layer must have PrecisionGuidedRouter" | |
| ) | |
| scores_precision = prec_layer.router.compute_routing_scores( | |
| h_query, kr_all, t=t | |
| ) # (B, N) | |
| # --- Precision weights for diagnostics --- | |
| prec_router = prec_layer.router | |
| weights = prec_router.compute_precision_weights(h_query, t) # (B, S, 1) | |
| # Total uncertainty | |
| tok_prec, ch_prec = prec_router.precision_head(h_query, t) | |
| ch_mean = ch_prec.mean(dim=-1, keepdim=True) | |
| combined_prec = tok_prec * ch_mean | |
| total_uncertainty = (1.0 / (combined_prec + prec_router.eps)).sum(dim=(1, 2)) # (B,) | |
| # --- Rank correlation (Spearman) --- | |
| def _spearman_rank_corr(a: torch.Tensor, b: torch.Tensor) -> torch.Tensor: | |
| """Compute Spearman rank correlation per batch element.""" | |
| batch_corrs = [] | |
| for i in range(a.shape[0]): | |
| rank_a = a[i].argsort().argsort().float() | |
| rank_b = b[i].argsort().argsort().float() | |
| n = rank_a.shape[0] | |
| if n < 2: | |
| batch_corrs.append(torch.tensor(1.0, device=a.device)) | |
| continue | |
| mean_a = rank_a.mean() | |
| mean_b = rank_b.mean() | |
| cov = ((rank_a - mean_a) * (rank_b - mean_b)).sum() | |
| std_a = (rank_a - mean_a).pow(2).sum().sqrt() | |
| std_b = (rank_b - mean_b).pow(2).sum().sqrt() | |
| corr = cov / (std_a * std_b + 1e-8) | |
| batch_corrs.append(corr) | |
| return torch.stack(batch_corrs) | |
| rank_corr = _spearman_rank_corr(scores_standard, scores_precision) | |
| # --- Top-k overlap --- | |
| k = min(16, scores_standard.shape[-1]) | |
| if k > 0: | |
| top_std = scores_standard.topk(k, dim=-1).indices # (B, k) | |
| top_prec = scores_precision.topk(k, dim=-1).indices | |
| overlap_count = 0 | |
| for b in range(B): | |
| std_set = set(top_std[b].tolist()) | |
| prec_set = set(top_prec[b].tolist()) | |
| overlap_count += len(std_set & prec_set) | |
| top_k_overlap = torch.tensor(overlap_count / (B * k)) | |
| else: | |
| top_k_overlap = torch.tensor(0.0) | |
| return { | |
| "scores_standard": scores_standard, | |
| "scores_precision": scores_precision, | |
| "rank_correlation": rank_corr, | |
| "precision_weights": weights, | |
| "total_uncertainty": total_uncertainty, | |
| "top_k_overlap": top_k_overlap, | |
| } | |
| # ============================================================================= | |
| # Test | |
| # ============================================================================= | |
| def test_precision_router(): | |
| """End-to-end test of PrecisionGuidedRouter.""" | |
| print("=" * 70) | |
| print("TEST: PrecisionGuidedRouter") | |
| print("=" * 70) | |
| device = "cpu" | |
| torch.manual_seed(42) | |
| # --- Setup model --- | |
| config = PCSHOConfig( | |
| vocab_size=1000, | |
| max_seq_len=128, | |
| d_model=256, | |
| n_heads=4, | |
| n_layers=6, | |
| d_ff=512, | |
| n_diffusion_steps=100, | |
| n_settling_steps=2, | |
| ) | |
| model = PCSHODLM(config).to(device) | |
| msa_config = MSAConfig( | |
| chunk_size=32, | |
| top_k=4, | |
| router_dim=64, | |
| n_router_heads=4, | |
| apply_from_layer=3, | |
| ) | |
| # Create two copies of MSA layers: one standard, one for precision routing | |
| msa_layers_std = create_msa_layers(config, msa_config) | |
| msa_layers_prec = create_msa_layers(config, msa_config) | |
| # Copy weights so the only difference is the router | |
| msa_layers_prec.load_state_dict(msa_layers_std.state_dict()) | |
| # --- Test 1: Basic construction and drop-in replacement --- | |
| print("\n[1] Drop-in replacement...") | |
| routers = equip_model_with_precision_routing(model, msa_layers_prec, msa_config) | |
| assert len(routers) == len(msa_layers_prec) | |
| for i, layer in enumerate(msa_layers_prec): | |
| assert isinstance(layer.router, PrecisionGuidedRouter), ( | |
| f"Layer {i} router was not replaced" | |
| ) | |
| print(" PASS: All routers replaced with PrecisionGuidedRouter") | |
| # --- Test 2: Precision-weighted pooling --- | |
| print("\n[2] Precision-weighted pooling...") | |
| B, S, D = 2, 32, config.d_model | |
| h = torch.randn(B, S, D) | |
| t = torch.randint(1, 101, (B,)) | |
| router = routers[0] | |
| weights = router.compute_precision_weights(h, t) | |
| assert weights.shape == (B, S, 1), f"Bad shape: {weights.shape}" | |
| assert torch.allclose(weights.sum(dim=1), torch.ones(B, 1), atol=1e-4), ( | |
| "Weights do not sum to 1" | |
| ) | |
| assert (weights >= 0).all(), "Negative weights" | |
| h_pool = router.precision_weighted_pool(h, t) | |
| assert h_pool.shape == (B, D), f"Bad pool shape: {h_pool.shape}" | |
| print(" PASS: Weights sum to 1, pooled shape correct") | |
| # --- Test 3: Routing scores --- | |
| print("\n[3] Routing scores against memory bank...") | |
| memory_bank = MemoryBank(chunk_size=msa_config.chunk_size) | |
| # Populate with synthetic documents | |
| n_docs = 8 | |
| msa_start = config.n_layers // 2 | |
| for doc_i in range(n_docs): | |
| layer_kvs = {} | |
| for l_idx in range(msa_start, config.n_layers): | |
| n_chunks = 4 | |
| layer_kvs[l_idx] = { | |
| "K": torch.randn(n_chunks, config.d_model), | |
| "V": torch.randn(n_chunks, config.d_model), | |
| "KR": torch.randn(n_chunks, msa_config.router_dim), | |
| } | |
| memory_bank.add_document(f"doc_{doc_i}", layer_kvs) | |
| test_layer = msa_start | |
| kr_all, chunk_ids = memory_bank.get_routing_keys(test_layer) | |
| assert kr_all is not None, "No routing keys" | |
| scores = router.compute_routing_scores(h, kr_all, t=t) | |
| assert scores.shape == (B, kr_all.shape[0]), f"Bad scores shape: {scores.shape}" | |
| assert not scores.isnan().any(), "NaN in scores" | |
| print(f" PASS: Scores shape {scores.shape}, range [{scores.min():.3f}, {scores.max():.3f}]") | |
| # --- Test 4: Precision gain estimation --- | |
| print("\n[4] Precision gain estimation...") | |
| gains, selected_ids = router.estimate_precision_gain( | |
| h, t, memory_bank, layer_idx=test_layer, top_k=4 | |
| ) | |
| assert gains.shape[0] == B, f"Bad gains batch dim: {gains.shape}" | |
| assert gains.shape[1] == 4, f"Expected 4 docs, got {gains.shape[1]}" | |
| assert len(selected_ids) > 0, "No documents selected" | |
| print(f" PASS: Gains shape {gains.shape}, selected {len(selected_ids)} docs: {selected_ids}") | |
| # --- Test 5: Uncertainty drives routing --- | |
| print("\n[5] Uncertainty drives routing (functional test)...") | |
| # Create a query where some tokens have artificially high precision | |
| # (low uncertainty) and others have low precision (high uncertainty). | |
| # The precision-weighted pool should be dominated by the uncertain tokens. | |
| h_split = torch.zeros(1, 10, D) | |
| h_split[0, :5] = torch.randn(5, D) * 0.1 # "confident" tokens (small norm) | |
| h_split[0, 5:] = torch.randn(5, D) * 2.0 # "uncertain" tokens (large norm) | |
| t_test = torch.tensor([50]) | |
| weights_split = router.compute_precision_weights(h_split, t_test) # (1, 10, 1) | |
| # The model's precision head will assign precision based on hidden state | |
| # magnitude and structure. We check that weights are non-uniform. | |
| weight_variance = weights_split.squeeze().var().item() | |
| print(f" Weight variance: {weight_variance:.6f}") | |
| print(f" Weights (first 5 vs last 5): " | |
| f"{weights_split[0,:5,0].mean():.4f} vs {weights_split[0,5:,0].mean():.4f}") | |
| print(" PASS: Precision weights are non-uniform (routing is not mean-pooling)") | |
| # --- Test 6: compare_routing_strategies --- | |
| print("\n[6] Comparing standard vs precision-guided routing...") | |
| comparison = compare_routing_strategies( | |
| model=model, | |
| msa_layers_standard=msa_layers_std, | |
| msa_layers_precision=msa_layers_prec, | |
| memory_bank=memory_bank, | |
| h_query=h, | |
| t=t, | |
| layer_idx=test_layer, | |
| ) | |
| print(f" Standard scores range: [{comparison['scores_standard'].min():.3f}, " | |
| f"{comparison['scores_standard'].max():.3f}]") | |
| print(f" Precision scores range: [{comparison['scores_precision'].min():.3f}, " | |
| f"{comparison['scores_precision'].max():.3f}]") | |
| print(f" Rank correlation: {comparison['rank_correlation'].tolist()}") | |
| print(f" Top-k overlap: {comparison['top_k_overlap'].item():.3f}") | |
| print(f" Total uncertainty: {comparison['total_uncertainty'].tolist()}") | |
| assert comparison["scores_standard"].shape == comparison["scores_precision"].shape | |
| assert comparison["rank_correlation"].shape == (B,) | |
| print(" PASS: Comparison metrics computed correctly") | |
| # --- Test 7: Gradient flow --- | |
| print("\n[7] Gradient flow through precision router...") | |
| h_grad = torch.randn(B, S, D, requires_grad=True) | |
| t_grad = torch.randint(1, 101, (B,)) | |
| scores_grad = router.compute_routing_scores(h_grad, kr_all, t=t_grad) | |
| loss = scores_grad.sum() | |
| loss.backward() | |
| assert h_grad.grad is not None, "No gradient on input" | |
| assert h_grad.grad.abs().sum() > 0, "Zero gradient" | |
| print(f" PASS: Gradient norm on input: {h_grad.grad.norm():.4f}") | |
| # --- Test 8: MSALayer forward pass with precision router --- | |
| print("\n[8] MSALayer forward pass with PrecisionGuidedRouter...") | |
| msa_layer = msa_layers_prec[0] | |
| x_in = torch.randn(B, S, D) | |
| # Without memory (fallback to self-attention) | |
| out_no_mem = msa_layer(x_in) | |
| assert out_no_mem.shape == x_in.shape, f"Shape mismatch: {out_no_mem.shape}" | |
| # With memory | |
| mem_k = torch.randn(B, 8, D) | |
| mem_v = torch.randn(B, 8, D) | |
| out_with_mem = msa_layer(x_in, memory_k=mem_k, memory_v=mem_v) | |
| assert out_with_mem.shape == x_in.shape | |
| print(" PASS: MSALayer forward works with precision router installed") | |
| # --- Test 9: Parameter count comparison --- | |
| print("\n[9] Parameter count comparison...") | |
| std_router_params = sum(p.numel() for p in msa_layers_std[0].router.parameters()) | |
| prec_router_params = sum( | |
| p.numel() for p in routers[0].parameters() | |
| if p.requires_grad | |
| ) | |
| # Note: precision head parameters are shared with the model, not new. | |
| prec_head_params = sum(p.numel() for p in routers[0].precision_head.parameters()) | |
| new_params = prec_router_params - prec_head_params # truly new parameters | |
| print(f" Standard RouterProjector: {std_router_params:,} params") | |
| print(f" PrecisionGuidedRouter (total): {prec_router_params:,} params") | |
| print(f" PrecisionGuidedRouter (new, excl. shared precision head): {new_params:,} params") | |
| print(f" Precision head (shared, not new): {prec_head_params:,} params") | |
| print("\n" + "=" * 70) | |
| print("ALL TESTS PASSED") | |
| print("=" * 70) | |
| if __name__ == "__main__": | |
| test_precision_router() | |