1
v = [1, 2, 3, 4]

1
import torch
2

3
x = torch.tensor([[1.0, 2.0], [3.0, 4.0]])   # 形状 (2, 2)
4
print(x.shape)          # torch.Size([2, 2])
5
print(x[0, 1])          # 2.0 —— 索引
6
print(x @ x)            # 矩阵乘法
7
print(x.transpose(0, 1))# 转置
8

9
a = torch.randn(2, 3)   # 标准正态随机数，形状 (2, 3)
10
b = torch.zeros(1, 3)   # 全 0，形状 (1, 3)
11
print((a + b).shape)    # (2, 3) —— 广播

1
x = torch.tensor(3.0, requires_grad=True)   # 声明"我要对它求导"
2
y = x ** 2                                  # y = x²
3
y.backward()                                # 计算 dy/dx
4
print(x.grad)                               # 6.0 —— 即 2x|x=3

1
import torch.nn as nn
2

3
class MyModel(nn.Module):                    # 所有模型都继承 nn.Module
4
    def __init__(self):
5
        super().__init__()
6
        self.fc = nn.Linear(4, 2)            # 线性层: y = xW^T + b, 4 入 2 出
7
        # W 形状 (2,4), b 形状 (2,), 自动注册为模型参数并参与梯度
8

9
    def forward(self, x):                    # 定义前向计算
10
        return self.fc(x)
11

12
model = MyModel()
13
x = torch.randn(8, 4)                        # (B=8, 4)
14
print(model(x).shape)                        # (8, 2)
15
print(sum(p.numel() for p in model.parameters()))   # 参数量: 2*4+2=10

1
import torch, torch.nn as nn
2

3
x = torch.randn(100, 4)
4
y = (x.sum(dim=1) > 0).long()                # 假任务: 和 >0 则类别 1
5

6
model = nn.Linear(4, 2)
7
opt = torch.optim.Adam(model.parameters(), lr=0.01)   # 优化器
8
loss_fn = nn.CrossEntropyLoss()              # 损失函数
9

10
for step in range(200):
11
    logits = model(x)                        # ① 前向: 预测
12
    loss = loss_fn(logits, y)                # ② 算损失
13
    opt.zero_grad()                          # ③ 清空旧梯度（否则会累加）
14
    loss.backward()                          # ④ 反向传播: 填满 .grad
15
    opt.step()                               # ⑤ 按梯度更新参数
16

17
print("acc =", (model(x).argmax(1) == y).float().mean().item())
18
# 输出接近 1.0，说明训练成功

1
import torch
2
print(torch.__version__)                      # 例如 2.4.0
3
print("CUDA:", torch.cuda.is_available())     # True 表示有 GPU

1
X = [[1, 0],      ← token 0
2
     [0, 1]]      ← token 1

1
Q = K = V = X = [[1, 0],
2
                 [0, 1]]

1
[[0.707, 0.000],
2
 [0.000, 0.707]]

1
输入 X        (B, N, d)
2
投影 Q,K,V    (B, N, d)                     三个线性层各自
3
拆头 reshape  (B, N, H, d_h) → permute → (B, H, N, d_h)
4
打分 Q@K^T    (B, H, N, N)                  每个头一张 N×N 注意力图
5
softmax       (B, H, N, N)                  沿最后维（对 key）归一化
6
加权 attn@V   (B, H, N, d_h)
7
拼头          transpose → reshape → (B, N, d)
8
输出投影 W_O  (B, N, d)

1
pos=0: [sin(0), cos(0), sin(0), cos(0)] = [0, 1, 0, 1]
2
pos=1: [sin(1), cos(1), sin(1/100), cos(1/100)] ≈ [0.841, 0.540, 0.010, 1.000]

1
self.pos_embed = nn.Parameter(torch.zeros(1, N, d))   # 随训练更新

1
mean = 2.5
2
var  = ((1-2.5)²+(2-2.5)²+(3-2.5)²+(4-2.5)²)/4 = 1.25
3
标准化: [-1.342, -0.447, 0.447, 1.342]    （γ=1, β=0 时输出即此）

1
[[1, 0, 0, 0],
2
 [1, 1, 0, 0],
3
 [1, 1, 1, 0],
4
 [1, 1, 1, 1]]    位置 i 只能看到 j ≤ i

