lrn.091705

Appearance

DUET · Layer 0 — 接入界面 ​

§1 在调用链中的位置 ​

TFB 框架通过 DUET（duet.py）适配器类将 DUETModel 包装起来。DUET 继承 DeepForecastingModelBase，框架调用 _process() 执行前向传播。

§2 I/O 接口定义 ​

实例化侧：

python
DUET(**kwargs)

kwargs 由框架从命令行参数 + MODEL_HYPER_PARAMS 默认值合并而来，存入 self.config（Config 对象）。

调用侧（_process）：

python
def _process(self, input, target, input_mark, target_mark)
参数shape含义
input(B, seq_len, N) = (3, 16, 7)历史观测值
target(B, pred_len, N) = (3, 5, 7)真实未来值（仅训练时用于损失计算）
input_mark(B, seq_len, 4) = (3, 16, 4)时间特征（hour/day/month/weekday）
target_mark(B, pred_len, 4) = (3, 5, 4)同上，对应预测时段
DUET 不使用时间标记

DUETModel.forward() 只接受 input，input_mark 和 target_mark 在 _process 中被传入但未被转发给模型。DUET 的设计完全基于时序值本身（不依赖时间戳特征）。

返回值：

python
{"output": output}                              # 推理时
{"output": output, "additional_loss": loss}     # 训练时

output shape: (B, pred_len, N) = (3, 5, 7)

§3 顺序图（具体层） ​

§4 语义分组图（索引层） ​

§5 逐步骤精读 ​

§5.0 完整原始代码 ​

python
class DUET(DeepForecastingModelBase):
    def __init__(self, **kwargs):
        super(DUET, self).__init__(MODEL_HYPER_PARAMS, **kwargs)

    @property
    def model_name(self):
        return "DUET"

    def _init_model(self):
        return DUETModel(self.config)

    def _process(self, input, target, input_mark, target_mark):
        output, loss_importance = self.model(input)
        out_loss = {"output": output}
        if self.model.training:
            out_loss["additional_loss"] = loss_importance
        return out_loss

MODEL_HYPER_PARAMS（相关默认值，非全量）：

python
MODEL_HYPER_PARAMS = {
    "enc_in": 1,       # 框架自动覆盖为实际变量数
    "d_model": 512,    # MoE expert 输出维度
    "d_ff": 2048,
    "hidden_size": 256,  # 门控 MLP 隐层
    "n_heads": 8,
    "e_layers": 2,
    "num_experts": 4,
    "noisy_gating": True,
    "k": 1,            # top-k 激活专家数
    "CI": True,        # Channel Independent 模式
    "moving_avg": 25,  # series_decomp 核大小
    "dropout": 0.2,
    "fc_dropout": 0.2,
    "loss": "huber",   # 注意：DUET 用 Huber Loss，不是 MSE
    ...
}

§5.1 config 的组装方式 ​

DeepForecastingModelBase.__init__ 把 MODEL_HYPER_PARAMS 和 kwargs 合并为 Config 对象：

python
self.config = Config(model_config, **kwargs)

Config.__init__ 依次写入：DEFAULT_HYPER_PARAMS → model_config（即 MODEL_HYPER_PARAMS）→ kwargs（命令行）。后者覆盖前者。因此命令行参数优先级最高。

框架自动推断的参数（不可从命令行控制）：

参数来源值
config.enc_inmulti_forecasting_hyper_param_tune数据集实际变量列数
config.dec_in同上同 enc_in
config.c_out同上同 enc_in
config.label_len同上seq_len // 2
config.pred_lenkwargs["horizon"] → 被框架传入命令行指定的预测步数
horizon → pred_len 重命名

框架通过 required_hyper_params() 把外部参数名 output_chunk_length 映射为内部的 horizon，再在 Config.__init__ 里把 horizon 赋值给 pred_len（附带一条 deprecated warning）。代码里 config.pred_len 和 config.horizon 同时存在，值相同。

§5.2 _init_model：实例化 DUETModel ​

python
def _init_model(self):
    return DUETModel(self.config)

DUETModel(config) 在 __init__ 中构建：

  • self.cluster = Linear_extractor_cluster(config)：MoE 时序提取器
  • self.mask_generator = Mahalanobis_mask(config.seq_len)：通道相似度掩码生成器
  • self.Channel_transformer = Encoder([EncoderLayer(...) for _ in range(config.e_layers)])：带掩码的变量 Transformer
  • self.linear_head = nn.Sequential(Linear(d_model, pred_len), Dropout)：预测头

toy 参数下（d_model=8, pred_len=5, n_heads=2, d_ff=32, e_layers=2）：

  • linear_head：Linear(8, 5) → 参数量 8×5+5 = 45
  • Channel_transformer：2 层 EncoderLayer，每层含 FullAttention(d_model=8, n_heads=2) + FFN(8→32→8)

§5.3 _process：前向传播与 additional_loss ​

python
def _process(self, input, target, input_mark, target_mark):
    output, loss_importance = self.model(input)
    out_loss = {"output": output}
    if self.model.training:
        out_loss["additional_loss"] = loss_importance
    return out_loss

loss_importance 的作用：MoE 负载均衡损失（cv_squared(importance) + cv_squared(load)），防止所有输入都路由到同一个专家。框架检测到 additional_loss 时自动将其加到主任务损失（Huber）上。

toy 数值追踪（input shape）：

输入 input 形状 (3, 16, 7)：B=3 个样本，每个样本 seq_len=16 个时间步，N=7 个变量。

output 形状 (3, 5, 7)：每个样本预测 pred_len=5 个时间步，N=7 个变量。

loss_importance 是一个标量 tensor（cv_squared 输出 (1,) 再求和），训练时叠加到主损失。

§6 下钻子组件 ​

子组件职责下层文档
DUETModel双路 forward 主链[[02-Layer1-DUETModel主链]]

创建：2026-04-24

*记录并在线阅读我的笔记*