深入理解 GPT 大模型

从 Token、Embedding、Attention、KV Cache 到输出预测的全链路拆解。

深入理解 GPT 大模型:从 Token 到文本生成的全链路解析

在现代自然语言处理领域,大语言模型(LLM)看似是一个神秘的“黑盒”,但如果我们潜入其底层架构,会发现它其实是一条极其严密、优雅的数学与工程流水线。

本文将以一个标准 GPT 模型的架构为例,带你一步步穿透它的大脑,看看一句话是如何变成向量、如何经过千锤百炼的思考,最终预测出下一个词的。

第一阶段:数据的入场与初始化 (Embedding)

这一阶段可以拆成两篇细节文章:

1. Tokenization 与查表映射

模型的输入 XX 是一段人类语言,但机器只认识数字。我们的首要目标是预测一句话的下一个内容(这里的一句话指的是一句知识语料等)。这是最基础的训练模式 和我们见到的对话模型是不是一样。我们目前的阶段是给模型一段话让他学习完成之后,下次见到这段话的前一部分能预测输出下一个token。我想应该叫他学习阶段。这一段的作用其实就是让大模型学会世界上尽可能多的公开的知识语料(最好是确保正确的知识)。

  • 首先,我们将这句话切分为多个 Token,并转化为数字索引(idx)。

  • 接着,我们通过词嵌入层(wte,Word Token Embedding),将每一个离散的 Token 转化为一个维度为 CC(即 n_embd)的密集语义向量。

  • 工程黑科技: 为了压榨 GPU 的 Tensor Core 算力,我们通常会在初始化时,将 vocab_size 向上填充取整到 64 的倍数(padded_vocab_size)。

    # nano GPT
    # Tokenizer based on rustbpe + tiktoken combo
    import pickle
    import rustbpe
    import tiktoken
    
    class RustBPETokenizer:
        """Light wrapper around tiktoken (for efficient inference) but train with rustbpe"""
    
        def __init__(self, enc, bos_token):
            self.enc = enc
            self.bos_token_id = self.encode_special(bos_token)
    
        @classmethod
        def train_from_iterator(cls, text_iterator, vocab_size):
            # 1) train using rustbpe
            tokenizer = rustbpe.Tokenizer()
            # the special tokens are inserted later in __init__, we don't train them here
            vocab_size_no_special = vocab_size - len(SPECIAL_TOKENS)
            assert vocab_size_no_special >= 256, f"vocab_size_no_special must be at least 256, got {vocab_size_no_special}"
            tokenizer.train_from_iterator(text_iterator, vocab_size_no_special, pattern=SPLIT_PATTERN)
            # 2) construct the associated tiktoken encoding for inference
            pattern = tokenizer.get_pattern()
            mergeable_ranks_list = tokenizer.get_mergeable_ranks()
            mergeable_ranks = {bytes(k): v for k, v in mergeable_ranks_list}
            tokens_offset = len(mergeable_ranks)
            special_tokens = {name: tokens_offset + i for i, name in enumerate(SPECIAL_TOKENS)}
            enc = tiktoken.Encoding(
                name="rustbpe",
                pat_str=pattern,
                mergeable_ranks=mergeable_ranks, # dict[bytes, int] (token bytes -> merge priority rank)
                special_tokens=special_tokens, # dict[str, int] (special token name -> token id)
            )
            return cls(enc, "<|bos|>")
    
        @classmethod
        def from_directory(cls, tokenizer_dir):
            pickle_path = os.path.join(tokenizer_dir, "tokenizer.pkl")
            with open(pickle_path, "rb") as f:
                enc = pickle.load(f)
            return cls(enc, "<|bos|>")
    
        @classmethod
        def from_pretrained(cls, tiktoken_name):
            # https://github.com/openai/tiktoken/blob/eedc8563/tiktoken_ext/openai_public.py
            enc = tiktoken.get_encoding(tiktoken_name)
            # tiktoken calls the special document delimiter token "<|endoftext|>"
            # yes this is confusing because this token is almost always PREPENDED to the beginning of the document
            # it most often is used to signal the start of a new sequence to the LLM during inference etc.
            # so in nanoChat we always use "<|bos|>" short for "beginning of sequence", but historically it is often called "<|endoftext|>".
            return cls(enc, "<|endoftext|>")
    
        def get_vocab_size(self):
            return self.enc.n_vocab
    
        def get_special_tokens(self):
            return self.enc.special_tokens_set
    
        def id_to_token(self, id):
            return self.enc.decode([id])
    
        @lru_cache(maxsize=32)
        def encode_special(self, text):
            return self.enc.encode_single_token(text)
    
        def get_bos_token_id(self):
            return self.bos_token_id
    
        def encode(self, text, prepend=None, append=None, num_threads=8):
            # text can be either a string or a list of strings
    
            if prepend is not None:
                prepend_id = prepend if isinstance(prepend, int) else self.encode_special(prepend)
            if append is not None:
                append_id = append if isinstance(append, int) else self.encode_special(append)
    
            if isinstance(text, str):
                ids = self.enc.encode_ordinary(text)
                if prepend is not None:
                    ids.insert(0, prepend_id) # TODO: slightly inefficient here? :( hmm
                if append is not None:
                    ids.append(append_id)
            elif isinstance(text, list):
                ids = self.enc.encode_ordinary_batch(text, num_threads=num_threads)
                if prepend is not None:
                    for ids_row in ids:
                        ids_row.insert(0, prepend_id) # TODO: same
                if append is not None:
                    for ids_row in ids:
                        ids_row.append(append_id)
            else:
                raise ValueError(f"Invalid input type: {type(text)}")
    
            return ids
    
        def __call__(self, *args, **kwargs):
            return self.encode(*args, **kwargs)
    
        def decode(self, ids):
            return self.enc.decode(ids)
    
        def save(self, tokenizer_dir):
            # save the encoding object to disk
            os.makedirs(tokenizer_dir, exist_ok=True)
            pickle_path = os.path.join(tokenizer_dir, "tokenizer.pkl")
            with open(pickle_path, "wb") as f:
                pickle.dump(self.enc, f)
            print(f"Saved tokenizer encoding to {pickle_path}")
    
        def render_conversation(self, conversation, max_tokens=2048):
            """
            Tokenize a single Chat conversation (which we call a "doc" or "document" here).
            Returns:
            - ids: list[int] is a list of token ids of this rendered conversation
            - mask: list[int] of same length, mask = 1 for tokens that the Assistant is expected to train on.
            """
            # ids, masks that we will return and a helper function to help build them up.
            ids, mask = [], []
            def add_tokens(token_ids, mask_val):
                if isinstance(token_ids, int):
                    token_ids = [token_ids]
                ids.extend(token_ids)
                mask.extend([mask_val] * len(token_ids))
    
            # sometimes the first message is a system message...
            # => just merge it with the second (user) message
            if conversation["messages"][0]["role"] == "system":
                # some conversation surgery is necessary here for now...
                conversation = copy.deepcopy(conversation) # avoid mutating the original
                messages = conversation["messages"]
                assert messages[1]["role"] == "user", "System message must be followed by a user message"
                messages[1]["content"] = messages[0]["content"] + "\n\n" + messages[1]["content"]
                messages = messages[1:]
            else:
                messages = conversation["messages"]
            assert len(messages) >= 1, f"Conversation has less than 1 message: {messages}"
    
            # fetch all the special tokens we need
            bos = self.get_bos_token_id()
            user_start, user_end = self.encode_special("<|user_start|>"), self.encode_special("<|user_end|>")
            assistant_start, assistant_end = self.encode_special("<|assistant_start|>"), self.encode_special("<|assistant_end|>")
            python_start, python_end = self.encode_special("<|python_start|>"), self.encode_special("<|python_end|>")
            output_start, output_end = self.encode_special("<|output_start|>"), self.encode_special("<|output_end|>")
    
            # now we can tokenize the conversation
            add_tokens(bos, 0)
            for i, message in enumerate(messages):
    
                # some sanity checking here around assumptions, to prevent footguns
                must_be_from = "user" if i % 2 == 0 else "assistant"
                assert message["role"] == must_be_from, f"Message {i} is from {message['role']} but should be from {must_be_from}"
    
                # content can be either a simple string or a list of parts (e.g. containing tool calls)
                content = message["content"]
    
                if message["role"] == "user":
                    assert isinstance(content, str), "User messages are simply expected to be strings"
                    value_ids = self.encode(content)
                    add_tokens(user_start, 0)
                    add_tokens(value_ids, 0)
                    add_tokens(user_end, 0)
                elif message["role"] == "assistant":
                    add_tokens(assistant_start, 0)
                    if isinstance(content, str):
                        # simple string => simply add the tokens
                        value_ids = self.encode(content)
                        add_tokens(value_ids, 1)
                    elif isinstance(content, list):
                        for part in content:
                            value_ids = self.encode(part["text"])
                            if part["type"] == "text":
                                # string part => simply add the tokens
                                add_tokens(value_ids, 1)
                            elif part["type"] == "python":
                                # python tool call => add the tokens inside <|python_start|> and <|python_end|>
                                add_tokens(python_start, 1)
                                add_tokens(value_ids, 1)
                                add_tokens(python_end, 1)
                            elif part["type"] == "python_output":
                                # python output => add the tokens inside <|output_start|> and <|output_end|>
                                # none of these tokens are supervised because the tokens come from Python at test time
                                add_tokens(output_start, 0)
                                add_tokens(value_ids, 0)
                                add_tokens(output_end, 0)
                            else:
                                raise ValueError(f"Unknown part type: {part['type']}")
                    else:
                        raise ValueError(f"Unknown content type: {type(content)}")
                    add_tokens(assistant_end, 1)
    
            # truncate to max_tokens tokens MAX (helps prevent OOMs)
            ids = ids[:max_tokens]
            mask = mask[:max_tokens]
            return ids, mask
    
        def visualize_tokenization(self, ids, mask, with_token_id=False):
            """Small helper function useful in debugging: visualize the tokenization of render_conversation"""
            RED = '\033[91m'
            GREEN = '\033[92m'
            RESET = '\033[0m'
            GRAY = '\033[90m'
            tokens = []
            for i, (token_id, mask_val) in enumerate(zip(ids, mask)):
                token_str = self.decode([token_id])
                color = GREEN if mask_val == 1 else RED
                tokens.append(f"{color}{token_str}{RESET}")
                if with_token_id:
                    tokens.append(f"{GRAY}({token_id}){RESET}")
            return '|'.join(tokens)
    
        def render_for_completion(self, conversation):
            """
            Used during Reinforcement Learning. In that setting, we want to
            render the conversation priming the Assistant for a completion.
            Unlike the Chat SFT case, we don't need to return the mask.
            """
            # We have some surgery to do: we need to pop the last message (of the Assistant)
            conversation = copy.deepcopy(conversation) # avoid mutating the original
            messages = conversation["messages"]
            assert messages[-1]["role"] == "assistant", "Last message must be from the Assistant"
            messages.pop() # remove the last message (of the Assistant) inplace
    
            # Now tokenize the conversation
            ids, mask = self.render_conversation(conversation)
    
            # Finally, to prime the Assistant for a completion, append the Assistant start token
            assistant_start = self.encode_special("<|assistant_start|>")
            ids.append(assistant_start)
            return ids
    
    