1
输入序列 → 词嵌入 + 位置编码
2
   │
3
   ▼
4
┌───────────────────────────────────┐
5
│ Encoder ×N 层                      │
6
│   x = x + MultiHeadAttn(LN(x))    │  双向：看整句
7
│   x = x + FFN(LN(x))              │
8
└───────────────────────────────────┘
9
   │  （编码器的输出 = 对源句的"理解"）
10
   ▼
11
┌───────────────────────────────────┐
12
│ Decoder ×N 层                      │
13
│   x = x + MaskedAttn(LN(x))       │  ① 掩码自注意力：只看已生成的部分
14
│   x = x + CrossAttn(LN(x))        │  ② 交叉注意力：Q 来自解码器, K/V 来自编码器
15
│   x = x + FFN(LN(x))              │
16
└───────────────────────────────────┘
17
   │
18
   ▼
19
Linear + Softmax → 词表概率分布（下一个词）

1
import numpy as np
2

3
def softmax(z, axis=-1):
4
    """数值稳定版 softmax（2.5 节）：先减最大值再指数。"""
5
    z = z - z.max(axis=axis, keepdims=True)      # 防止 exp 溢出
6
    e = np.exp(z)
7
    return e / e.sum(axis=axis, keepdims=True)
8

9
def attention_numpy(X, Wq, Wk, Wv, d_k):
10
    """X: (N, d) 一个序列。返回注意力输出与权重矩阵。"""
11
    Q = X @ Wq          # (N, d_k)   —— 投影出查询
12
    K = X @ Wk          # (N, d_k)   —— 投影出键
13
    V = X @ Wv          # (N, d_k)   —— 投影出值
14
    scores = Q @ K.T / np.sqrt(d_k)  # (N, N)    —— 相似度打分并缩放
15
    A = softmax(scores, axis=1)      # (N, N)    —— 逐行 softmax
16
    out = A @ V                      # (N, d_k)  —— 加权组合
17
    return out, A
18

19
# ---- 复现 2.2 节的例子 ----
20
X = np.array([[1.0, 0.0],
21
              [0.0, 1.0]])           # 2 个 token, 每 token 2 维
22
I = np.eye(2)                        # 单位矩阵: Q=K=V=X
23
out, A = attention_numpy(X, I, I, I, d_k=2)
24

25
print("注意力权重 A =\n", np.round(A, 3))
26
print("输出 =\n", np.round(out, 3))

1
注意力权重 A =
2
 [[0.67  0.33 ]
3
  [0.33  0.67 ]]
4
输出 =
5
 [[0.67  0.33 ]
6
  [0.33  0.67 ]]

1
import math
2
import torch
3
import torch.nn as nn
4
import torch.nn.functional as F
5

6
torch.manual_seed(0)   # 固定随机种子，保证结果可复现
7

8
# ---- 设备选择：默认 GPU ----
9
# set_default_device 让之后创建的所有张量/模块自动落在 GPU 上，正文代码无需再写 .to()/.cuda()
10
torch.set_default_device('cuda')
11
print("默认设备: cuda ——", torch.cuda.get_device_name(0))
12

13

14
class InputEmbedding(nn.Module):
15
    """两种嵌入方式：
16
    vocab 不为 None: 查表嵌入（NLP 用）——token 是词的编号;
17
    vocab 为 None  : 线性嵌入（信号/数值用）——token 是连续数值。
18
    """
19

20
    def __init__(self, d_model: int, vocab: int = None, in_features: int = 1):
21
        super().__init__()
22
        self.d_model = d_model
23
        if vocab is not None:
24
            self.embed = nn.Embedding(vocab, d_model)   # 查表: (B,N) → (B,N,d)
25
        else:
26
            self.embed = nn.Linear(in_features, d_model)  # 线性: (B,N,1) → (B,N,d)
27

28
    def forward(self, x: torch.Tensor) -> torch.Tensor:
29
        return self.embed(x) * math.sqrt(self.d_model)
30

31
# ---- 使用示例 ----
32
emb = InputEmbedding(d_model=16, vocab=1000)   # NLP 风格
33
ids = torch.randint(0, 1000, (2, 5))           # (B=2, N=5) 词编号
34
print(emb(ids).shape)                          # (2, 5, 16)
35

