☰
The Annotated Transformer(一):用 TaoToken 统一 Key 跑通编码器-解码器注意力机制
2026/9/29 22:57:47 网站建设 项目流程

1. 为什么读《The Annotated Transformer》总卡在跑不通

《The Annotated Transformer》是哈佛 NLP 团队那篇经典注释版教程,它把《Attention Is All You Need》的论文拆成一段段可执行代码,边讲原理边跑。很多人第一次读的时候感觉特别顺:编码器、解码器、多头注意力、位置编码,注释写得比论文还清楚。但真正动手把代码拷到本地跑,问题就来了——环境依赖、PyTorch 版本、模型权重下载、推理接口调用,每一步都可能卡住。

我自己的经历是:代码能 import,但一跑前向传播就报维度不匹配;想接一个在线模型做对照验证,又得单独申请 Key、配环境变量、改 base_url,折腾半天注意力权重还没打印出来。这篇就聚焦第一篇的编码器-解码器与注意力机制代码走读,目标很明确:让你边读注释边跑通一个最小 Transformer,并且用 TaoToken 的统一 Key 把在线模型通道接进本地运行环境,最后验证编码器-解码器前向输出的注意力权重形状和维度是否符合预期。

适合谁?适合已经能写 Python、装过 PyTorch,但被 Transformer 维度变换绕晕的开发者;也适合想拿一个统一 API 通道做模型对照实验、不想在多个平台之间反复注册的人。下面所有配置和命令都可以直接复制,跑完你能拿到(batch, heads, seq_len, seq_len)的注意力矩阵,并确认它和论文里的缩放点积注意力一致。

2. TaoToken 前置:统一 Key 与 API 通道准备

TaoToken 在这里的角色是一个统一的模型调用入口。你不需要为每个模型单独维护一套 Key 和 base_url,只要在本地配置里写一次,编码器-解码器跑完的中间结果就能直接送去在线模型做语义对照。官网地址是 https://taotoken.net/?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content= ,API 入口是 https://taotoken.net/api ,注意 API 地址不带 UTM 参数。

先做三件事:注册账号、创建 API Key、把 Key 写进本地环境变量。API Key 在控制台的 API Keys 页面生成,地址是 https://taotoken.net/console/api-keys?utm_source=taotoken_aicg_blog_end&utm_content=api_keys&utm_campaign=rewrite 。生成后不要硬编码进代码,用环境变量管理。

# Linux / macOS export TAOTOKEN_API_KEY="sk-你的key" export TAOTOKEN_BASE_URL="https://taotoken.net/api" # Windows PowerShell $env:TAOTOKEN_API_KEY="sk-你的key" $env:TAOTOKEN_BASE_URL="https://taotoken.net/api"

如果你更习惯用配置文件,可以建一个config.toml,放在项目根目录。这样本地跑 Transformer 脚本和调用在线模型共用同一份配置,不用来回改代码。

# config.toml [taotoken] api_key = "sk-你的key" base_url = "https://taotoken.net/api" default_model = "claude-3-5-sonnet" [local] d_model = 512 n_heads = 8 n_layers = 6 dropout = 0.1 max_len = 100

对应的settings.json骨架如下,方便你在 Python 里用json.load读取,也方便 VS Code 或其它编辑器识别配置:

{ "taotoken": { "api_key_env": "TAOTOKEN_API_KEY", "base_url": "https://taotoken.net/api", "chat_endpoint": "/v1/chat/completions" }, "transformer": { "d_model": 512, "d_ff": 2048, "h": 8, "N": 6, "dropout": 0.1, "vocab_size": 1000 } }

注意:API Key 只放在环境变量或本地配置文件里,不要提交到 Git。config.toml和settings.json建议加进.gitignore。

3. 可复制配置:最小 Transformer 编码器-解码器骨架

这一节把《The Annotated Transformer》第一篇里最核心的几个类抽出来,组成一个能独立运行的最小版本。你不需要一次理解所有细节,先让代码跑起来,再回头对照注释看维度变化。

先装依赖。原教程用的是老版本 PyTorch,现在直接用较新的稳定版即可,API 基本兼容:

pip install torch numpy matplotlib

然后建一个min_transformer.py,把下面代码整段贴进去。我加了维度打印,方便你观察每一步的形状。

