runtime/litgpt/generate/tp.py

"""Tensor-parallel implementation adapted from https://github.com/pytorch-labs/gpt-fast/blob/14df27/tp.py"""

import logging
import sys
import time
import warnings
from functools import partial
from pathlib import Path
from pprint import pprint
from typing import Literal

import lightning as L
import torch
import torch._dynamo.config
import torch._inductor.config
from lightning.fabric.plugins import BitsandbytesPrecision
from lightning.fabric.utilities import rank_zero_only

import litgpt.generate.base as generate_base
from litgpt.config import Config
from litgpt.constants import _BITANDBYTES_AVAILABLE_NOT_EQUAL_0_42_0
from litgpt.model import GPT, CausalSelfAttention, GptNeoxMLP, LLaMAMLP, LLaMAMoE
from litgpt.prompts import PromptStyle, has_prompt_style, load_prompt_style
from litgpt.tokenizer import Tokenizer
from litgpt.utils import (
    check_nvlink_connectivity,
    check_valid_checkpoint_dir,
    extend_checkpoint_dir,
    get_default_supported_precision,
)


def tensor_parallel_linear(fabric: L.Fabric, linear: torch.nn.Linear, style: str) -> None:
    world_size = fabric.world_size
    dim, attr = {"colwise": (0, "out_features"), "rowwise": (1, "in_features")}[style]
    size = getattr(linear, attr)
    if size % world_size != 0:
        raise ValueError(
            f"This linear's {attr} value ({size}) is not evenly divisible by the world size ({world_size})"
        )

    shard = torch.tensor_split(linear.weight, world_size, dim=dim)[fabric.global_rank]
    # overwrite `.data` instead of recreating the parameter for quantization (bitsandbytes) support.
    # the bitsandbytes linear classes use custom `torch.nn.Parameter` subclasses
    linear.weight.data = shard
    setattr(linear, attr, shard.size(dim))

    if linear.bias is not None and dim == 0:
        shard = torch.tensor_split(linear.bias, world_size)[fabric.global_rank]
        linear.bias = torch.nn.Parameter(shard, requires_grad=linear.bias.requires_grad)


def tensor_parallel_mlp(fabric: L.Fabric, mlp: GptNeoxMLP | LLaMAMLP | LLaMAMoE) -> None:
    if isinstance(mlp, LLaMAMLP):
        tensor_parallel_linear(fabric, mlp.fc_1, "colwise")
        tensor_parallel_linear(fabric, mlp.fc_2, "colwise")
        tensor_parallel_linear(fabric, mlp.proj, "rowwise")
        mlp.register_forward_hook(partial(all_reduce_output, fabric.world_size))
    elif isinstance(mlp, GptNeoxMLP):
        tensor_parallel_linear(fabric, mlp.fc, "colwise")
        tensor_parallel_linear(fabric, mlp.proj, "rowwise")
        mlp.register_forward_hook(partial(all_reduce_output, fabric.world_size))
    elif isinstance(mlp, LLaMAMoE):
        # we use expert slicing across ranks, alternatively, we could create a expert parallelism group
        # when the number of experts is a multiple of the world size
        for expert in mlp.experts:
            tensor_parallel_mlp(fabric, expert)
    else:
        raise NotImplementedError


def tensor_parallel_attn(fabric: L.Fabric, attn: CausalSelfAttention) -> None:
    tensor_parallel_linear(fabric, attn.qkv, "colwise")
    tensor_parallel_linear(fabric, attn.proj, "rowwise")
    attn.register_forward_hook(partial(all_reduce_output, fabric.world_size))


def all_reduce_output(world_size: int, module: torch.nn.Module, ins, outs) -> torch.Tensor:
    from torch.distributed._functional_collectives import all_reduce

    return all_reduce(outs, "sum", list(range(world_size)))