36
emb2 = InputEmbedding(d_model=16, in_features=1)   # 信号风格
37
sig = torch.randn(2, 5, 1)                         # (B=2, N=5, 1) 标量序列
38
print(emb2(sig).shape)                             # (2, 5, 16)

1
class PositionalEncoding(nn.Module):
2
    """正弦位置编码（2.7.1 节）。register_buffer 的内容不参与训练但会随模型保存。"""
3

4
    def __init__(self, d_model: int, max_len: int = 5000, dropout: float = 0.1):
5
        super().__init__()
6
        pe = torch.zeros(max_len, d_model)               # (max_len, d) 先全 0
7
        position = torch.arange(0, max_len).unsqueeze(1).float()   # (max_len, 1): 0,1,2,...
8
        # 分母 10000^(2i/d)：用 exp(log) 形式写，数值更稳（对 i=0,2,4,... 生成 d/2 个）
9
        div_term = torch.exp(
10
            torch.arange(0, d_model, 2).float() * (-math.log(10000.0) / d_model)
11
        )                                                # (d/2,)
12
        pe[:, 0::2] = torch.sin(position * div_term)     # 偶数维填 sin
13
        pe[:, 1::2] = torch.cos(position * div_term)     # 奇数维填 cos
14
        self.register_buffer("pe", pe.unsqueeze(0))      # (1, max_len, d) 预备广播
15
        self.dropout = nn.Dropout(dropout)
16

17
    def forward(self, x: torch.Tensor) -> torch.Tensor:
18
        # x: (B, N, d)。pe[:, :N] 取前 N 个位置，靠广播加到每个样本上
19
        x = x + self.pe[:, : x.size(1), :]
20
        return self.dropout(x)
21

22

23
class LearnablePositionalEncoding(nn.Module):
24
    """可学习位置编码（2.7.2 节，ViT 与两个项目同款）。"""
25

26
    def __init__(self, num_tokens: int, d_model: int, dropout: float = 0.1):
27
        super().__init__()
28
        self.pos_embed = nn.Parameter(torch.zeros(1, num_tokens, d_model))
29
        self.dropout = nn.Dropout(dropout)
30

31
    def forward(self, x: torch.Tensor) -> torch.Tensor:
32
        # x: (B, num_tokens, d)
33
        return self.dropout(x + self.pos_embed)
34

35

36
# ---- 验证：位置编码形状与数值 ----
37
pe = PositionalEncoding(d_model=4, max_len=100)
38
x = torch.zeros(2, 3, 4)                     # (B=2, N=3, d=4)
39
out = pe(x)
40
print(out.shape)                             # (2, 3, 4) —— 形状不变，只加信息
41
print("位置 0 编码:", pe.pe[0, 0].tolist())   # [0, 1, 0, 1]     —— 对照 2.7.1 手算
42
print("位置 1 编码:", [round(v, 3) for v in pe.pe[0, 1].tolist()])  # ≈ [0.841, 0.540, 0.010, 1.0]

1
class MultiHeadAttention(nn.Module):
2
    """多头自注意力。实现方式：Q/K/V 合并为一个大线性层（原论文同款）。
3

4
    形状流程（对照 2.6 节表格）:
5
        x (B,N,d) → qkv (B,N,3d) → reshape (B,N,3,H,d_h) → permute (3,B,H,N,d_h)
6
        → scores (B,H,N,N) → softmax → attn@V → (B,H,N,d_h)
7
        → transpose+reshape (B,N,d) → proj (B,N,d)
8
    """
9

10
    def __init__(self, d_model: int, n_heads: int, dropout: float = 0.1):
11
        super().__init__()
12
        assert d_model % n_heads == 0, f"d_model({d_model}) 必须能被 n_heads({n_heads}) 整除"
13
        self.d_model = d_model
14
        self.n_heads = n_heads
15
        self.d_head = d_model // n_heads        # 每个头的维度 d_h
16
        self.qkv = nn.Linear(d_model, 3 * d_model)   # 一次投影出 Q,K,V（省 3 次调用）
17
        self.proj = nn.Linear(d_model, d_model)      # 输出投影 W_O