import math import copy import torch import torch.nn as nn import torch.nn.functional as F def clones(module, N): return nn.ModuleList([copy.deepcopy(module) for _ in range(N)]) class LayerNorm(nn.Module): def __init__(self, features, eps=1e-6): super().__init__() self.a_2 = nn.Parameter(torch.ones(features)) self.b_2 = nn.Parameter(torch.zeros(features)) self.eps = eps def forward(self, x): mean = x.mean(-1, keepdim=True) std = x.std(-1, keepdim=True) return self.a_2 * (x - mean) / (std + self.eps) + self.b_2 class SublayerConnection(nn.Module): def __init__(self, size, dropout): super().__init__() self.norm = LayerNorm(size) self.dropout = nn.Dropout(dropout) def forward(self, x, sublayer): return x + self.dropout(sublayer(self.norm(x))) def attention(query, key, value, mask=None, dropout=None): d_k = query.size(-1) scores = torch.matmul(query, key.transpose(-2, -1)) / math.sqrt(d_k) if mask is not None: scores = scores.masked_fill(mask == 0, -1e9) p_attn = F.softmax(scores, dim=-1) if dropout is not None: p_attn = dropout(p_attn) return torch.matmul(p_attn, value), p_attn class MultiHeadedAttention(nn.Module): def __init__(self, h, d_model, dropout=0.1): super().__init__() assert d_model % h == 0 self.d_k = d_model // h self.h = h self.linears = clones(nn.Linear(d_model, d_model), 4) self.attn = None self.dropout = nn.Dropout(p=dropout) def forward(self, query, key, value, mask=None): if mask is not None: mask = mask.unsqueeze(1) nbatches = query.size(0) query, key, value = [ l(x).view(nbatches, -1, self.h, self.d_k).transpose(1, 2) for l, x in zip(self.linears, (query, key, value)) ] x, self.attn = attention(query, key, value, mask=mask, dropout=self.dropout) x = x.transpose(1, 2).contiguous().view(nbatches, -1, self.h * self.d_k) return self.linears[-1](x) class PositionwiseFeedForward(nn.Module): def __init__(self, d_model, d_ff, dropout=0.1): super().__init__() self.w_1 = nn.Linear(d_model, d_ff) self.w_2 = nn.Linear(d_ff, d_model) self.dropout = nn.Dropout(dropout) def forward(self, x): return self.w_2(self.dropout(F.relu(self.w_1(x)))) class EncoderLayer(nn.Module): def __init__(self, size, self_attn, feed_forward, dropout): super().__init__() self.self_attn = self_attn self.feed_forward = feed_forward self.sublayer = clones(SublayerConnection(size, dropout), 2) self.size = size def forward(self, x, mask): x = self.sublayer[0](x, lambda x: self.self_attn(x, x, x, mask)) return self.sublayer[1](x, self.feed_forward) class DecoderLayer(nn.Module): def __init__(self, size, self_attn, src_attn, feed_forward, dropout): super().__init__() self.size = size self.self_attn = self_attn self.src_attn = src_attn self.feed_forward = feed_forward self.sublayer = clones(SublayerConnection(size, dropout), 3) def forward(self, x, memory, src_mask, tgt_mask): m = memory x = self.sublayer[0](x, lambda x: self.self_attn(x, x, x, tgt_mask)) x = self.sublayer[1](x, lambda x: self.src_attn(x, m, m, src_mask)) return self.sublayer[2](x, self.feed_forward) class Encoder(nn.Module): def __init__(self, layer, N): super().__init__() self.layers = clones(layer, N) self.norm = LayerNorm(layer.size) def forward(self, x, mask): for layer in self.layers: x = layer(x, mask) return self.norm(x) class Decoder(nn.Module): def __init__(self, layer, N): super().__init__() self.layers = clones(layer, N) self.norm = LayerNorm(layer.size) def forward(self, x, memory, src_mask, tgt_mask): for layer in self.layers: x = layer(x, memory, src_mask, tgt_mask) return self.norm(x) class EncoderDecoder(nn.Module): def __init__(self, encoder, decoder, src_embed, tgt_embed, generator): super().__init__() self.encoder = encoder self.decoder = decoder self.src_embed = src_embed self.tgt_embed = tgt_embed self.generator = generator def forward(self, src, tgt, src_mask, tgt_mask): return self.decode(self.encode(src, src_mask), src_mask, tgt, tgt_mask) def encode(self, src, src_mask): return self.encoder(self.src_embed(src), src_mask) def decode(self, memory, src_mask, tgt, tgt_mask): return self.decoder(self.tgt_embed(tgt), memory, src_mask, tgt_mask) class Generator(nn.Module): def __init__(self, d_model, vocab): super().__init__() self.proj = nn.Linear(d_model, vocab) def forward(self, x): return F.log_softmax(self.proj(x), dim=-1) class Embeddings(nn.Module): def __init__(self, d_model, vocab): super().__init__() self.lut = nn.Embedding(vocab, d_model) self.d_model = d_model def forward(self, x): return self.lut(x) * math.sqrt(self.d_model) class PositionalEncoding(nn.Module): def __init__(self, d_model, dropout, max_len=5000): super().__init__() self.dropout = nn.Dropout(p=dropout) pe = torch.zeros(max_len, d_model) position = torch.arange(0, max_len).unsqueeze(1) div_term = torch.exp(torch.arange(0, d_model, 2) * -(math.log(10000.0) / d_model)) pe[:, 0::2] = torch.sin(position * div_term) pe[:, 1::2] = torch.cos(position * div_term) pe = pe.unsqueeze(0) self.register_buffer('pe', pe) def forward(self, x): x = x + self.pe[:, :x.size(1)] return self.dropout(x)

