LLaMA 系列

import os  # 导入 os 模块，用于操作系统功能，如文件路径
from logging import getLogger  # 从 logging 模块导入 getLogger，用于日志记录
from typing import List  # 从 typing 模块导入 List，用于指定列表类型的注释

from sentencepiece import SentencePieceProcessor  # 从 sentencepiece 导入 SentencePieceProcessor，用于文本分词和编码/解码


logger = getLogger()  # 获取一个日志记录器对象


class Tokenizer:
    """使用 SentencePiece 进行文本的分词和编码/解码。"""
    def __init__(self, model_path: str):
        """
        使用 SentencePiece 模型初始化 Tokenizer。

        参数:
            model_path (str): SentencePiece 模型文件的路径。
        """
        # 重新加载分词器
        assert os.path.isfile(model_path), model_path  # 断言模型路径是一个文件，如果不是则抛出异常
        self.sp_model = SentencePieceProcessor(model_file=model_path)  # 加载 SentencePiece 模型
        logger.info(f"Reloaded SentencePiece model from {model_path}")  # 记录日志，表示模型已重新加载

        # 设置开始（BOS）/结束（EOS）标记的 ID
        self.n_words: int = self.sp_model.vocab_size()  # 词汇表大小
        self.bos_id: int = self.sp_model.bos_id()  # 开始标记的 ID
        self.eos_id: int = self.sp_model.eos_id()  # 结束标记的 ID
        self.pad_id: int = self.sp_model.pad_id()  # 填充（PAD）标记的 ID
        logger.info(
            f"#words: {self.n_words} - BOS ID: {self.bos_id} - EOS ID: {self.eos_id}"
        )  # 记录词汇表大小和各标记的 ID
        assert self.sp_model.vocab_size() == self.sp_model.get_piece_size()  # 确保词汇表大小与分词器的大小一致

    def encode(self, s: str, bos: bool, eos: bool) -> List[int]:
        """
        将字符串编码成一个 token ID 列表。

        参数:
            s (str): 要编码的输入字符串。
            bos (bool): 是否在序列开始处添加开始标记。
            eos (bool): 是否在序列结束处添加结束标记。

        返回:
            List[int]: token ID 的列表。
        """
        assert type(s) is str  # 断言输入是字符串类型
        t = self.sp_model.encode(s)  # 使用 SentencePiece 模型编码字符串
        if bos:
            t = [self.bos_id] + t  # 如果需要，添加开始标记
        if eos:
            t = t + [self.eos_id]  # 如果需要，添加结束标记
        return t

    def decode(self, t: List[int]) -> str:
        """
        将 token ID 列表解码成字符串。

        参数:
            t (List[int]): 要解码的 token ID 列表。

        返回:
            str: 解码后的字符串。
        """
        return self.sp_model.decode(t)  # 使用 SentencePiece 模型解码 token ID 列表

"low": l o w </w>, 5次
"lower": l o w e r </w>, 3次
"new": n e w </w>, 4次

l o: 8次（5次来自 "low"，3次来自 "lower"）
o w: 8次（5次来自 "low"，3次来自 "lower"）
w </w>: 9次（5次来自 "low"，4次来自 "new"）

"low": lo w </w>, 5次
"lower": lo w e r </w>, 3次
"new": n e w </w>, 4次

[l, o, w, e, r, n, </w>, lo, low, new, ...]

import torch
import torch.nn as nn

