import math import torch import torch.nn as nn def fourier_encode(positions: torch.Tensor, max_pos: int, num_frequencies: int = 16) -> torch.Tensor: """ Fourier positional encoding. positions: [B] integer positions (0..max_pos) returns: [B, 1+2*F] (normalized scalar + sin/cos features) """ # Normalize to [0, 1] (or [-1,1] style) # Use float32 pos_norm = positions.float() / float(max_pos) # [B] # pos_norm unsqueezed for broadcasting feats = [pos_norm.unsqueeze(-1)] # [B,1] for i in range(num_frequencies): freq = 2 ** i # scale frequency by pi to capture high frequency variation # sin(2^i * pi * pos_norm) feats.append(torch.sin(freq * math.pi * pos_norm).unsqueeze(-1)) feats.append(torch.cos(freq * math.pi * pos_norm).unsqueeze(-1)) return torch.cat(feats, dim=-1) # [B, 1+2F] class PiPredictor(nn.Module): def __init__( self, max_pos: int = 10000, num_frequencies: int = 16, hidden_dims: list = None, dropout: float = 0.1, encoding: str = "fourier", # "fourier" | "scalar" | "embedding" embedding_dim: int = 64, ): super().__init__() self.max_pos = max_pos self.num_frequencies = num_frequencies self.encoding = encoding if hidden_dims is None: hidden_dims = [256, 256, 128] if encoding == "fourier": input_dim = 1 + 2 * num_frequencies elif encoding == "scalar": input_dim = 1 elif encoding == "embedding": input_dim = embedding_dim self.embedding = nn.Embedding(max_pos + 5, embedding_dim) else: raise ValueError(f"unknown encoding {encoding}") layers = [] prev_dim = input_dim for h in hidden_dims: layers.append(nn.Linear(prev_dim, h)) layers.append(nn.LayerNorm(h)) layers.append(nn.GELU()) if dropout > 0: layers.append(nn.Dropout(dropout)) prev_dim = h self.backbone = nn.Sequential(*layers) self.head = nn.Linear(prev_dim, 10) # 10 digits # store for checkpoint reloading self.hidden_dims = hidden_dims self.dropout = dropout self.embedding_dim = embedding_dim def encode(self, positions: torch.Tensor) -> torch.Tensor: if self.encoding == "fourier": return fourier_encode(positions, self.max_pos, self.num_frequencies) elif self.encoding == "scalar": return (positions.float() / float(self.max_pos)).unsqueeze(-1) elif self.encoding == "embedding": return self.embedding(positions.long()) else: raise ValueError def forward(self, positions: torch.Tensor) -> torch.Tensor: """ positions: [B] long returns logits [B, 10] """ x = self.encode(positions) x = self.backbone(x) logits = self.head(x) return logits def predict(self, positions: torch.Tensor) -> torch.Tensor: logits = self.forward(positions) return torch.argmax(logits, dim=-1)