importar torch.nn como nn importar torch.nn.functional como F desde clases de datos importar clase de datos torch.manual_seed(0) @dataclass class Cfg: d_model: int = 192 n_head: int = 6 n_layer: int = 4 ffn_mult: int = 2 n_mod: int = 3 text_vocab:int = 16 vis_dim: int = 8 act_dim: int = 4 Lt: int = 8 Lv: int = 8 La: int = 6 cfg = Cfg() clase RMSNorm(nn.Module): def __init__(self, d, eps=1e-6): super().__init__(); self.w = nn.Parameter(antorcha.ones(d)); self.eps = eps def adelante(self, x): return self.w * x * torch.rsqrt(x.pow(2).mean(-1, keepdim=True) + self.eps) def build_rope(T, hd, dispositivo, base=10000.0): pos = torch.arange(T, dispositivo=dispositivo, dtype=torch.float32)[:, None]
idx = antorcha.arange(0, hd, 2, dispositivo=dispositivo, dtype=torch.float32)[None, :]
frecuencia = 1.0 / (base ** (idx / hd)) ang = pos * frecuencia cos = torch.cos(ang).repeat(1, 2)[None, None]
pecado = antorcha.sin(ang).repetir(1, 2)[None, None]
devolver cos, sin def rotar_half(x): hd = x.shape[-1]; x1, x2 = x[…, :hd // 2]x[…, hd // 2:]
devolver antorcha.cat([-x2, x1]-1) def apply_rope(q, k, cos, sin): return q * cos + rotar_half(q) * sin, k * cos + rotar_half(k) * sin clase Atención(nn.Module): “””Autoatención causal multimodal compartida con incrustaciones rotativas.””” def __init__(self, c: Cfg): super().__init__() self.H, self.hd = c.n_head, c.d_model // c.n_head self.qkv = nn.Linear(c.d_model, 3 * c.d_model, sesgo=False) self.proj = nn.Linear(c.d_model, c.d_model, sesgo=False) def adelante(self, x, cos, sin, máscara): B, T, D = x.shape q, k, v = self.qkv(x).chunk(3, -1) q = q.view(B, T, self.H, self.hd).transpose(1, 2) k = k.view(B, T, self.H, self.hd).transpose(1, 2) v = v.view(B, T, self.H, self.hd).transpose(1, 2) q, k = apply_rope(q, k, cos, sin) att = (q @ k.transpose(-2, -1)) / math.sqrt(self.hd) att = att.masked_fill(mask, float(“-inf”)).softmax(-1) o = (att @ v).transpose(1, 2).reshape(B, T, D) return self.proj(o) class Expert(nn.Module): “””Un avance SwiGLU por modalidad ‘experto en transformadores’.””” def __init__(self, d, mult): super().__init__(); h = d * mult self.w1 = nn.Linear(d, h, sesgo=False) self.w3 = nn.Linear(d, h, sesgo=False) self.w2 = nn.Linear(h, d, sesgo=False) def adelante(self, x): return self.w2(F.silu(self.w1(x)) * self.w3(x)) clase MoTBlock(nn.Module): “””Atención compartida + enrutamiento de mezcla de transformadores (experto por modalidad).””” def __init__(self, c: Cfg): super().__init__() self.attn_norm = RMSNorm(c.d_model) self.attn = Attention(c) self.ffn_norm = nn.ModuleList([RMSNorm(c.d_model) for _ in range(c.n_mod)]) self.expertos = nn.ModuleList([Expert(c.d_model, c.ffn_mult) for _ in range(c.n_mod)]) def adelante(self, x, cos, sin, máscara, mod_id): x = x + self.attn(self.attn_norm(x), cos, sin, máscara) out = torch.zeros_like(x) para i, exp in enumerate(self.experts): sel = (mod_id == i).view(1, -1, 1).to(x.dtype) out = out + sel * exp(self.ffn_norm[i](x)) devolver x + salida clase OmniMoT(nn.Module): def __init__(self, c: Cfg): super().__init__(); self.c = c self.text_emb = nn.Embedding(c.text_vocab, c.d_model) self.vis_in = nn.Linear(c.vis_dim, c.d_model) self.act_in = nn.Linear(c.act_dim, c.d_model) self.mod_emb = nn.Embedding(c.n_mod, c.d_model) self.blocks = nn.ModuleList([MoTBlock(c) for _ in range(c.n_layer)]) self.norm = RMSNorm(c.d_model) self.text_head = nn.Linear(c.d_model, c.text_vocab, sesgo=False) self.vis_head = nn.Linear(c.d_model, c.vis_dim, sesgo=False) self.act_head = nn.Linear(c.d_model, c.act_dim, sesgo=False) ids = antorcha.cat([torch.zeros(c.Lt), torch.ones(c.Lv), torch.full((c.La,), 2)]).long() self.register_buffer(“mod_id”, ids, persistente=False) def forward(self, text, vis, act): c = self.c x = torch.cat([self.text_emb(text), self.vis_in(vis), self.act_in(act)]1) x = x + self.mod_emb(self.mod_id)[None]
B, T, D = x.shape cos, sin = build_rope(T, D // c.n_head, x.device) máscara = torch.triu(torch.ones(T, T, dtype=torch.bool, dispositivo=x.device), 1)[None, None]
para negro en self.blocks: x = blk(x, cos, sin, máscara, self.mod_id) x = self.norm(x) ht = self.text_head(x[:, :c.Lt]) hv = self.vis_head(x[:, c.Lt:c.Lt + c.Lv]) ha = self.act_head(x[:, c.Lt + c.Lv:]) return ht, hv, ha model = OmniMoT(cfg).to(DEVICE) n_params = sum(p.numel() for p in model.parameters()) print(f”Modelo creado: OmniMoT | {n_params/1e6:.2f}M parámetros | {cfg.n_layer} MoT bloquea ” f”x {cfg.n_mod} expertos | dispositivo={DEVICE}”)
idx = antorcha.arange(0, hd, 2, dispositivo=dispositivo, dtype=torch.float32)[None, :]
frecuencia = 1.0 / (base ** (idx / hd)) ang = pos * frecuencia cos = torch.cos(ang).repeat(1, 2)[None, None]
pecado = antorcha.sin(ang).repetir(1, 2)[None, None]
devolver cos, sin def rotar_half(x): hd = x.shape[-1]; x1, x2 = x[…, :hd // 2]x[…, hd // 2:]
devolver antorcha.cat([-x2, x1]-1) def apply_rope(q, k, cos, sin): return q * cos + rotar_half(q) * sin, k * cos + rotar_half(k) * sin clase Atención(nn.Module): “””Autoatención causal multimodal compartida con incrustaciones rotativas.””” def __init__(self, c: Cfg): super().__init__() self.H, self.hd = c.n_head, c.d_model // c.n_head self.qkv = nn.Linear(c.d_model, 3 * c.d_model, sesgo=False) self.proj = nn.Linear(c.d_model, c.d_model, sesgo=False) def adelante(self, x, cos, sin, máscara): B, T, D = x.shape q, k, v = self.qkv(x).chunk(3, -1) q = q.view(B, T, self.H, self.hd).transpose(1, 2) k = k.view(B, T, self.H, self.hd).transpose(1, 2) v = v.view(B, T, self.H, self.hd).transpose(1, 2) q, k = apply_rope(q, k, cos, sin) att = (q @ k.transpose(-2, -1)) / math.sqrt(self.hd) att = att.masked_fill(mask, float(“-inf”)).softmax(-1) o = (att @ v).transpose(1, 2).reshape(B, T, D) return self.proj(o) class Expert(nn.Module): “””Un avance SwiGLU por modalidad ‘experto en transformadores’.””” def __init__(self, d, mult): super().__init__(); h = d * mult self.w1 = nn.Linear(d, h, sesgo=False) self.w3 = nn.Linear(d, h, sesgo=False) self.w2 = nn.Linear(h, d, sesgo=False) def adelante(self, x): return self.w2(F.silu(self.w1(x)) * self.w3(x)) clase MoTBlock(nn.Module): “””Atención compartida + enrutamiento de mezcla de transformadores (experto por modalidad).””” def __init__(self, c: Cfg): super().__init__() self.attn_norm = RMSNorm(c.d_model) self.attn = Attention(c) self.ffn_norm = nn.ModuleList([RMSNorm(c.d_model) for _ in range(c.n_mod)]) self.expertos = nn.ModuleList([Expert(c.d_model, c.ffn_mult) for _ in range(c.n_mod)]) def adelante(self, x, cos, sin, máscara, mod_id): x = x + self.attn(self.attn_norm(x), cos, sin, máscara) out = torch.zeros_like(x) para i, exp in enumerate(self.experts): sel = (mod_id == i).view(1, -1, 1).to(x.dtype) out = out + sel * exp(self.ffn_norm[i](x)) devolver x + salida clase OmniMoT(nn.Module): def __init__(self, c: Cfg): super().__init__(); self.c = c self.text_emb = nn.Embedding(c.text_vocab, c.d_model) self.vis_in = nn.Linear(c.vis_dim, c.d_model) self.act_in = nn.Linear(c.act_dim, c.d_model) self.mod_emb = nn.Embedding(c.n_mod, c.d_model) self.blocks = nn.ModuleList([MoTBlock(c) for _ in range(c.n_layer)]) self.norm = RMSNorm(c.d_model) self.text_head = nn.Linear(c.d_model, c.text_vocab, sesgo=False) self.vis_head = nn.Linear(c.d_model, c.vis_dim, sesgo=False) self.act_head = nn.Linear(c.d_model, c.act_dim, sesgo=False) ids = antorcha.cat([torch.zeros(c.Lt), torch.ones(c.Lv), torch.full((c.La,), 2)]).long() self.register_buffer(“mod_id”, ids, persistente=False) def forward(self, text, vis, act): c = self.c x = torch.cat([self.text_emb(text), self.vis_in(vis), self.act_in(act)]1) x = x + self.mod_emb(self.mod_id)[None]
B, T, D = x.shape cos, sin = build_rope(T, D // c.n_head, x.device) máscara = torch.triu(torch.ones(T, T, dtype=torch.bool, dispositivo=x.device), 1)[None, None]
para negro en self.blocks: x = blk(x, cos, sin, máscara, self.mod_id) x = self.norm(x) ht = self.text_head(x[:, :c.Lt]) hv = self.vis_head(x[:, c.Lt:c.Lt + c.Lv]) ha = self.act_head(x[:, c.Lt + c.Lv:]) return ht, hv, ha model = OmniMoT(cfg).to(DEVICE) n_params = sum(p.numel() for p in model.parameters()) print(f”Modelo creado: OmniMoT | {n_params/1e6:.2f}M parámetros | {cfg.n_layer} MoT bloquea ” f”x {cfg.n_mod} expertos | dispositivo={DEVICE}”)