2. 确定输入维度

此时,我们输入模型的数据维度正式确定为:(B,T,C)(B, T, C)

  • BB (Batch):同时输入了几句话(批次大小)。
  • TT (Time):每句话包含多少个 Token(序列长度)。
  • CC (Channel):每个 Token 的特征维度(即 n_embd)。

3. 预埋“位置罗盘” (RoPE)

因为 Transformer 天生是“路痴”,我们在初始化模型时,会提前计算好一条长度为我们所能接受的序列长度 10倍 的“旋转角度卷尺”(包含了 cos 和 sin 矩阵),方便后续精准注入位置信息。输入层中的代码实现见:GPT输入层:Token ID、Embedding与RoPE位置编码,RoPE和GQA的进一步解释见:从 Attention 到 RoPE 与 GQA。


第二阶段:模型的心脏 (Transformer Block 内部流转)

数据 (B,T,C)(B, T, C) 准备就绪后,开始依次穿过模型的核心——由多层(例如 12 层)Transformer Block 组成的深层网络。每个 Block 内部主要包含两部分:因果自注意力机制 (Causal Self-Attention) 和 多层感知机 (MLP),它们之间由前置归一化 (Pre-Norm) 和残差连接 (Residual Connection) 紧密串联。

1. 注意力机制的精密计算 (Attention)

  • 参数初始化: 注意力层内的 QQ、KK、VV 权重矩阵通常采用均匀分布初始化,并严格限制边界 ss 以防止极端值。
  • 归一化: 数据 XX 首先经过 LayerNorm 进行稳压。
  • 生成 Q、K、V: * 通过线性映射,生成 Query (查询)、Key (标签) 和 Value (内容)。
    • 这里引入了前沿的 GQA (Grouped Query Attention) 机制:QQ 分配了 n_head 个头,而 KK 和 VV 只分配了少量的 n_kv_head 个头,以此大幅节省显存。
  • 维度拆解与注入位置: * 将通道维度 CC 拆解为 head 和 head_dim,此时形状变为 (B,T,head,head_dim)(B, T, \text{head}, \text{head\_dim})。
    • 对 QQ 和 KK 的 head_dim 向量分组,乘以预先计算好的 cos 和 sin 角度矩阵,完成旋转位置编码 (RoPE) 的注入。
    • 交换维度,使形状变为 (B,head,T,head_dim)(B, \text{head}, T, \text{head\_dim})。
  • 记忆缓存 (KV Cache): 将当前的 KK、VV 追加到历史缓存中。此时取它们的序列长度维度,就能精确知道本次查询涉及多少个新 Token (TqT_q),以及历史累积了多少个 Token (TkT_k)。
  • 缩放点积与注意力分数:
    • QQ 与历史 KTK^T 做内积运算,此时消掉了 head_dim,得到注意力分数矩阵,形状为 (B,head,Tq,Tk)(B, \text{head}, T_q, T_k)。这代表了每个当前 Token 对历史所有 Token 的关注度。
    • 将分数矩阵乘以内容矩阵 VV(形状为 (B,head,Tk,head_dim)(B, \text{head}, T_k, \text{head\_dim})),得到融合后的特征,形状变为 (B,head,Tq,head_dim)(B, \text{head}, T_q, \text{head\_dim})。
  • 组装出厂: (注:此处修正了原总结中的微小形状失误)
    • 将结果转置重排,并使用 .view() 将所有的头拉平拼接在一起,使形状重新恢复成最初的 (B,T,C)(B, T, C)。
    • 经过最后的线性投影层(c_proj),完成 Attention 计算。