class RMSNorm(torch.nn.Module):
    # 定义一个继承自 torch.nn.Module 的 RMSNorm 类
    def __init__(self, dim: int, eps: float = 1e-6):
        """
        初始化 RMSNorm 归一化层。
        参数:
            dim (int): 输入张量的维度。
            eps (float, optional): 为了数值稳定性添加到分母的小值，默认为 1e-6。
        属性:
            eps (float): 为了数值稳定性添加到分母的小值。
            weight (nn.Parameter): 可学习的缩放参数。
        """
        super().__init__()  # 调用父类的初始化方法
        self.eps = eps  # 存储数值稳定性参数
        self.weight = nn.Parameter(torch.ones(dim))  # 初始化可学习的缩放参数

    def _norm(self, x):
        """
        对输入张量应用 RMSNorm 归一化。
        参数:
            x (torch.Tensor): 输入张量。
        返回:
            torch.Tensor: 归一化后的张量。
        """
        # 计算归一化值，使用均方根（RMS）作为分母
        return x * torch.rsqrt(x.pow(2).mean(-1, keepdim=True) + self.eps)

    def forward(self, x):
        """
        RMSNorm 层的前向传播。
        参数:
            x (torch.Tensor): 输入张量。
        返回:
            torch.Tensor: 应用 RMSNorm 后的输出张量。
        """
        # 将输入张量转换为 float 类型，应用归一化，然后恢复原始类型
        output = self._norm(x.float()).type_as(x)
        # 应用可学习的缩放参数并返回结果
        return output * self.weight

import torch
import torch.nn as nn
import torch.nn.functional as F