def tensor_parallel(fabric: L.Fabric, model: GPT) -> GPT:
    for block in model.transformer.h:
        tensor_parallel_mlp(fabric, block.mlp)
        tensor_parallel_attn(fabric, block.attn)

    # update the config values to the shard sizes
    # this is only relevant for `tensor_parallel_attn`, but it needs to run only once
    world_size = fabric.world_size
    attrs = ["n_head", "n_embd", "n_query_groups"]
    for attr in attrs:
        size = getattr(model.config, attr)
        if size % world_size != 0:
            raise ValueError(f"This {attr} value ({size}) is not evenly divisible by the world size ({world_size})")
        setattr(model.config, attr, size // world_size)

    return model


@torch.inference_mode()
def main(
    checkpoint_dir: Path,
    prompt: str = "What food do llamas eat?",
    *,
    sys_prompt: str | None = None,
    num_samples: int = 1,
    max_new_tokens: int = 50,
    top_k: int | None = 50,
    top_p: float = 1.0,
    temperature: float = 0.8,
    quantize: Literal["bnb.nf4", "bnb.nf4-dq", "bnb.fp4", "bnb.fp4-dq"] | None = None,
    precision: str | None = None,
    compile: bool = False,
) -> None:
    """Generation script that uses tensor parallelism to run across devices.

    Generates text samples based on a pre-trained model and tokenizer.

    Args:
        checkpoint_dir: The checkpoint directory to load.
        prompt: The prompt string to use for generating the samples.
        sys_prompt: The system prompt to use for generating the samples.
        num_samples: The number of text samples to generate.
        max_new_tokens: The number of generation steps to take.
        top_k: The number of top most probable tokens to consider in the sampling process.
        top_p: If specified, it represents the cumulative probability threshold to consider in the sampling process.
            In top-p sampling, the next token is sampled from the highest probability tokens
            whose cumulative probability exceeds the threshold `top_p`. When specified,
            it must be `0 <= top_p <= 1`. Here, `top_p=0` is equivalent
            to sampling the most probable token, while `top_p=1` samples from the whole distribution.
            It can be used in conjunction with `top_k` and `temperature` with the following order
            of application:

            1. `top_k` sampling
            2. `temperature` scaling
            3. `top_p` sampling

            For more details, see https://arxiv.org/abs/1904.09751
            or https://huyenchip.com/2024/01/16/sampling.html#top_p
        temperature: A value controlling the randomness of the sampling process. Higher values result in more random
            samples.
        quantize: Whether to quantize the model and using which method:
            - bnb.nf4, bnb.nf4-dq, bnb.fp4, bnb.fp4-dq: 4-bit quantization from bitsandbytes
            for more details, see https://github.com/Lightning-AI/litgpt/blob/main/tutorials/quantize.md
        precision: Indicates the Fabric precision setting to use.
        compile: Whether to compile the model.
    """
    checkpoint_dir = extend_checkpoint_dir(checkpoint_dir)
    pprint(locals())

    precision = precision or get_default_supported_precision(training=False)

    plugins = None
    if quantize is not None:
        if compile:
            raise NotImplementedError  # untested
        if "mixed" in precision:
            raise ValueError("Quantization and mixed precision is not supported.")
        if _BITANDBYTES_AVAILABLE_NOT_EQUAL_0_42_0:
            warnings.warn(
                "LitGPT only supports bitsandbytes v0.42.0. This may result in errors when using quantization."
            )
        dtype = {"16-true": torch.float16, "bf16-true": torch.bfloat16, "32-true": torch.float32}[precision]
        bnb_logger = logging.getLogger("lightning.fabric.plugins.precision.bitsandbytes")
        bnb_logger.setLevel(logging.DEBUG)
        bnb_logger.debug = rank_zero_only(bnb_logger.debug)
        plugins = BitsandbytesPrecision(quantize[4:], dtype)
        precision = None

    # set "ddp" as the strategy for the launching functionality, but there's no data-parallelism
    fabric = L.Fabric(devices="auto", strategy="ddp", precision=precision, plugins=plugins)
    if torch.cuda.is_available() and fabric.accelerator.auto_device_count() > 1:
        check_nvlink_connectivity(fabric)
    fabric.launch()

    check_valid_checkpoint_dir(checkpoint_dir)
    config = Config.from_file(checkpoint_dir / "model_config.yaml")

    model_file = "lit_model.pth"
    checkpoint_path = checkpoint_dir / model_file

    tokenizer = Tokenizer(checkpoint_dir)
    prompt_style = (
        load_prompt_style(checkpoint_dir) if has_prompt_style(checkpoint_dir) else PromptStyle.from_config(config)
    )
    prompt = prompt_style.apply(prompt, sys_prompt=sys_prompt)
    encoded = tokenizer.encode(prompt, device=fabric.device)
    prompt_length = encoded.size(0)
    max_returned_tokens = prompt_length + max_new_tokens

    fabric.print(f"Loading model {str(checkpoint_path)!r} with {config.__dict__}", file=sys.stderr)
    t0 = time.perf_counter()
    # cannot use `init_module` because if bitsandbytes is used, the Linear layers will be replaced
    # which means that the weights will get quantized on cuda:0 on checkpoint load. we need to load and then convert
    # still, use init_tensor for the precision
    with fabric.init_tensor(), torch.device("meta"):
        model = GPT(config)
    fabric.print(f"Time to instantiate model: {time.perf_counter() - t0:.02f} seconds.", file=sys.stderr)

    # sequentially do: load the checkpoint on CPU -> quantize -> apply tp -> move to device
    # so that the CPU RAM doesn't OOM with larger models
    for rank in range(fabric.world_size):
        if fabric.global_rank == rank:
            t0 = time.perf_counter()
            state_dict = torch.load(str(checkpoint_path), mmap=True, map_location="cpu")
            model.load_state_dict(state_dict, assign=True)
            print(f"[{rank}] Time to load the model weights: {time.perf_counter() - t0:.02f} seconds.", file=sys.stderr)

            # cannot use `.setup_module` because it will wrap with DDP
            model = fabric._precision.convert_module(model)

            t0 = time.perf_counter()
            model = tensor_parallel(fabric, model)
            print(
                f"[{rank}] Time to tensor-parallelize the model: {time.perf_counter() - t0:.02f} seconds.",
                file=sys.stderr,
            )

            with fabric.init_tensor():
                # set the max_seq_length to limit the memory usage to what we need
                model.max_seq_length = max_returned_tokens
                # the rope cache which is on meta device
                model.cos, model.sin = model.rope_cache()
                # enable the kv cache
                model.set_kv_cache(batch_size=1)
            model.eval()

            t0 = time.perf_counter()
            model = fabric.to_device(model)
            print(f"[{rank}] Time to move the model: {time.perf_counter() - t0:.02f} seconds.", file=sys.stderr)
        fabric.barrier()

    if compile:
        torch._dynamo.config.automatic_dynamic_shapes = True
        torch._inductor.config.triton.unique_kernel_names = True
        torch._inductor.config.coordinate_descent_tuning = True
        generate_base.next_token = torch.compile(generate_base.next_token, mode="reduce-overhead")

    L.seed_everything(1234)
    for i in range(num_samples):
        t0 = time.perf_counter()
        y = generate_base.generate(
            model, encoded, max_returned_tokens, temperature=temperature, top_k=top_k, eos_id=tokenizer.eos_id
        )
        t = time.perf_counter() - t0
        for block in model.transformer.h:
            block.attn.kv_cache.reset_parameters()
        fabric.print(tokenizer.decode(y))
        tokens_generated = y.size(0) - prompt_length
        fabric.print(
            f"Time for inference {i + 1}: {t:.02f} sec total, {tokens_generated / t:.02f} tokens/sec", file=sys.stderr
        )
    if fabric.device.type == "cuda":
        fabric.print(f"Memory used: {torch.cuda.max_memory_allocated() / 1e9:.02f} GB", file=sys.stderr)