From 987ecf6386e505ee747642ce972f2c01fa25b4dc Mon Sep 17 00:00:00 2001 From: DicongLi <2024379585@qq.com> Date: Sun, 5 Jul 2026 01:07:26 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E5=90=8C=E6=AD=A5=20RAG=20=E8=9E=8D?= =?UTF-8?q?=E5=90=88=E4=B8=8E=E8=AE=AD=E7=BB=83=E6=A0=B8=E5=BF=83=E4=BB=A3?= =?UTF-8?q?=E7=A0=81(fusion=20=E6=94=AF=E6=8C=81=20f=5Fretr=E3=80=81models?= =?UTF-8?q?/trainer/moe/dataset)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- lnp_ml/modeling/layers/fusion.py | 306 +++++++++++---------- lnp_ml/modeling/layers/moe.py | 456 +++++++++++++++---------------- 2 files changed, 383 insertions(+), 379 deletions(-) diff --git a/lnp_ml/modeling/layers/fusion.py b/lnp_ml/modeling/layers/fusion.py index 70ef066..3824eb8 100644 --- a/lnp_ml/modeling/layers/fusion.py +++ b/lnp_ml/modeling/layers/fusion.py @@ -1,152 +1,156 @@ -import torch -import torch.nn as nn -import torch.nn.functional as F -from typing import Dict, List, Literal, Optional, Tuple, Union - - -PoolingStrategy = Literal["concat", "avg", "max", "attention"] - - -class FusionLayer(nn.Module): - """ - 将多个 token 融合成单个向量。 - - 输入: Dict[str, Tensor] 或 [B, n_tokens, d_model] - 输出: [B, fusion_dim] - - 策略: - - concat: [B, n_tokens, d_model] -> [B, n_tokens * d_model] - - avg: [B, n_tokens, d_model] -> [B, d_model] - - max: [B, n_tokens, d_model] -> [B, d_model] - - attention: [B, n_tokens, d_model] -> [B, d_model] (learnable attention pooling) - """ - - def __init__( - self, - d_model: int, - n_tokens: int, - strategy: PoolingStrategy = "attention", - ) -> None: - """ - Args: - d_model: 每个 token 的维度 - n_tokens: token 数量(如 8) - strategy: 融合策略 - """ - super().__init__() - self.d_model = d_model - self.n_tokens = n_tokens - self.strategy = strategy - - if strategy == "concat": - self.fusion_dim = n_tokens * d_model - else: - self.fusion_dim = d_model - - # Attention pooling: learnable query - if strategy == "attention": - self.attn_query = nn.Parameter(torch.randn(1, 1, d_model)) - self.attn_proj = nn.Linear(d_model, d_model) - - def forward( - self, - x: Union[Dict[str, torch.Tensor], torch.Tensor], - return_attn_weights: bool = False, - ) -> Union[torch.Tensor, Tuple[torch.Tensor, torch.Tensor]]: - """ - Args: - x: Dict[str, Tensor] 每个 [B, d_model],或已 stack 的 [B, n_tokens, d_model] - return_attn_weights: 若为 True 且策略为 attention,额外返回 attn_weights [B, n_tokens] - - Returns: - return_attn_weights=False: [B, fusion_dim] - return_attn_weights=True: ([B, fusion_dim], [B, n_tokens]) - """ - if isinstance(x, dict): - x = torch.stack(list(x.values()), dim=1) - - if self.strategy == "concat": - out = x.flatten(start_dim=1) - return (out, None) if return_attn_weights else out - - elif self.strategy == "avg": - out = x.mean(dim=1) - return (out, None) if return_attn_weights else out - - elif self.strategy == "max": - out = x.max(dim=1).values - return (out, None) if return_attn_weights else out - - elif self.strategy == "attention": - return self._attention_pooling(x, return_attn_weights) - - else: - raise ValueError(f"Unknown strategy: {self.strategy}") - - def _attention_pooling( - self, x: torch.Tensor, return_attn_weights: bool = False, - ) -> Union[torch.Tensor, Tuple[torch.Tensor, torch.Tensor]]: - """ - Attention pooling: 用可学习 query 对 tokens 做加权求和 - - Args: - x: [B, n_tokens, d_model] - return_attn_weights: 是否返回权重 - - Returns: - return_attn_weights=False: [B, d_model] - return_attn_weights=True: ([B, d_model], [B, n_tokens]) - """ - B = x.size(0) - query = self.attn_query.expand(B, -1, -1) - - keys = self.attn_proj(x) - scores = torch.bmm(query, keys.transpose(1, 2)) / (self.d_model ** 0.5) - attn_weights = F.softmax(scores, dim=-1) # [B, 1, n_tokens] - - out = torch.bmm(attn_weights, x).squeeze(1) # [B, d_model] - - if return_attn_weights: - return out, attn_weights.squeeze(1) # [B, n_tokens] - return out - - -class ResidualConcatFusion(nn.Module): - """对真实 token 做 attention pooling,再用零初始化门把 MoE/LLM 旁路以残差方式加入。 - - g_moe / g_llm 初始为 0 → +moe/+llm 起点严格等于 baseline; - 旁路只有确实有用时才会被训练打开,从机制上保证“加了不会更差”。 - """ - - def __init__(self, d_model: int, strategy: PoolingStrategy = "attention") -> None: - super().__init__() - if strategy == "concat": - raise ValueError("ResidualConcatFusion 不支持 concat(token 数随开关变化)") - self.d_model = d_model - self.pool = FusionLayer(d_model=d_model, n_tokens=1, strategy=strategy) - self.fusion_dim = self.pool.fusion_dim - # 零初始化门控(可学习标量),旁路初始不参与 - self.g_moe = nn.Parameter(torch.zeros(())) - self.g_llm = nn.Parameter(torch.zeros(())) - - def forward( - self, - chem: torch.Tensor, - tab: torch.Tensor, - f_moe: Optional[torch.Tensor] = None, - f_llm: Optional[torch.Tensor] = None, - return_attn_weights: bool = False, - ) -> Union[torch.Tensor, Tuple[torch.Tensor, torch.Tensor]]: - # 只对真实 token(chem + tab)做注意力池化,旁路不参与 softmax 竞争 - seq = torch.cat([chem, tab], dim=1) # [B, n_chem + n_cond, d_model] - pooled = self.pool(seq, return_attn_weights=return_attn_weights) - if return_attn_weights: - pooled, attn = pooled - - out = pooled - if f_moe is not None: - out = out + self.g_moe * f_moe # 残差 + 零初始化门 - if f_llm is not None: - out = out + self.g_llm * f_llm - +import torch +import torch.nn as nn +import torch.nn.functional as F +from typing import Dict, List, Literal, Optional, Tuple, Union + + +PoolingStrategy = Literal["concat", "avg", "max", "attention"] + + +class FusionLayer(nn.Module): + """ + 将多个 token 融合成单个向量。 + + 输入: Dict[str, Tensor] 或 [B, n_tokens, d_model] + 输出: [B, fusion_dim] + + 策略: + - concat: [B, n_tokens, d_model] -> [B, n_tokens * d_model] + - avg: [B, n_tokens, d_model] -> [B, d_model] + - max: [B, n_tokens, d_model] -> [B, d_model] + - attention: [B, n_tokens, d_model] -> [B, d_model] (learnable attention pooling) + """ + + def __init__( + self, + d_model: int, + n_tokens: int, + strategy: PoolingStrategy = "attention", + ) -> None: + """ + Args: + d_model: 每个 token 的维度 + n_tokens: token 数量(如 8) + strategy: 融合策略 + """ + super().__init__() + self.d_model = d_model + self.n_tokens = n_tokens + self.strategy = strategy + + if strategy == "concat": + self.fusion_dim = n_tokens * d_model + else: + self.fusion_dim = d_model + + # Attention pooling: learnable query + if strategy == "attention": + self.attn_query = nn.Parameter(torch.randn(1, 1, d_model)) + self.attn_proj = nn.Linear(d_model, d_model) + + def forward( + self, + x: Union[Dict[str, torch.Tensor], torch.Tensor], + return_attn_weights: bool = False, + ) -> Union[torch.Tensor, Tuple[torch.Tensor, torch.Tensor]]: + """ + Args: + x: Dict[str, Tensor] 每个 [B, d_model],或已 stack 的 [B, n_tokens, d_model] + return_attn_weights: 若为 True 且策略为 attention,额外返回 attn_weights [B, n_tokens] + + Returns: + return_attn_weights=False: [B, fusion_dim] + return_attn_weights=True: ([B, fusion_dim], [B, n_tokens]) + """ + if isinstance(x, dict): + x = torch.stack(list(x.values()), dim=1) + + if self.strategy == "concat": + out = x.flatten(start_dim=1) + return (out, None) if return_attn_weights else out + + elif self.strategy == "avg": + out = x.mean(dim=1) + return (out, None) if return_attn_weights else out + + elif self.strategy == "max": + out = x.max(dim=1).values + return (out, None) if return_attn_weights else out + + elif self.strategy == "attention": + return self._attention_pooling(x, return_attn_weights) + + else: + raise ValueError(f"Unknown strategy: {self.strategy}") + + def _attention_pooling( + self, x: torch.Tensor, return_attn_weights: bool = False, + ) -> Union[torch.Tensor, Tuple[torch.Tensor, torch.Tensor]]: + """ + Attention pooling: 用可学习 query 对 tokens 做加权求和 + + Args: + x: [B, n_tokens, d_model] + return_attn_weights: 是否返回权重 + + Returns: + return_attn_weights=False: [B, d_model] + return_attn_weights=True: ([B, d_model], [B, n_tokens]) + """ + B = x.size(0) + query = self.attn_query.expand(B, -1, -1) + + keys = self.attn_proj(x) + scores = torch.bmm(query, keys.transpose(1, 2)) / (self.d_model ** 0.5) + attn_weights = F.softmax(scores, dim=-1) # [B, 1, n_tokens] + + out = torch.bmm(attn_weights, x).squeeze(1) # [B, d_model] + + if return_attn_weights: + return out, attn_weights.squeeze(1) # [B, n_tokens] + return out + + +class ResidualConcatFusion(nn.Module): + """对真实 token 做 attention pooling,再用零初始化门把 MoE/LLM 旁路以残差方式加入。 + + g_moe / g_llm 初始为 0 → +moe/+llm 起点严格等于 baseline; + 旁路只有确实有用时才会被训练打开,从机制上保证“加了不会更差”。 + """ + + def __init__(self, d_model: int, strategy: PoolingStrategy = "attention") -> None: + super().__init__() + if strategy == "concat": + raise ValueError("ResidualConcatFusion 不支持 concat(token 数随开关变化)") + self.d_model = d_model + self.pool = FusionLayer(d_model=d_model, n_tokens=1, strategy=strategy) + self.fusion_dim = self.pool.fusion_dim + # 零初始化门控(可学习标量),旁路初始不参与 + self.g_moe = nn.Parameter(torch.zeros(())) + self.g_llm = nn.Parameter(torch.zeros(())) + self.g_retr = nn.Parameter(torch.zeros(())) # 检索旁路零初始化门控 + + def forward( + self, + chem: torch.Tensor, + tab: torch.Tensor, + f_moe: Optional[torch.Tensor] = None, + f_llm: Optional[torch.Tensor] = None, + f_retr: Optional[torch.Tensor] = None, + return_attn_weights: bool = False, + ) -> Union[torch.Tensor, Tuple[torch.Tensor, torch.Tensor]]: + # 只对真实 token(chem + tab)做注意力池化,旁路不参与 softmax 竞争 + seq = torch.cat([chem, tab], dim=1) # [B, n_chem + n_cond, d_model] + pooled = self.pool(seq, return_attn_weights=return_attn_weights) + if return_attn_weights: + pooled, attn = pooled + + out = pooled + if f_moe is not None: + out = out + self.g_moe * f_moe # 残差 + 零初始化门 + if f_llm is not None: + out = out + self.g_llm * f_llm + if f_retr is not None: + out = out + self.g_retr * f_retr # 检索旁路,零初始化门保证起点=不开 + return (out, attn) if return_attn_weights else out \ No newline at end of file diff --git a/lnp_ml/modeling/layers/moe.py b/lnp_ml/modeling/layers/moe.py index 8a07c05..aab389d 100644 --- a/lnp_ml/modeling/layers/moe.py +++ b/lnp_ml/modeling/layers/moe.py @@ -1,229 +1,229 @@ -""" -MoE (Mixture-of-Experts) 模块:sample-level + 跨模态路由。 - -设计要点(与现有 8-token 架构对齐): - - Router 输入:tab token 池化后的 [B, d](即配方/实验条件向量)。 - - Expert 输入:chem token flatten 后的 [B, T_chem * d](化学侧 token)。 - - 路由粒度:每个样本只算一次 router → gates [B, K]。 - - Top-k 稀疏激活 + 可选训练态 jitter 噪声。 - - 返回 load-balancing aux loss,由 trainer 端按权重加进总 loss。 - -输出: - F_moe: [B, d_model] 与单个 token 同维度,便于追加到 fusion 序列中。 - extras: dict 诊断与监控信息(aux loss、gates 等)。 -""" - -from typing import Dict, Tuple - -import torch -import torch.nn as nn -import torch.nn.functional as F - - -class MoEAttentionPool(nn.Module): - """与 FusionLayer 同款的 attention pooling,参数独立。 - - 将一组 token [B, T, d] 池化成单个查询向量 [B, d],用作 router 的输入。 - """ - - def __init__(self, d_model: int) -> None: - super().__init__() - self.d_model = d_model - self.query = nn.Parameter(torch.randn(1, 1, d_model)) - self.proj = nn.Linear(d_model, d_model) - - def forward(self, x: torch.Tensor) -> torch.Tensor: - """ - Args: - x: [B, T, d_model] - - Returns: - [B, d_model] - """ - B = x.size(0) - q = self.query.expand(B, -1, -1) # [B, 1, d] - k = self.proj(x) # [B, T, d] - scores = torch.bmm(q, k.transpose(1, 2)) / (self.d_model ** 0.5) - weights = F.softmax(scores, dim=-1) # [B, 1, T] - return torch.bmm(weights, x).squeeze(1) # [B, d] - - -class MoERouter(nn.Module): - """单层 softmax router + Top-k 稀疏激活。 - - Args: - d_model: router 输入维度。 - n_experts: 专家数量 K。 - top_k: 每个样本激活的专家数。 - jitter_noise: 训练态加在 logits 上的均匀噪声幅度,用于探索;评估态自动关闭。 - """ - - def __init__( - self, - d_model: int, - n_experts: int, - top_k: int = 2, - jitter_noise: float = 0.0, - ) -> None: - super().__init__() - if not 1 <= top_k <= n_experts: - raise ValueError(f"top_k 必须在 [1, n_experts={n_experts}],收到 {top_k}") - self.n_experts = n_experts - self.top_k = top_k - self.jitter_noise = jitter_noise - self.linear = nn.Linear(d_model, n_experts) - - def forward( - self, q: torch.Tensor, - ) -> Tuple[torch.Tensor, torch.Tensor, torch.Tensor]: - """ - Args: - q: [B, d_model] 路由查询向量。 - - Returns: - gates: [B, n_experts] top-k 后再归一化的稀疏概率。 - probs_full: [B, n_experts] 未掩码的 softmax 概率(用于 aux loss)。 - expert_mask: [B, n_experts] 0/1 掩码,标记被激活的专家。 - """ - logits = self.linear(q) # [B, K] - if self.training and self.jitter_noise > 0.0: - noise = (torch.rand_like(logits) - 0.5) * self.jitter_noise - logits = logits + noise - - probs_full = F.softmax(logits, dim=-1) # [B, K] - - topk_vals, topk_idx = probs_full.topk(self.top_k, dim=-1) # [B, k] - expert_mask = torch.zeros_like(probs_full) - expert_mask.scatter_(1, topk_idx, 1.0) - - gates = probs_full * expert_mask - gates = gates / gates.sum(dim=-1, keepdim=True).clamp(min=1e-9) - return gates, probs_full, expert_mask - - -class MoEExpert(nn.Module): - """单个专家 MLP:in_dim → hidden_dim → out_dim。""" - - def __init__( - self, in_dim: int, hidden_dim: int, out_dim: int, dropout: float = 0.1, - ) -> None: - super().__init__() - self.net = nn.Sequential( - nn.Linear(in_dim, hidden_dim), - nn.GELU(), - nn.Dropout(dropout), - nn.Linear(hidden_dim, out_dim), - ) - - def forward(self, x: torch.Tensor) -> torch.Tensor: - return self.net(x) - - -class MoEBlock(nn.Module): - """ - Sample-level 跨模态 MoE。 - - 流程: - 1. tab tokens [B, T_tab, d] -> attention pool -> q_tab [B, d] - 2. q_tab -> Router -> gates [B, K] (+ probs_full / expert_mask) - 3. chem tokens [B, T_chem, d] -> flatten -> [B, T_chem * d] - 4. K 个 Expert MLP 并行处理 -> stack [B, K, d] - 5. 加权求和 -> F_moe [B, d] - 6. 计算 load-balancing aux loss - - Args: - d_model: token 维度。 - n_chem_tokens: chem 侧 token 数(决定 expert 输入维度)。 - n_experts: 专家数量 K。 - top_k: 每个样本激活的专家数。 - expert_hidden_mult: expert 中间层维度 = expert_hidden_mult * d_model。 - dropout: expert 内部 dropout。 - jitter_noise: router 训练态噪声幅度,0.0 表示关闭。 - """ - - def __init__( - self, - d_model: int, - n_chem_tokens: int = 4, - n_experts: int = 4, - top_k: int = 2, - expert_hidden_mult: int = 2, - dropout: float = 0.1, - jitter_noise: float = 0.0, - ) -> None: - super().__init__() - self.d_model = d_model - self.n_chem_tokens = n_chem_tokens - self.n_experts = n_experts - self.top_k = top_k - - self.tab_pool = MoEAttentionPool(d_model) - self.router = MoERouter( - d_model, n_experts, top_k=top_k, jitter_noise=jitter_noise, - ) - - expert_in = d_model * n_chem_tokens - expert_hidden = d_model * expert_hidden_mult - self.experts = nn.ModuleList([ - MoEExpert(expert_in, expert_hidden, d_model, dropout=dropout) - for _ in range(n_experts) - ]) - - def forward( - self, - chem: torch.Tensor, - tab: torch.Tensor, - ) -> Tuple[torch.Tensor, Dict[str, torch.Tensor]]: - """ - Args: - chem: [B, T_chem, d_model] 化学侧 token(router 不看)。 - tab: [B, T_tab, d_model] 配方/实验侧 token(router 看)。 - - Returns: - F_moe: [B, d_model] - extras: { - "lb_loss": 标量 load-balancing aux loss(带梯度), - "gates": [B, K] 稀疏归一后的门控(detached), - "probs": [B, K] 原始 softmax 概率(detached), - } - """ - if chem.size(1) != self.n_chem_tokens: - raise ValueError( - f"chem token 数不匹配:期望 {self.n_chem_tokens}, " - f"实际 {chem.size(1)}" - ) - - q_tab = self.tab_pool(tab) # [B, d] - gates, probs_full, expert_mask = self.router(q_tab) # 各 [B, K] - - flat = chem.flatten(start_dim=1) # [B, T_chem * d] - expert_outs = torch.stack( - [expert(flat) for expert in self.experts], dim=1, - ) # [B, K, d] - - F_moe = (gates.unsqueeze(-1) * expert_outs).sum(dim=1) # [B, d] - - lb_loss = self._load_balancing_loss(probs_full, expert_mask) - - return F_moe, { - "lb_loss": lb_loss, - "gates": gates.detach(), - "probs": probs_full.detach(), - } - - def _load_balancing_loss( - self, - probs_full: torch.Tensor, - expert_mask: torch.Tensor, - ) -> torch.Tensor: - """Switch Transformer 风格的 load-balancing loss。 - - f_i = 该 batch 中 expert_i 被激活的样本占比(top-k mask 的均值) - p_i = 该 batch 中 expert_i 的 softmax 概率均值 - loss = K * Σ_i f_i * p_i - - 理想情况下 f_i 与 p_i 都接近 1/K,loss ≈ 1。 - """ - f = expert_mask.mean(dim=0) # [K] - p = probs_full.mean(dim=0) # [K] +""" +MoE (Mixture-of-Experts) 模块:sample-level + 跨模态路由。 + +设计要点(与现有 8-token 架构对齐): + - Router 输入:tab token 池化后的 [B, d](即配方/实验条件向量)。 + - Expert 输入:chem token flatten 后的 [B, T_chem * d](化学侧 token)。 + - 路由粒度:每个样本只算一次 router → gates [B, K]。 + - Top-k 稀疏激活 + 可选训练态 jitter 噪声。 + - 返回 load-balancing aux loss,由 trainer 端按权重加进总 loss。 + +输出: + F_moe: [B, d_model] 与单个 token 同维度,便于追加到 fusion 序列中。 + extras: dict 诊断与监控信息(aux loss、gates 等)。 +""" + +from typing import Dict, Tuple + +import torch +import torch.nn as nn +import torch.nn.functional as F + + +class MoEAttentionPool(nn.Module): + """与 FusionLayer 同款的 attention pooling,参数独立。 + + 将一组 token [B, T, d] 池化成单个查询向量 [B, d],用作 router 的输入。 + """ + + def __init__(self, d_model: int) -> None: + super().__init__() + self.d_model = d_model + self.query = nn.Parameter(torch.randn(1, 1, d_model)) + self.proj = nn.Linear(d_model, d_model) + + def forward(self, x: torch.Tensor) -> torch.Tensor: + """ + Args: + x: [B, T, d_model] + + Returns: + [B, d_model] + """ + B = x.size(0) + q = self.query.expand(B, -1, -1) # [B, 1, d] + k = self.proj(x) # [B, T, d] + scores = torch.bmm(q, k.transpose(1, 2)) / (self.d_model ** 0.5) + weights = F.softmax(scores, dim=-1) # [B, 1, T] + return torch.bmm(weights, x).squeeze(1) # [B, d] + + +class MoERouter(nn.Module): + """单层 softmax router + Top-k 稀疏激活。 + + Args: + d_model: router 输入维度。 + n_experts: 专家数量 K。 + top_k: 每个样本激活的专家数。 + jitter_noise: 训练态加在 logits 上的均匀噪声幅度,用于探索;评估态自动关闭。 + """ + + def __init__( + self, + d_model: int, + n_experts: int, + top_k: int = 2, + jitter_noise: float = 0.0, + ) -> None: + super().__init__() + if not 1 <= top_k <= n_experts: + raise ValueError(f"top_k 必须在 [1, n_experts={n_experts}],收到 {top_k}") + self.n_experts = n_experts + self.top_k = top_k + self.jitter_noise = jitter_noise + self.linear = nn.Linear(d_model, n_experts) + + def forward( + self, q: torch.Tensor, + ) -> Tuple[torch.Tensor, torch.Tensor, torch.Tensor]: + """ + Args: + q: [B, d_model] 路由查询向量。 + + Returns: + gates: [B, n_experts] top-k 后再归一化的稀疏概率。 + probs_full: [B, n_experts] 未掩码的 softmax 概率(用于 aux loss)。 + expert_mask: [B, n_experts] 0/1 掩码,标记被激活的专家。 + """ + logits = self.linear(q) # [B, K] + if self.training and self.jitter_noise > 0.0: + noise = (torch.rand_like(logits) - 0.5) * self.jitter_noise + logits = logits + noise + + probs_full = F.softmax(logits, dim=-1) # [B, K] + + topk_vals, topk_idx = probs_full.topk(self.top_k, dim=-1) # [B, k] + expert_mask = torch.zeros_like(probs_full) + expert_mask.scatter_(1, topk_idx, 1.0) + + gates = probs_full * expert_mask + gates = gates / gates.sum(dim=-1, keepdim=True).clamp(min=1e-9) + return gates, probs_full, expert_mask + + +class MoEExpert(nn.Module): + """单个专家 MLP:in_dim → hidden_dim → out_dim。""" + + def __init__( + self, in_dim: int, hidden_dim: int, out_dim: int, dropout: float = 0.1, + ) -> None: + super().__init__() + self.net = nn.Sequential( + nn.Linear(in_dim, hidden_dim), + nn.GELU(), + nn.Dropout(dropout), + nn.Linear(hidden_dim, out_dim), + ) + + def forward(self, x: torch.Tensor) -> torch.Tensor: + return self.net(x) + + +class MoEBlock(nn.Module): + """ + Sample-level 跨模态 MoE。 + + 流程: + 1. tab tokens [B, T_tab, d] -> attention pool -> q_tab [B, d] + 2. q_tab -> Router -> gates [B, K] (+ probs_full / expert_mask) + 3. chem tokens [B, T_chem, d] -> flatten -> [B, T_chem * d] + 4. K 个 Expert MLP 并行处理 -> stack [B, K, d] + 5. 加权求和 -> F_moe [B, d] + 6. 计算 load-balancing aux loss + + Args: + d_model: token 维度。 + n_chem_tokens: chem 侧 token 数(决定 expert 输入维度)。 + n_experts: 专家数量 K。 + top_k: 每个样本激活的专家数。 + expert_hidden_mult: expert 中间层维度 = expert_hidden_mult * d_model。 + dropout: expert 内部 dropout。 + jitter_noise: router 训练态噪声幅度,0.0 表示关闭。 + """ + + def __init__( + self, + d_model: int, + n_chem_tokens: int = 4, + n_experts: int = 4, + top_k: int = 2, + expert_hidden_mult: int = 2, + dropout: float = 0.1, + jitter_noise: float = 0.0, + ) -> None: + super().__init__() + self.d_model = d_model + self.n_chem_tokens = n_chem_tokens + self.n_experts = n_experts + self.top_k = top_k + + self.tab_pool = MoEAttentionPool(d_model) + self.router = MoERouter( + d_model, n_experts, top_k=top_k, jitter_noise=jitter_noise, + ) + + expert_in = d_model * n_chem_tokens + expert_hidden = d_model * expert_hidden_mult + self.experts = nn.ModuleList([ + MoEExpert(expert_in, expert_hidden, d_model, dropout=dropout) + for _ in range(n_experts) + ]) + + def forward( + self, + chem: torch.Tensor, + tab: torch.Tensor, + ) -> Tuple[torch.Tensor, Dict[str, torch.Tensor]]: + """ + Args: + chem: [B, T_chem, d_model] 化学侧 token(router 不看)。 + tab: [B, T_tab, d_model] 配方/实验侧 token(router 看)。 + + Returns: + F_moe: [B, d_model] + extras: { + "lb_loss": 标量 load-balancing aux loss(带梯度), + "gates": [B, K] 稀疏归一后的门控(detached), + "probs": [B, K] 原始 softmax 概率(detached), + } + """ + if chem.size(1) != self.n_chem_tokens: + raise ValueError( + f"chem token 数不匹配:期望 {self.n_chem_tokens}, " + f"实际 {chem.size(1)}" + ) + + q_tab = self.tab_pool(tab) # [B, d] + gates, probs_full, expert_mask = self.router(q_tab) # 各 [B, K] + + flat = chem.flatten(start_dim=1) # [B, T_chem * d] + expert_outs = torch.stack( + [expert(flat) for expert in self.experts], dim=1, + ) # [B, K, d] + + F_moe = (gates.unsqueeze(-1) * expert_outs).sum(dim=1) # [B, d] + + lb_loss = self._load_balancing_loss(probs_full, expert_mask) + + return F_moe, { + "lb_loss": lb_loss, + "gates": gates.detach(), + "probs": probs_full.detach(), + } + + def _load_balancing_loss( + self, + probs_full: torch.Tensor, + expert_mask: torch.Tensor, + ) -> torch.Tensor: + """Switch Transformer 风格的 load-balancing loss。 + + f_i = 该 batch 中 expert_i 被激活的样本占比(top-k mask 的均值) + p_i = 该 batch 中 expert_i 的 softmax 概率均值 + loss = K * Σ_i f_i * p_i + + 理想情况下 f_i 与 p_i 都接近 1/K,loss ≈ 1。 + """ + f = expert_mask.mean(dim=0) # [K] + p = probs_full.mean(dim=0) # [K] return self.n_experts * (f * p).sum() \ No newline at end of file