2. 闭门深思 (MLP)

从 Attention 出来的数据,加上原始的 XX(残差连接)后,再次经过归一化,进入 MLP。

  • 在这里,数据维度会被放大 4 倍展开,经过激活函数(如 Squared ReLU)过滤强化后,再压缩回原来的维度 CC。
  • 最终再次与主干道的 XX 相加,形成这个 Block 的最终输出 (B,T,C)(B, T, C),送入下一层。

第三阶段:从思维到表达 (输出与预测)

当数据走完全部 12 个 Block 后,它已经是一团包含了极其丰富上下文逻辑的思维向量 (B,T,C)(B, T, C)。

语言翻译机 (lm_head): (注:此处纠正了非线性运算的误解)

  • 这团数据进入 lm_head 层。这是一个纯线性的全连接层(Linear),它的权重矩阵形状为 (C,padded_vocab_size)(C, \text{padded\_vocab\_size})。
  • 经过矩阵相乘,我们得到了形状为 (B,T,padded_vocab_size)(B, T, \text{padded\_vocab\_size}) 的 Logits(原始得分)。
  • 为了防止数值溢出导致训练崩溃,现代大模型通常会在这里加一个 Softcap 操作(平滑截断),将极其庞大的得分限制在一个安全范围内(如 [−15,15][-15, 15])。

