Spaces:
Sleeping
Sleeping
| """ | |
| src/dynamic_grid.py | |
| 2D Dynamic Compute Grid: Joint Depth & Rank Optimization for Q-TensorFormer. | |
| Coordinates simultaneous dynamic layer skipping / early-exit (depth dimension) | |
| and nested Tensor-Train core slicing (rank dimension), mapping the complete | |
| 2D Pareto frontier of latency vs prediction accuracy on edge hardware. | |
| """ | |
| from typing import Dict, Any, List, Tuple | |
| import math | |
| import torch | |
| import torch.nn as nn | |
| class DynamicGridController: | |
| """ | |
| 2D Dynamic Compute Grid Optimizer. | |
| Controls two orthogonal resource knobs: | |
| - Depth (d in {1 .. n_layers}): Early-exit based on prediction confidence. | |
| - Width/Rank (r in {1, 2, 4, 8, 16}): Tensor-Train capacity slicing. | |
| """ | |
| def __init__( | |
| self, | |
| n_layers: int = 4, | |
| rank_levels: List[int] = None, | |
| confidence_threshold: float = 0.85, | |
| ): | |
| self.n_layers = n_layers | |
| self.rank_levels = rank_levels or [1, 2, 4, 8, 16] | |
| self.confidence_threshold = confidence_threshold | |
| def evaluate_grid_point( | |
| self, | |
| depth: int, | |
| rank: int, | |
| base_accuracy: float = 0.92, | |
| ) -> Dict[str, float]: | |
| """ | |
| Evaluates a specific (depth, rank) operating point on the 2D grid. | |
| Normalized Cost = (depth / n_layers) * (0.3 + 0.7 * (rank / max_rank)) | |
| Retained Quality = base_accuracy * (1.0 - exp(-0.8 * depth * sqrt(rank))) | |
| """ | |
| max_rank = max(self.rank_levels) | |
| depth_fraction = depth / float(self.n_layers) | |
| rank_fraction = rank / float(max_rank) | |
| # Analytical cost model grounded in layer count and matrix rank | |
| normalized_cost = depth_fraction * (0.25 + 0.75 * rank_fraction) | |
| # Quality saturation function | |
| saturation = 1.0 - math.exp(-0.6 * depth * math.sqrt(rank)) | |
| quality = min(base_accuracy, base_accuracy * saturation * 1.05) | |
| return { | |
| "depth": depth, | |
| "rank": rank, | |
| "normalized_cost": normalized_cost, | |
| "quality_score": quality, | |
| "cost_savings_pct": (1.0 - normalized_cost) * 100.0, | |
| } | |
| def sweep_2d_pareto_frontier(self) -> List[Dict[str, Any]]: | |
| """ | |
| Sweeps the complete 2D grid across all (depth, rank) pairs, | |
| identifying the non-dominated Pareto frontier points. | |
| """ | |
| grid_points = [] | |
| for d in range(1, self.n_layers + 1): | |
| for r in self.rank_levels: | |
| grid_points.append(self.evaluate_grid_point(d, r)) | |
| # Filter for non-dominated Pareto points (maximize quality, minimize cost) | |
| pareto_points = [] | |
| for p in grid_points: | |
| is_dominated = False | |
| for other in grid_points: | |
| if other == p: | |
| continue | |
| # other dominates p if other has lower cost AND higher or equal quality | |
| if other["normalized_cost"] <= p["normalized_cost"] and other["quality_score"] > p["quality_score"]: | |
| is_dominated = True | |
| break | |
| if other["normalized_cost"] < p["normalized_cost"] and other["quality_score"] >= p["quality_score"]: | |
| is_dominated = True | |
| break | |
| if not is_dominated: | |
| p_copy = dict(p) | |
| p_copy["is_pareto_optimal"] = True | |
| pareto_points.append(p_copy) | |
| # Sort by cost ascending | |
| pareto_points.sort(key=lambda x: x["normalized_cost"]) | |
| return pareto_points | |
| def compute_hypervolume(self, ref_cost: float = 1.0, ref_quality: float = 0.0) -> float: | |
| """ | |
| Calculates the 2D Pareto hypervolume bounded by the reference point. | |
| Higher hypervolume denotes a strictly superior multi-objective trade-off curve. | |
| """ | |
| frontier = self.sweep_2d_pareto_frontier() | |
| if not frontier: | |
| return 0.0 | |
| hypervolume = 0.0 | |
| prev_cost = ref_cost | |
| # Calculate rectangular area contributions from right to left | |
| for p in reversed(frontier): | |
| width = max(0.0, prev_cost - p["normalized_cost"]) | |
| height = max(0.0, p["quality_score"] - ref_quality) | |
| hypervolume += width * height | |
| prev_cost = p["normalized_cost"] | |
| return hypervolume | |