18
        self.attn_drop = nn.Dropout(dropout)
19
        self.proj_drop = nn.Dropout(dropout)
20

21
    def forward(self, x: torch.Tensor, mask: torch.Tensor = None,
22
                return_attn: bool = False):
23
        B, N, d = x.shape
24

25
        # ---- 第 1 步：投影 + 拆出 Q/K/V ----
26
        # Linear 输出 (B,N,3d)；reshape 成 (B,N,3,H,d_h)；
27
        # permute(2,0,3,1,4) 把"第 2 维(3)"提到最前 → (3,B,H,N,d_h)，方便拆包
28
        qkv = (
29
            self.qkv(x)
30
            .reshape(B, N, 3, self.n_heads, self.d_head)
31
            .permute(2, 0, 3, 1, 4)
32
        )
33
        q, k, v = qkv[0], qkv[1], qkv[2]        # 各 (B,H,N,d_h)
34

35
        # ---- 第 2 步：打分 + 缩放 ----
36
        # q @ k^T: (B,H,N,d_h)×(B,H,N,d_h)^T → (B,H,N,N)
37
        # 语义：每个头里，每个 query 与每个 key 的相似度
38
        attn = (q @ k.transpose(-2, -1)) * (self.d_head ** -0.5)
39

40
        # ---- 第 3 步：掩码（可选）----
41
        # mask 为 0 的位置置 -inf；softmax(exp(-inf)=0) 后权重为 0
42
        # 支持两种形状：(B,N) padding 掩码 → (B,1,1,N)；(N,N) 因果掩码 → (1,1,N,N)
43
        if mask is not None:
44
            if mask.dim() == 2 and mask.shape == attn.shape[-2:]:
45
                mask = mask.unsqueeze(0).unsqueeze(0)   # 方阵掩码：广播到 batch 与头
46
            else:
47
                mask = mask.unsqueeze(1).unsqueeze(2)   # 逐样本掩码：广播到头与 query
48
            attn = attn.masked_fill(mask == 0, float("-inf"))
49

50
        # ---- 第 4 步：softmax 归一化（沿最后一维 = 对 key 求和为 1）----
51
        attn = attn.softmax(dim=-1)
52
        attn = self.attn_drop(attn)
53

54
        # ---- 第 5 步：加权求和 + 拼头 + 输出投影 ----
55
        out = attn @ v                            # (B,H,N,d_h)
56
        out = out.transpose(1, 2).reshape(B, N, d)  # (B,N,d)：把头拼回特征维
57
        out = self.proj_drop(self.proj(out))
58

59
        if return_attn:
60
            return out, attn                      # 附带注意力图（可视化用）
61
        return out

1
mha = MultiHeadAttention(d_model=32, n_heads=4)
2
x = torch.randn(2, 10, 32)                    # (B=2, N=10, d=32)
3
out, attn = mha(x, return_attn=True)
4
print("输出形状:", out.shape)                  # (2, 10, 32) —— 形状不变
5
print("注意力图形状:", attn.shape)             # (2, 4, 10, 10) —— 每头一张 10×10 图
6
print("每行权重和:", attn[0, 0, 0].sum().item())  # ≈1.0 —— softmax 性质
7

8
# 可训练性检查：参数能收到梯度
9
loss = out.sum()
10
loss.backward()
11
print("qkv 权重梯度非空:", mha.qkv.weight.grad is not None)   # True
12

13
# 掩码检查：屏蔽第 0 个 key 后，第 0 列权重应为 0
14
mask = torch.ones(2, 10); mask[:, 0] = 0      # (B,N) 形状的 padding 掩码
15
out_m, attn_m = mha(x, mask=mask, return_attn=True)
16
print("被屏蔽列权重:", attn_m[0, 0, :, 0].sum().item())        # 0.0 ✓

1
class PositionwiseFeedForward(nn.Module):
2
    """FFN（2.8 节）: d → 4d → d。逐 token 独立，形状永不改变。"""
3

4
    def __init__(self, d_model: int, d_ff: int = None, dropout: float = 0.1):
5
        super().__init__()
6
        d_ff = d_ff or 4 * d_model
7
        self.fc1 = nn.Linear(d_model, d_ff)   # 放大 4 倍给"加工空间"