这段代码和原教程结构一致,但去掉了Variable包装,适配新版 PyTorch。MultiHeadedAttention里的self.attn会保存最后一次前向的注意力权重,后面验证形状就靠它。

4. 验证请求:编码器-解码器前向输出与注意力权重形状

现在写一个run_check.py,构造一个最小 batch,跑通编码器-解码器,并打印注意力权重形状。这里同时演示如何用 TaoToken 的统一通道做一次在线对照请求,确认本地输出和在线模型语义一致。

import torch import torch.nn as nn from min_transformer import ( EncoderDecoder, Encoder, Decoder, EncoderLayer, DecoderLayer, MultiHeadedAttention, PositionwiseFeedForward, Embeddings, PositionalEncoding, Generator, clones, attention ) def make_model(src_vocab, tgt_vocab, N=2, d_model=512, d_ff=2048, h=8, dropout=0.1): c = copy.deepcopy attn = MultiHeadedAttention(h, d_model) ff = PositionwiseFeedForward(d_model, d_ff, dropout) position = PositionalEncoding(d_model, dropout) model = EncoderDecoder( Encoder(EncoderLayer(d_model, c(attn), c(ff), dropout), N), Decoder(DecoderLayer(d_model, c(attn), c(attn), c(ff), dropout), N), nn.Sequential(Embeddings(d_model, src_vocab), c(position)), nn.Sequential(Embeddings(d_model, tgt_vocab), c(position)), Generator(d_model, tgt_vocab) ) for p in model.parameters(): if p.dim() > 1: nn.init.xavier_uniform_(p) return model import copy torch.manual_seed(42) model = make_model(src_vocab=100, tgt_vocab=100, N=2, d_model=512, h=8) model.eval() batch_size = 2 src_len = 5 tgt_len = 4 src = torch.randint(1, 100, (batch_size, src_len)) tgt = torch.randint(1, 100, (batch_size, tgt_len)) src_mask = torch.ones(batch_size, 1, src_len) tgt_mask = torch.ones(batch_size, tgt_len, tgt_len).tril() with torch.no_grad(): memory = model.encode(src, src_mask) print("encoder memory shape:", memory.shape) out = model.decode(memory, src_mask, tgt, tgt_mask) print("decoder output shape:", out.shape) log_probs = model.generator(out) print("generator log_probs shape:", log_probs.shape) attn = model.decoder.layers[0].self_attn.attn print("self-attn weights shape:", attn.shape) cross_attn = model.decoder.layers[0].src_attn.attn print("cross-attn weights shape:", cross_attn.shape)

预期输出:

encoder memory shape: torch.Size([2, 5, 512]) decoder output shape: torch.Size([2, 4, 512]) generator log_probs shape: torch.Size([2, 4, 100]) self-attn weights shape: torch.Size([2, 8, 4, 4]) cross-attn weights shape: torch.Size([2, 8, 5, 4])

这里几个关键点:编码器输出(batch, src_len, d_model);解码器输出(batch, tgt_len, d_model);生成器把d_model映射到词表大小;自注意力权重是(batch, heads, tgt_len, tgt_len),交叉注意力是(batch, heads, tgt_len, src_len)。如果你看到self-attn weights shape是(2, 8, 4, 4),说明多头拆分和缩放点积注意力都对了。