class FeedForward(nn.Module):
    def __init__(
        self,
        dim: int,          # 输入维度（等于 Transformer 中的 hidden_size）
        hidden_dim: int,   # FFN 隐藏层维度（原始 Transformer 通常为 4 × dim）
        multiple_of: int,  # 用于将隐藏层维度对齐到指定倍数（如 64 的倍数，以加速 GPU 计算）
    ):
        super().__init__()

        # ============================================================
        # Step 1: 调整隐藏层维度（LLaMA 特有）
        # 原始 Transformer 使用 hidden_dim = 4 × dim
        # LLaMA 使用 SwiGLU 激活，因此将维度缩小为原来的 2/3，
        # 即 hidden_dim = (2/3) × (4 × dim) ≈ 2.67 × dim
        # 这样可以在保持性能的同时减少参数量和计算量。
        # ============================================================
        hidden_dim = int(2 * hidden_dim / 3)

        # ============================================================
        # Step 2: 对齐 hidden_dim 至 multiple_of 的倍数
        # 例如 multiple_of = 64 时，将 hidden_dim 向上取整为 64 的倍数，
        # 便于 GPU 矩阵乘法在 CUDA Tensor Core 上高效执行。
        # ============================================================
        hidden_dim = multiple_of * ((hidden_dim + multiple_of - 1) // multiple_of)

        # ============================================================
        # Step 3: 定义前馈网络的线性层
        # LLaMA 的 FFN 采用 SwiGLU 结构：
        #   FFN(x) = W2( SiLU(W1(x)) * W3(x) )
        #
        # - W1 和 W3 都是升维层（输入维度 dim → hidden_dim）
        # - W2 是降维层（hidden_dim → dim）
        # - “*” 表示逐元素相乘（门控机制）
        # ============================================================

        # W1：输入升维，用于生成主特征分支 A
        self.w1 = ColumnParallelLinear(
            dim, hidden_dim, bias=False, gather_output=False, init_method=lambda x: x
        )

        # W2：输出降维，将门控结果投影回原维度
        self.w2 = RowParallelLinear(
            hidden_dim, dim, bias=False, input_is_parallel=True, init_method=lambda x: x
        )

        # W3：输入升维，用于生成门控分支 B
        self.w3 = ColumnParallelLinear(
            dim, hidden_dim, bias=False, gather_output=False, init_method=lambda x: x
        )

    def forward(self, x):
        # ============================================================
        # 前向传播过程：
        # 1. w1(x)：主分支特征 A
        # 2. w3(x)：门控分支特征 B
        # 3. F.silu(w1(x))：对主分支使用 SiLU 激活函数（Swish）
        # 4. F.silu(w1(x)) * w3(x)：门控机制，控制信息通过比例
        # 5. w2(...)：将结果映射回原维度
        #
        # 最终公式：
        # FFN(x) = W2( SiLU(W1(x)) * W3(x) )
        # ============================================================
        return self.w2(F.silu(self.w1(x)) * self.w3(x))

# 代码增加了注释，可以看到和原始公式的对应关系。
class LlamaRotaryEmbedding(torch.nn.Module):
    def __init__(self, dim, max_position_embeddings=2048, base=10000, device=None):
        super().__init__()
        # 此处 inv_freq 对应公式中的 theta
        inv_freq = 1.0 / (base ** (torch.arange(0, dim, 2).float().to(device) / dim))
        self.register_buffer("inv_freq", inv_freq)

        self.max_seq_len_cached = max_position_embeddings
        t = torch.arange(self.max_seq_len_cached, device=self.inv_freq.device, dtype=self.inv_freq.dtype)
        # 此处 freqs 对应公式中的 m * theta, t 对应公式中的 m，表示位置
        freqs = torch.einsum("i,j->ij", t, self.inv_freq)
        # Different from paper, but it uses a different permutation in order to obtain the same calculation
        # 此处和原始公式不同，theta_0 和 theta_0 不再相邻
        # 而是分在向量的前半部分和后半部分
        emb = torch.cat((freqs, freqs), dim=-1)
        dtype = torch.get_default_dtype()
        self.register_buffer("cos_cached", emb.cos()[None, None, :, :].to(dtype), persistent=False)
        self.register_buffer("sin_cached", emb.sin()[None, None, :, :].to(dtype), persistent=False)

    def forward(self, x, seq_len=None):
        # x: [bs, num_attention_heads, seq_len, head_size]
        if seq_len > self.max_seq_len_cached:
            self.max_seq_len_cached = seq_len
            t = torch.arange(self.max_seq_len_cached, device=x.device, dtype=self.inv_freq.dtype)
            freqs = torch.einsum("i,j->ij", t, self.inv_freq)
            # Different from paper, but it uses a different permutation in order to obtain the same calculation
            emb = torch.cat((freqs, freqs), dim=-1).to(x.device)
            self.register_buffer("cos_cached", emb.cos()[None, None, :, :].to(x.dtype), persistent=False)
            self.register_buffer("sin_cached", emb.sin()[None, None, :, :].to(x.dtype), persistent=False)
        # 大部分情况下，直接从这里返回
        return (
            self.cos_cached[:, :, :seq_len, ...].to(dtype=x.dtype),
            self.sin_cached[:, :, :seq_len, ...].to(dtype=x.dtype),
        )


def rotate_half(x):
    """Rotates half the hidden dims of the input."""
    # 此次和原始推导中不同，正负号不是间隔的，而是分前半部分和后半部分。但对于结果没有影响
    x1 = x[..., : x.shape[-1] // 2]
    x2 = x[..., x.shape[-1] // 2 :]
    return torch.cat((-x2, x1), dim=-1)


def apply_rotary_pos_emb(q, k, cos, sin, position_ids):
    # The first two dimensions of cos and sin are always 1, so we can `squeeze` them.
    cos = cos.squeeze(1).squeeze(0)  # [seq_len, dim]
    sin = sin.squeeze(1).squeeze(0)  # [seq_len, dim]
    cos = cos[position_ids].unsqueeze(1)  # [bs, 1, seq_len, dim]
    sin = sin[position_ids].unsqueeze(1)  # [bs, 1, seq_len, dim]
    # 对应上图中 RoPE 的简化计算
    q_embed = (q * cos) + (rotate_half(q) * sin)
    k_embed = (k * cos) + (rotate_half(k) * sin)
    return q_embed, k_embed

class Attention(nn.Module):
    def __init__(self, args: ModelArgs):
        super().__init__()

        self.n_local_heads = args.n_heads // fs_init.get_model_parallel_world_size()
        self.head_dim = args.dim // args.n_heads

        self.wq = ColumnParallelLinear(
            args.dim,
            args.n_heads * self.head_dim,
            bias=False,
            gather_output=False,
            init_method=lambda x: x,
        )
        self.wk = ColumnParallelLinear(
            args.dim,
            args.n_heads * self.head_dim,
            bias=False,
            gather_output=False,
            init_method=lambda x: x,
        )
        self.wv = ColumnParallelLinear(
            args.dim,
            args.n_heads * self.head_dim,
            bias=False,
            gather_output=False,
            init_method=lambda x: x,
        )
        self.wo = RowParallelLinear(
            args.n_heads * self.head_dim,
            args.dim,
            bias=False,
            input_is_parallel=True,
            init_method=lambda x: x,
        )

        self.cache_k = torch.zeros(
            (args.max_batch_size, args.max_seq_len, self.n_local_heads, self.head_dim)
        ).cuda()
        self.cache_v = torch.zeros(
            (args.max_batch_size, args.max_seq_len, self.n_local_heads, self.head_dim)
        ).cuda()

    def forward(self, x: torch.Tensor, start_pos: int, freqs_cis: torch.Tensor, mask: Optional[torch.Tensor]):
        bsz, seqlen, _ = x.shape
        xq, xk, xv = self.wq(x), self.wk(x), self.wv(x)

        xq = xq.view(bsz, seqlen, self.n_local_heads, self.head_dim)
        xk = xk.view(bsz, seqlen, self.n_local_heads, self.head_dim)
        xv = xv.view(bsz, seqlen, self.n_local_heads, self.head_dim)

        xq, xk = apply_rotary_emb(xq, xk, freqs_cis=freqs_cis)

        self.cache_k = self.cache_k.to(xq)
        self.cache_v = self.cache_v.to(xq)

        self.cache_k[:bsz, start_pos : start_pos + seqlen] = xk
        self.cache_v[:bsz, start_pos : start_pos + seqlen] = xv

        keys = self.cache_k[:bsz, : start_pos + seqlen]
        values = self.cache_v[:bsz, : start_pos + seqlen]

        xq = xq.transpose(1, 2)
        keys = keys.transpose(1, 2)
        values = values.transpose(1, 2)
        scores = torch.matmul(xq, keys.transpose(2, 3)) / math.sqrt(self.head_dim)
        if mask is not None:
            scores = scores + mask  # (bs, n_local_heads, slen, cache_len + slen)
        scores = F.softmax(scores.float(), dim=-1).type_as(xq)
        output = torch.matmul(scores, values)  # (bs, n_local_heads, slen, head_dim)
        output = output.transpose(
            1, 2
        ).contiguous().view(bsz, seqlen, -1)

        return self.wo(output)

class TransformerBlock(nn.Module):
    def __init__(self, layer_id: int, args: ModelArgs):
        super().__init__()
        self.n_heads = args.n_heads
        self.dim = args.dim
        self.head_dim = args.dim // args.n_heads
        self.attention = Attention(args)
        self.feed_forward = FeedForward(
            dim=args.dim, hidden_dim=4 * args.dim, multiple_of=args.multiple_of
        )
        self.layer_id = layer_id
        self.attention_norm = RMSNorm(args.dim, eps=args.norm_eps)
        self.ffn_norm = RMSNorm(args.dim, eps=args.norm_eps)

    def forward(self, x: torch.Tensor, start_pos: int, freqs_cis: torch.Tensor, mask: Optional[torch.Tensor]):
        h = x + self.attention.forward(self.attention_norm(x), start_pos, freqs_cis, mask)
        out = h + self.feed_forward.forward(self.ffn_norm(h))
        return out

class Transformer(nn.Module):
    def __init__(self, params: ModelArgs):
        super().__init__()
        self.params = params
        self.vocab_size = params.vocab_size
        self.n_layers = params.n_layers

        self.tok_embeddings = ParallelEmbedding(
            params.vocab_size, params.dim, init_method=lambda x: x
        )

        self.layers = torch.nn.ModuleList()
        for layer_id in range(params.n_layers):
            self.layers.append(TransformerBlock(layer_id, params))

        self.norm = RMSNorm(params.dim, eps=params.norm_eps)
        self.output = ColumnParallelLinear(
            params.dim, params.vocab_size, bias=False, init_method=lambda x: x
        )

        self.freqs_cis = precompute_freqs_cis(
            self.params.dim // self.params.n_heads, self.params.max_seq_len * 2
        )

    @torch.inference_mode()
    def forward(self, tokens: torch.Tensor, start_pos: int):
        _bsz, seqlen = tokens.shape
        h = self.tok_embeddings(tokens)
        self.freqs_cis = self.freqs_cis.to(h.device)
        freqs_cis = self.freqs_cis[start_pos : start_pos + seqlen]

        mask = None
        if seqlen > 1:
            mask = torch.full((1, 1, seqlen, seqlen), float("-inf"), device=tokens.device)
            mask = torch.triu(mask, diagonal=start_pos + 1).type_as(h)

        for layer in self.layers:
            h = layer(h, start_pos, freqs_cis, mask)
        h = self.norm(h)
        output = self.output(h[:, -1, :])  # only compute last logits
        return output.float()

class LLaMA:
    def __init__(self, model: Transformer, tokenizer: Tokenizer):
        self.model = model
        self.tokenizer = tokenizer

    def generate(
        self,
        prompts: List[str],
        max_gen_len: int,
        temperature: float = 0.8,
        top_p: float = 0.95,
    ) -> List[str]:
        bsz = len(prompts)
        params = self.model.params
        assert bsz <= params.max_batch_size, (bsz, params.max_batch_size)

        prompt_tokens = [self.tokenizer.encode(x, bos=True, eos=False) for x in prompts]

        min_prompt_size = min([len(t) for t in prompt_tokens])
        max_prompt_size = max([len(t) for t in prompt_tokens])

        total_len = min(params.max_seq_len, max_gen_len + max_prompt_size)

        tokens = torch.full((bsz, total_len), self.tokenizer.pad_id).cuda().long()
        for k, t in enumerate(prompt_tokens):
            tokens[k, : len(t)] = torch.tensor(t).long()
        input_text_mask = tokens != self.tokenizer.pad_id
        start_pos = min_prompt_size
        prev_pos = 0
        for cur_pos in range(start_pos, total_len):
            logits = self.model.forward(tokens[:, prev_pos:cur_pos], prev_pos)
            if temperature > 0:
                probs = torch.softmax(logits / temperature, dim=-1)
                next_token = sample_top_p(probs, top_p)
            else:
                next_token = torch.argmax(logits, dim=-1)
            next_token = next_token.reshape(-1)
            # only replace token if prompt has already been generated
            next_token = torch.where(
                input_text_mask[:, cur_pos], tokens[:, cur_pos], next_token
            )
            tokens[:, cur_pos] = next_token
            prev_pos = cur_pos

        decoded = []
        for i, t in enumerate(tokens.tolist()):
            # cut to max gen len
            t = t[: len(prompt_tokens[i]) + max_gen_len]
            # cut to eos tok if any
            try:
                t = t[: t.index(self.tokenizer.eos_id)]
            except ValueError:
                pass
            decoded.append(self.tokenizer.decode(t))
        return decoded


def sample_top_p(probs, p):
    probs_sort, probs_idx = torch.sort(probs, dim=-1, descending=True)
    probs_sum = torch.cumsum(probs_sort, dim=-1)
    mask = probs_sum - probs_sort > p
    probs_sort[mask] = 0.0
    probs_sort.div_(probs_sort.sum(dim=-1, keepdim=True))
    next_token = torch.multinomial(probs_sort, num_samples=1)
    next_token = torch.gather(probs_idx, -1, next_token)
    return next_token