8
        self.fc2 = nn.Linear(d_ff, d_model)   # 缩回原维度
9
        self.dropout = nn.Dropout(dropout)
10

11
    def forward(self, x: torch.Tensor) -> torch.Tensor:
12
        # x: (B, N, d) —— Linear 自动作用在最后一维，对每个 token 都一样
13
        return self.fc2(self.dropout(F.gelu(self.fc1(x))))
14

15

16
ffn = PositionwiseFeedForward(32)
17
print(ffn(torch.randn(2, 10, 32)).shape)      # (2, 10, 32)

1
class EncoderBlock(nn.Module):
2
    """Pre-LN 编码块（2.9.3 节）：
3
        x = x + Attn(LN(x))     ← 先归一化再进子层，最后残差相加
4
        x = x + FFN(LN(x))
5
    """
6

7
    def __init__(self, d_model: int, n_heads: int, d_ff: int = None,
8
                 dropout: float = 0.1):
9
        super().__init__()
10
        self.norm1 = nn.LayerNorm(d_model)                  # 注意力前的 LN
11
        self.attn = MultiHeadAttention(d_model, n_heads, dropout)
12
        self.norm2 = nn.LayerNorm(d_model)                  # FFN 前的 LN
13
        self.ffn = PositionwiseFeedForward(d_model, d_ff, dropout)
14

15
    def forward(self, x: torch.Tensor, mask: torch.Tensor = None,
16
                return_attn: bool = False):
17
        if return_attn:
18
            a_out, attn = self.attn(self.norm1(x), mask, return_attn=True)
19
            x = x + a_out                                # 残差：梯度直通
20
        else:
21
            x = x + self.attn(self.norm1(x), mask)
22
        x = x + self.ffn(self.norm2(x))                  # FFN 支路 + 残差
23
        if return_attn:
24
            return x, attn
25
        return x
26

27

28
block = EncoderBlock(32, 4)
29
x = torch.randn(2, 10, 32)
30
out, attn = block(x, return_attn=True)
31
print("块输出形状:", out.shape)                # (2, 10, 32)
32
print("块内注意力图:", attn.shape)             # (2, 4, 10, 10)

1
class TransformerEncoder(nn.Module):
2
    def __init__(self, n_layers: int, d_model: int, n_heads: int,
3
                 d_ff: int = None, dropout: float = 0.1,
4
                 block_cls: nn.Module = EncoderBlock):
5
        super().__init__()
6
        # ModuleList 才能被优化器"看见"全部参数；普通 list 不会注册参数！
7
        # block_cls 参数让消融实验可以注入无残差等变体块
8
        self.layers = nn.ModuleList(
9
            [block_cls(d_model, n_heads, d_ff, dropout) for _ in range(n_layers)]
10
        )
11
        self.norm = nn.LayerNorm(d_model)      # 末尾 LN（ViT 惯例）
12

13
    def forward(self, x: torch.Tensor, mask: torch.Tensor = None,
14
                return_attn: bool = False):
15
        all_attns = []
16
        for layer in self.layers:
17
            if return_attn:
18
                x, attn = layer(x, mask, return_attn=True)
19
                all_attns.append(attn)
20
            else:
21
                x = layer(x, mask)
22
        if return_attn:
23
            return self.norm(x), all_attns
24
        return self.norm(x)
25

26

27
encoder = TransformerEncoder(n_layers=2, d_model=32, n_heads=4)
28
x = torch.randn(2, 10, 32)
29
out, attns = encoder(x, return_attn=True)
30
print("编码器输出:", out.shape)                # (2, 10, 32)
31
print("层数 × 每层注意力图:", len(attns), attns[0].shape)   # 2 张 (2,4,10,10)

1
class DecoderBlock(nn.Module):
2
    """解码块（2.11 节）。与编码块的区别：
3
    ① 自注意力加因果掩码（只看过去）；
4
    ② 多一个交叉注意力（Q 来自解码器，K/V 来自编码器输出）。"""
5

6
    def __init__(self, d_model: int, n_heads: int, d_ff: int = None,
7
                 dropout: float = 0.1):
8
        super().__init__()
9
        self.norm1 = nn.LayerNorm(d_model)
10
        self.self_attn = MultiHeadAttention(d_model, n_heads, dropout)   # ① 掩码自注意力