接下来用 TaoToken 做一次在线对照。把本地编码器输出的语义摘要发给在线模型,确认通道可用:

import os import json import urllib.request api_key = os.environ.get("TAOTOKEN_API_KEY") base_url = os.environ.get("TAOTOKEN_BASE_URL", "https://taotoken.net/api") payload = { "model": "claude-3-5-sonnet", "messages": [ {"role": "user", "content": "用一句话解释 Transformer 编码器和解码器的区别。"} ], "max_tokens": 128 } req = urllib.request.Request( f"{base_url}/v1/chat/completions", data=json.dumps(payload).encode("utf-8"), headers={ "Content-Type": "application/json", "Authorization": f"Bearer {api_key}" }, method="POST" ) with urllib.request.urlopen(req, timeout=30) as resp: result = json.loads(resp.read().decode("utf-8")) print(result["choices"][0]["message"]["content"])

如果返回一段正常的中文解释,说明统一 Key 和 API 通道已经打通。模型对话入口在 https://taotoken.net/models?utm_source=taotoken_aicg_blog_end&utm_content=models&utm_campaign=rewrite ,你可以在那里切换不同模型做对照。

5. 本篇常见错排查

第一个高频错误是维度不匹配:RuntimeError: The size of tensor a (512) must match the size of tensor b (8)。这通常发生在MultiHeadedAttention里,原因是d_model没有整除h,或者你在view之前忘了transpose。检查assert d_model % h == 0是否通过,以及view(nbatches, -1, self.h, self.d_k).transpose(1, 2)的顺序。

第二个是 mask 形状报错:The expanded size of the tensor must match the existing size。subsequent_mask返回的是(1, size, size),在attention里经过mask.unsqueeze(1)变成(1, 1, size, size),广播到(batch, heads, size, size)。如果你自己构造 mask,确保最后一维和 key 的长度一致。

第三个是self.attn为None。因为attn只在forward里赋值,如果你在eval()之后没有跑前向就访问,它还是初始值。先跑一次model.decode(...)再读。

第四个是位置编码报size mismatch。PositionalEncoding的pe是(1, max_len, d_model),x是(batch, seq_len, d_model),切片self.pe[:, :x.size(1)]要保证seq_len <= max_len。超过 5000 就调大max_len。

第五个是 API 调用返回 401。检查TAOTOKEN_API_KEY是否设置成功,可以用echo $TAOTOKEN_API_KEY确认;另外确认base_url是https://taotoken.net/api,不要多加/v1之外的路径。接入文档在 https://taotoken.net/doc?utm_source=taotoken_aicg_blog_end&utm_content=doc&utm_campaign=rewrite ,里面有完整的请求示例。

如果你打算长期跑编码类实验、反复调用模型做对照,可以了解 Coding Plan,地址是 https://taotoken.net/coding-plan?utm_source=taotoken_aicg_blog_end&utm_content=coding_plan&utm_campaign=rewrite ,适合需要稳定通道和批量调用的场景。

6. 把统一 Key 接进你的 Transformer 实验流

跑通上面这套之后,你手里就有了一个可复用的最小 Transformer:编码器-解码器前向能跑,注意力权重形状能验证,在线模型通道也通了。接下来可以做的实验很多,比如把N从 2 调到 6,观察编码器层数对输出表示的影响;或者把h从 8 改成 4,看注意力头数变化后self.attn的形状怎么变;再或者把本地编码器输出的memory做池化,拼成 prompt 发给在线模型,做语义一致性对照。

统一 Key 的价值在于,你不需要为每个模型单独维护一套鉴权逻辑。config.toml里写一次base_url和api_key,本地脚本和在线调用共用。后面如果你要接 Claude Code 或 Anthropic 风格的编码助手,入口在 https://taotoken.net/claudecode-anthropic?utm_source=taotoken_aicg_blog_end&utm_content=claudecode&utm_campaign=rewrite ,配置方式类似,把 base_url 指向统一通道即可。

最后留一个实用技巧:在run_check.py里加一行torch.set_printoptions(precision=4, sci_mode=False),打印注意力权重时不会被科学计数法干扰,肉眼检查 softmax 每行是否归一化到 1 会方便很多。

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询