第四阶段:完整的训练与优化循环 (Training Loop)

(注:这部分补全了未写完的优化器闭环)

模型是如何变聪明的?在训练阶段,模型会进行成千上万次的迭代循环,每次循环严格执行以下五步:

  1. 前向传播 (Forward): 将题目 XX 输入模型,拿到预测得分 Logits。
  2. 计算误差 (Loss): 拿到标准答案 YY(即 XX 往后错位一个字的序列),将 Logits 和 YY 一起送入交叉熵损失函数(Cross Entropy)。在这个函数内部,Logits 会被 Softmax 转换为加起来等于 1 的概率,并与正确答案对比,计算出一个误差分数(Loss)。
  3. 清空旧记忆 (Zero Grad): 通知优化器(如 AdamW 或 Muon)擦除上一次计算的梯度记录。
  4. 反向传播 (Backward): 调用 loss.backward(),根据微积分链式法则,从后往前算出模型里每一个权重矩阵对这次误差负有多少责任(计算梯度)。
  5. 更新权重 (Step): 优化器根据算出的梯度,结合动态缩放的学习率(Learning Rate),对模型大脑里的参数进行微调。

就这样,Loss 一点点下降,模型对语言规律的掌握越来越精准。


第五阶段:自回归生成 (Inference / Generation)

当模型训练完毕,进入实战聊天时,我们不再计算 Loss,而是让模型“开口说话”。

  1. 拿到当前最后一个 Token 面对整个词表生成的 Logits 得分。
  2. Top-K 截断: 淘汰掉得分排名靠后的废词(将它们的分数设为负无穷),只保留前 K 个最合理的候选词。
  3. 温度缩放 (Temperature): 根据我们设定的 Temperature 调整分数差距。
  4. Softmax 概率化与抽奖: 将分数转化为概率分布,然后像转盘抽奖一样(Multinomial),按照概率大小随机抽取出这一个字。
  5. 自回归拼接: 把刚抽出的这个字,无缝拼接到原来的输入 XX 后面,重新扔进模型预测下一个字,直到完成整句话。

结语: 从一句简单的输入,到亿万参数矩阵中的穿梭,再到精准的概率输出。大语言模型并没有魔法,它只是将极其暴力的数学计算与极其精妙的工程架构完美结合。当你能手写并闭眼推演这一切时,你就已经站在了通往通用人工智能(AGI)的基石之上。

尺素

0 封尺素

正在取尺素...