11
        self.norm2 = nn.LayerNorm(d_model)
12
        self.cross_attn = CrossAttention(d_model, n_heads, dropout)      # ② 交叉注意力
13
        self.norm3 = nn.LayerNorm(d_model)
14
        self.ffn = PositionwiseFeedForward(d_model, d_ff, dropout)
15

16
    def forward(self, x: torch.Tensor, enc_out: torch.Tensor,
17
                causal_mask: torch.Tensor) -> torch.Tensor:
18
        x = x + self.self_attn(self.norm1(x), causal_mask)   # 只许看 j ≤ i
19
        x = x + self.cross_attn(self.norm2(x), enc_out)      # 向编码器"提问"
20
        x = x + self.ffn(self.norm3(x))
21
        return x
22

23

24
class CrossAttention(nn.Module):
25
    """交叉注意力：Q 来自 x（解码端），K/V 来自 enc（编码端）。
26
    与自注意力的唯一区别：K/V 的投影层接收的是另一个张量。"""
27

28
    def __init__(self, d_model: int, n_heads: int, dropout: float = 0.1):
29
        super().__init__()
30
        assert d_model % n_heads == 0
31
        self.d_model, self.n_heads = d_model, n_heads
32
        self.d_head = d_model // n_heads
33
        self.wq = nn.Linear(d_model, d_model)      # Q 投影（用于解码端）
34
        self.wk = nn.Linear(d_model, d_model)      # K 投影（用于编码端）
35
        self.wv = nn.Linear(d_model, d_model)      # V 投影（用于编码端）
36
        self.proj = nn.Linear(d_model, d_model)
37
        self.dropout = nn.Dropout(dropout)
38

39
    def forward(self, x: torch.Tensor, enc: torch.Tensor) -> torch.Tensor:
40
        B, N, d = x.shape
41
        E = enc.size(1)                            # 编码端 token 数（可与 N 不同）
42
        q = self.wq(x).reshape(B, N, self.n_heads, self.d_head).permute(0, 2, 1, 3)
43
        k = self.wk(enc).reshape(B, E, self.n_heads, self.d_head).permute(0, 2, 1, 3)
44
        v = self.wv(enc).reshape(B, E, self.n_heads, self.d_head).permute(0, 2, 1, 3)
45
        attn = (q @ k.transpose(-2, -1)) * (self.d_head ** -0.5)   # (B,H,N,E) 注意是 N×E!
46
        attn = attn.softmax(dim=-1)
47
        out = (attn @ v).transpose(1, 2).reshape(B, N, d)
48
        return self.proj(self.dropout(out))
49

50

51
def make_causal_mask(n: int, device=None) -> torch.Tensor:
52
    """下三角矩阵（2.10 节）：(i,j) 处 j ≤ i 才为 1。"""
53
    return torch.tril(torch.ones(n, n, device=device))
54

55

56
# ---- 验证 ----
57
cm = make_causal_mask(4)
58
print(cm)                                        # 对照 2.10 节的 4×4 矩阵
59
dec_block = DecoderBlock(32, 4)
60
x = torch.randn(2, 5, 32)                        # 解码端 5 个 token
61
enc = torch.randn(2, 10, 32)                     # 编码端 10 个 token
62
out = dec_block(x, enc, make_causal_mask(5))
63
print("解码块输出:", out.shape)                  # (2, 5, 32) —— 解码端 token 数不变

1
class Transformer(nn.Module):
2
    """原版 Encoder-Decoder 结构（2.11 节全景图）。"""
3

4
    def __init__(self, vocab_size: int, d_model: int = 32, n_heads: int = 4,
5
                 n_enc: int = 2, n_dec: int = 2, d_ff: int = None,
6
                 dropout: float = 0.1, max_len: int = 100):
7
        super().__init__()
8
        self.d_model = d_model
9
        self.src_embed = nn.Embedding(vocab_size, d_model)      # 源词嵌入
10
        self.tgt_embed = nn.Embedding(vocab_size, d_model)      # 目标词嵌入
11
        self.pos_enc = PositionalEncoding(d_model, max_len, dropout)  # 共用一套位置编码
12
        self.encoder = TransformerEncoder(n_enc, d_model, n_heads, d_ff, dropout)
13
        self.decoder = nn.ModuleList(
14
            [DecoderBlock(d_model, n_heads, d_ff, dropout) for _ in range(n_dec)]
15
        )
16
        self.head = nn.Linear(d_model, vocab_size)              # 词表概率
17

18
    def forward(self, src: torch.Tensor, tgt: torch.Tensor) -> torch.Tensor:
19
        tgt_len = tgt.size(1)
20
        enc_out = self.encoder(self.pos_enc(self.src_embed(src)))      # (B,S,d)
21
        x = self.pos_enc(self.tgt_embed(tgt))                          # (B,T,d)
22
        causal = make_causal_mask(tgt_len, tgt.device)                 # (T,T)
23
        for blk in self.decoder:
24
            x = blk(x, enc_out, causal)
25
        return self.head(x)                                            # (B,T,vocab)
26

27

28
# ---- 形状冒烟测试 ----
29
model = Transformer(vocab_size=100)
30
src = torch.randint(0, 100, (2, 8))     # 源句 8 词
31
tgt = torch.randint(0, 100, (2, 5))     # 目标句 5 词
32
logits = model(src, tgt)
33
print("输出 logits 形状:", logits.shape)  # (2, 5, 100) —— 每个位置预测下一个词

1
class NoamScheduler:
2
    """2.12.4 节 Noam 学习率调度：lr = d^(-0.5) × min(step^(-0.5), step × warmup^(-1.5))
3
    前 warmup 步线性爬升 → 之后按 1/√step 衰减。"""
4

5
    def __init__(self, optimizer, d_model: int, warmup_steps: int = 4000):
6
        self.optimizer = optimizer
7
        self.d_model = d_model
8
        self.warmup_steps = warmup_steps
9
        self.step_num = 0
10

11
    def step(self):
12
        """每次参数更新后调用一次。"""
13
        self.step_num += 1
14
        lr = (self.d_model ** -0.5) * min(
15
            self.step_num ** -0.5,                          # 衰减段
16
            self.step_num * self.warmup_steps ** -1.5,      # 预热段
17
        )
18
        for group in self.optimizer.param_groups:
19
            group["lr"] = lr
20

21
    def get_lr(self):
22
        return self.optimizer.param_groups[0]["lr"]
23

24

25
def label_smoothing_loss(logits, target, eps: float = 0.1):
26
    """2.12.2 节标签平滑交叉熵。
27
    数学: L = (1-ε)·NLL + ε·均匀惩罚。
28
    第一项让模型往正确类别靠，第二项惩罚"把所有概率押在一个类上"。"""
29
    log_probs = F.log_softmax(logits, dim=-1)                        # (B, K)
30
    nll = -log_probs.gather(-1, target.unsqueeze(-1)).squeeze(-1).mean()  # 正确类负对数似然
31
    smooth = -log_probs.mean(dim=-1).mean()                          # 对均匀分布的惩罚
32
    return (1 - eps) * nll + eps * smooth
33

34

35
# ---- 验证调度器曲线（前 10 步应为上升期）----
36
tmp_model = nn.Linear(4, 2)
37
tmp_opt = torch.optim.Adam(tmp_model.parameters(), lr=0)
38
sched = NoamScheduler(tmp_opt, d_model=32, warmup_steps=50)
39
for s in range(5):
40
    sched.step()
41
    print(f"step {s+1}: lr = {sched.get_lr():.6f}")
42
# 输出应逐行增大（预热段线性爬升）

1
class SequenceClassifier(nn.Module):
2
    """完整组装：嵌入 → 位置编码 → N 层 Encoder → 读出 → 分类头
3
    参数:
4
      readout: 'mean'（池化，最常用）| 'cls'（第 0 个 token）| 'last'（最后 token）
5
      pos_mode: 'sinusoidal' | 'learnable' | 'none'（'none' 供消融实验）
6
      block_cls: 默认 EncoderBlock，消融实验可换成无残差变体
7
    """
8

9
    def __init__(self, n_layers: int = 2, d_model: int = 32, n_heads: int = 4,
10
                 num_classes: int = 3, max_len: int = 64,
11
                 readout: str = "mean", pos_mode: str = "sinusoidal",
12
                 dropout: float = 0.1, in_features: int = 1,
13
                 block_cls: nn.Module = EncoderBlock):
14
        super().__init__()
15
        self.embed = nn.Linear(in_features, d_model)     # 标量采样点 → d 维
16
        self.readout = readout
17
        if pos_mode == "sinusoidal":
18
            self.pos_enc = PositionalEncoding(d_model, max_len, dropout)
19
        elif pos_mode == "learnable":
20
            self.pos_enc = LearnablePositionalEncoding(max_len, d_model, dropout)
21
        else:                                            # 'none'：消融用
22
            self.pos_enc = None
23
        self.encoder = TransformerEncoder(n_layers, d_model, n_heads,
24
                                          dropout=dropout, block_cls=block_cls)
25
        self.head = nn.Linear(d_model, num_classes)
26

27
    def forward(self, x: torch.Tensor) -> torch.Tensor:
28
        # x: (B, N) 标量序列
29
        x = self.embed(x.unsqueeze(-1))                  # (B,N,1) → (B,N,d)
30
        if self.pos_enc is not None:
31
            x = self.pos_enc(x)                          # + 位置信息
32
        x = self.encoder(x)                              # N 层注意力混合
33
        if self.readout == "cls":
34
            x = x[:, 0]
35
        elif self.readout == "last":
36
            x = x[:, -1]
37
        else:
38
            x = x.mean(dim=1)                            # 平均池化读出
39
        return self.head(x)                              # (B, num_classes)
40

41

42
def make_signal_data(n_per_class: int = 300, length: int = 64, seed: int = 0):
43
    """3 类"信号"：频率 1/2/3 的正弦波 + 高斯噪声。
44
    与两个项目的 AMC（调制识别）场景同构：类别差异藏在频率（周期结构）里。"""
45
    torch.manual_seed(seed)
46
    xs, ys = [], []
47
    t = torch.arange(length).float() / length * 2 * math.pi   # (length,) 0~2π
48
    for cls_id, freq in enumerate([1.0, 2.0, 3.0]):
49
        for _ in range(n_per_class):
50
            phase = torch.rand(1) * 2 * math.pi
51
            s = torch.sin(freq * t + phase) + 0.2 * torch.randn(length)
52
            xs.append(s)
53
            ys.append(cls_id)
54
    xs = torch.stack(xs)                            # (900, 64)
55
    ys = torch.tensor(ys)                           # (900,)
56
    perm = torch.randperm(len(ys))                  # 打乱顺序
57
    return xs[perm], ys[perm]
58

59

60
def train_demo():
61
    xs, ys = make_signal_data()
62
    n_train = int(len(ys) * 0.8)
63
    x_train, y_train = xs[:n_train], ys[:n_train]
64
    x_test, y_test = xs[n_train:], ys[n_train:]
65

66
    model = SequenceClassifier()
67
    opt = torch.optim.Adam(model.parameters(), lr=1e-3)
68
    sched = NoamScheduler(opt, d_model=32, warmup_steps=100)   # warmup + 衰减
69
    n_params = sum(p.numel() for p in model.parameters())
70
    print(f"模型参数量: {n_params}")
71

72
    for step in range(300):
73
        # ---- 随机采样一个 batch（玩具数据集不写 DataLoader，专注训练循环本身）----
74
        idx = torch.randint(0, len(x_train), (64,))
75
        xb, yb = x_train[idx], y_train[idx]
76

77
        opt.zero_grad()                             # ③ 清空旧梯度
78
        logits = model(xb)                          # ① 前向
79
        loss = label_smoothing_loss(logits, yb, eps=0.1)   # ② 损失（含标签平滑）
80
        loss.backward()                             # ④ 反向
81
        opt.step()                                  # ⑤ 更新
82
        sched.step()                                # 更新学习率
83

84
        if step % 50 == 0 or step == 299:
85
            model.eval()
86
            with torch.no_grad():
87
                acc = (model(x_test).argmax(1) == y_test).float().mean()
88
            model.train()
89
            print(f"step {step:3d}  loss={loss.item():.4f}  "
90
                  f"lr={sched.get_lr():.5f}  test_acc={acc:.3f}")
91

92

93
if __name__ == "__main__":
94
    train_demo()