llm.cli — CLI Tools

Command-line utilities registered under pyproject.toml [project.scripts] (llm-train lives in llm.training.train; the llm-serve entry point lives in llm.serving.api). Each module below is a thin Typer app.

Checkpoint migration

migrate_ckpt

llm-migrate-ckpt — convert legacy v0.0.5 .pt checkpoints to v2.

This CLI closes the gap introduced by the v2 split-layout format (ADR-006, commit 4b9cf68). Training checkpoints in v0.0.5 and earlier are a single torch.save blob; v2 splits them into three sidecars (<stem>.safetensors + <stem>.meta.json + <stem>.extra_state.pt).

The :class:~llm.training.core.checkpoint.CheckpointManager loader already auto-detects both layouts — legacy .pt files keep loading with a :class:DeprecationWarning. This CLI exists so users with a large fleet of v0.0.5 checkpoints can convert them once and silence the warning permanently.

Usage:

llm-migrate-ckpt path/to/latest.pt              # convert (keep legacy)
llm-migrate-ckpt path/to/latest                  # looks for latest.pt
llm-migrate-ckpt path/to/latest.pt --in-place    # delete legacy after success
llm-migrate-ckpt path/to/latest.pt --verify      # round-trip check (load new, compare)
llm-migrate-ckpt path/to/latest.pt --dry-run     # print plan, write nothing
llm-migrate-ckpt path/to/latest.pt --overwrite   # replace an existing split layout

Exit codes:

0 — conversion succeeded (or --dry-run plan was printed)
1 — conversion failed (legacy missing, split layout already present, etc.)
2 — verification failed (the new layout does not round-trip)

main

main(path=typer.Argument(..., help='Path to the legacy checkpoint — either `<name>.pt` or the stem `<name>` (the function appends `.pt`).', exists=False), in_place=typer.Option(False, '--in-place', help='Delete the legacy `.pt` file after a successful conversion. Off by default.'), verify=typer.Option(False, '--verify', help='Reload the new split layout and compare against the legacy blob. Exits 2 on mismatch.'), dry_run=typer.Option(False, '--dry-run', help='Print the conversion plan and exit without writing anything.'), overwrite=typer.Option(False, '--overwrite', help='Replace an existing split-layout trio at the same stem. Off by default.'))

Convert a legacy v0.0.5 .pt checkpoint to the v2 split layout.

源代码位于： src/llm/cli/migrate_ckpt.py

@app.command()
def main(
    path: Path = typer.Argument(
        ...,
        help="Path to the legacy checkpoint — either `<name>.pt` or the stem `<name>` (the function appends `.pt`).",
        exists=False,
    ),
    in_place: bool = typer.Option(
        False,
        "--in-place",
        help="Delete the legacy `.pt` file after a successful conversion. Off by default.",
    ),
    verify: bool = typer.Option(
        False,
        "--verify",
        help="Reload the new split layout and compare against the legacy blob. Exits 2 on mismatch.",
    ),
    dry_run: bool = typer.Option(
        False,
        "--dry-run",
        help="Print the conversion plan and exit without writing anything.",
    ),
    overwrite: bool = typer.Option(
        False,
        "--overwrite",
        help="Replace an existing split-layout trio at the same stem. Off by default.",
    ),
) -> None:
    """Convert a legacy v0.0.5 ``.pt`` checkpoint to the v2 split layout."""
    # Normalize the path: a stem is rewritten to `<stem>.pt` for the
    # resolution step below. The actual write goes to the stem (no
    # suffix), matching the v2 layout convention.
    legacy_path = path if path.suffix == ".pt" else path.with_suffix(".pt")

    if not legacy_path.exists():
        typer.echo(f"error: legacy checkpoint not found: {legacy_path}", err=True)
        raise typer.Exit(code=1)

    # A directory named ``foo.pt`` would pass the exists() gate, then
    # ``torch.load(directory)`` raised an IsADirectoryError raw traceback
    # (RIL ISS-213); mirror quantize's ``_validate_model_path`` which checks
    # ``is_file()`` so the error is the documented one-line exit-1 form.
    if not legacy_path.is_file():
        typer.echo(f"error: legacy checkpoint is not a regular file: {legacy_path}", err=True)
        raise typer.Exit(code=1)

    stem = legacy_path.with_suffix("")
    sidecars = {
        "weights": stem.with_suffix(".safetensors"),
        "meta": stem.with_name(stem.name + ".meta.json"),
        "extra_state": stem.with_name(stem.name + ".extra_state.pt"),
    }

    if dry_run:
        typer.echo("[dry-run] conversion plan:")
        _print_plan(legacy_path, sidecars, in_place=in_place)
        raise typer.Exit(code=0)

    # If --verify is passed AND the split layout already exists at the
    # same stem, skip the conversion step and verify the existing trio
    # against the legacy blob. This is the post-conversion re-check
    # workflow ("I converted yesterday, did anything drift?").
    split_exists = all(p.exists() for p in sidecars.values())
    if verify and split_exists:
        typer.echo("✓ Split layout already exists; running --verify on it.")
        written = sidecars
    else:
        # When --verify is requested, do NOT delete the legacy blob at
        # convert time even with --in-place: verification compares the
        # new trio against the legacy file, so it must still be present
        # on disk until the round-trip check runs. The legacy file is
        # unlinked below, but only after verification has passed.
        try:
            written = convert_legacy_checkpoint_to_split(
                legacy_path,
                in_place=in_place and not verify,
                overwrite=overwrite,
            )
        except CheckpointMigrationError as exc:
            typer.echo(f"error: {exc}", err=True)
            raise typer.Exit(code=1) from exc

        typer.echo("✓ Converted:")
        _print_plan(legacy_path, written, in_place=in_place)

    if verify:
        ok, msg = _verify_round_trip(legacy_path, written)
        if not ok:
            typer.echo(f"✗ verification failed: {msg}", err=True)
            # The split layout was written but does not round-trip —
            # leave the legacy file in place so the user can retry.
            typer.echo(
                f"  hint: the legacy {legacy_path} is preserved; investigate the mismatch before retrying.",
                err=True,
            )
            raise typer.Exit(code=2)
        # Honor --in-place now that verification passed: the round-trip
        # compare above used the legacy blob, so it is safe to delete.
        if in_place and legacy_path.exists():
            legacy_path.unlink()
        typer.echo("✓ verification passed (model_state tensors + metadata match)")

Quantization

quantize

llm-quantize CLI — currently supports the gptq subcommand.

This CLI closes the gap between the Python API (:func:llm.quantization.gptq.quantize_model_gptq) and the command line. The same validation rules that the Python API enforces (GPTQConfig __post_init__) are also enforced here so users get early, clear errors on bad input rather than a stack trace halfway through Hessian accumulation.

Subcommand surface (matches docs/superpowers/plans/2026-07-22-gptq-integration.md § Task 10):

llm-quantize gptq \
    --model PATH                 # torch.save blob with a DecoderModel \
    --output PATH                # where to write the quantized model \
    --calib-data PATH            # raw text (one sample per line) — needs --tokenizer \
    --calib-data-tokens PATH     # pre-tokenized tensor file — mutually exclusive with --calib-data \
    --tokenizer PATH             # HF tokenizer dir; required with --calib-data \
    --bits {4,8}                 # default 4 \
    --group-size N|-1            # default 128; -1 = per-channel \
    [--sym|--asym]               # default sym (--asym rejected: not implemented) \
    [--act-order|--no-act-order] # default off \
    --percdamp F                 # default 0.01 \
    --blocksize N                # default 128 \
    --target-modules m1,m2,...   # default: all nn.Linear layers

Exit codes:

0 — quantization succeeded
1 — argument validation failed (bad bits, missing tokenizer, etc.)
2 — runtime failure (model load, tokenization, etc.)
(Note: typer itself exits 2 — the same as a runtime failure — for
argparse-level usage errors like a missing required --model/--output,
so a bare mis-typed invocation is not distinguishable from an in-run
crash by exit code alone; the stderr message differs.)

gptq

gptq(model=typer.Option(..., '--model', help='Path to model checkpoint (.pt with DecoderModel state_dict).'), output=typer.Option(..., '--output', help='Output path for quantized model (torch.save blob).'), calib_data=typer.Option(None, '--calib-data', help='Path to raw text file (one sample per line). Requires --tokenizer.'), calib_data_tokens=typer.Option(None, '--calib-data-tokens', help='Path to pre-tokenized .pt file (tensor or list of tensors). Mutually exclusive with --calib-data.'), tokenizer=typer.Option(None, '--tokenizer', help='Path to HF tokenizer (required when --calib-data is set).'), bits=typer.Option(4, '--bits', help='Quantization bit width (4 or 8).'), group_size=typer.Option(128, '--group-size', help='Group size for per-group scales (-1 = per-channel).'), sym=typer.Option(True, '--sym/--asym', help='Symmetric (default) vs asymmetric quantization.'), percdamp=typer.Option(0.01, '--percdamp', help='Hessian damping fraction (must be in (0, 1)).'), blocksize=typer.Option(128, '--blocksize', help='Column block size for the GPTQ outer loop.'), act_order=typer.Option(False, '--act-order/--no-act-order', help='Sort weight columns by diag(H) descending (better accuracy, slower).'), target_modules=typer.Option(None, '--target-modules', help='Comma-separated layer names to quantize (default: all nn.Linear).'))

Quantize a model with GPTQ (Frantar 2022, arXiv:2210.17323).

源代码位于： src/llm/cli/quantize.py

@app.command()
def gptq(
    model: Path = typer.Option(
        ...,
        "--model",
        help="Path to model checkpoint (.pt with DecoderModel state_dict).",
    ),
    output: Path = typer.Option(
        ...,
        "--output",
        help="Output path for quantized model (torch.save blob).",
    ),
    calib_data: Path | None = typer.Option(
        None,
        "--calib-data",
        help="Path to raw text file (one sample per line). Requires --tokenizer.",
    ),
    calib_data_tokens: Path | None = typer.Option(
        None,
        "--calib-data-tokens",
        help="Path to pre-tokenized .pt file (tensor or list of tensors). Mutually exclusive with --calib-data.",
    ),
    tokenizer: Path | None = typer.Option(
        None,
        "--tokenizer",
        help="Path to HF tokenizer (required when --calib-data is set).",
    ),
    bits: int = typer.Option(
        4,
        "--bits",
        help="Quantization bit width (4 or 8).",
    ),
    group_size: int = typer.Option(
        128,
        "--group-size",
        help="Group size for per-group scales (-1 = per-channel).",
    ),
    sym: bool = typer.Option(
        True,
        "--sym/--asym",
        help="Symmetric (default) vs asymmetric quantization.",
    ),
    percdamp: float = typer.Option(
        0.01,
        "--percdamp",
        help="Hessian damping fraction (must be in (0, 1)).",
    ),
    blocksize: int = typer.Option(
        128,
        "--blocksize",
        help="Column block size for the GPTQ outer loop.",
    ),
    act_order: bool = typer.Option(
        False,
        "--act-order/--no-act-order",
        help="Sort weight columns by diag(H) descending (better accuracy, slower).",
    ),
    target_modules: str | None = typer.Option(
        None,
        "--target-modules",
        help="Comma-separated layer names to quantize (default: all nn.Linear).",
    ),
) -> None:
    """Quantize a model with GPTQ (Frantar 2022, arXiv:2210.17323)."""
    # --- 1. Argument validation (fail fast, exit 1) ---------------------
    _validate_quant_params(bits, group_size, percdamp, blocksize, sym=sym)
    _validate_calib_inputs(calib_data, calib_data_tokens, tokenizer)
    _validate_model_path(model)
    _reject_clobbering_input(output, model)

    # --- 2. Late imports (keep startup cheap) ---------------------------
    import torch

    from llm.quantization.gptq import GPTQConfig, quantize_model_gptq

    # --- 3. Load inputs -------------------------------------------------
    try:
        typer.echo(f"Loading model from {model}...")
        # Framework model/quantization classes are allowlisted and loaded with
        # ``weights_only=True`` (RIL ISS-211), closing the arbitrary-pickle RCE
        # that ``weights_only=False`` left open on a user-supplied model file
        # (a shared/community ``.pt`` can smuggle a ``__reduce__``). A pickle
        # referencing anything outside the framework + torch.nn allowlist is
        # REFUSED — matching the hardened serving loader (ISS-170). Callers
        # with a genuinely custom non-framework model class register it via
        # ``torch.serialization.add_safe_globals`` before invoking the CLI.
        from llm.utils.serialization import register_framework_safe_globals

        register_framework_safe_globals()
        model_obj = torch.load(model, map_location="cpu", weights_only=True)
    except Exception as exc:
        typer.echo(f"Error: failed to load model {model}: {exc}", err=True)
        raise typer.Exit(code=2) from exc

    try:
        batches = _load_calibration_batches(calib_data, calib_data_tokens, tokenizer)
    except Exception as exc:
        typer.echo(f"Error: failed to load calibration data: {exc}", err=True)
        raise typer.Exit(code=2) from exc

    # --- 4. Build config + run quantization -----------------------------
    target_list = _resolve_target_modules(target_modules)

    config = GPTQConfig(
        bits=bits,
        group_size=group_size,
        sym=sym,
        percdamp=percdamp,
        blocksize=blocksize,
        act_order=act_order,
    )

    typer.echo(
        f"Quantizing model with GPTQ (bits={bits}, group_size={group_size}, "
        f"sym={sym}, act_order={act_order}, target_modules={target_list or 'all'})..."
    )
    try:
        quantized = quantize_model_gptq(model_obj, iter(batches), config, target_list)
    except Exception as exc:
        typer.echo(f"Error: quantization failed: {exc}", err=True)
        raise typer.Exit(code=2) from exc

    # --- 5. Save --------------------------------------------------------
    try:
        _atomic_save_quantized(quantized, output)
    except Exception as exc:
        typer.echo(f"Error: failed to save quantized model to {output}: {exc}", err=True)
        raise typer.Exit(code=2) from exc

    typer.echo(f"Quantized model saved to {output}")

fp8

fp8(model=typer.Option(..., '--model', help='Path to model checkpoint (.pt with DecoderModel state_dict).'), output=typer.Option(..., '--output', help='Output path for quantized model (torch.save blob).'), weight_dtype=typer.Option('e4m3', '--weight-dtype', help="FP8 weight format: 'e4m3' (E4M3FN, default) or 'e5m2' (wider range)."), per_channel=typer.Option(True, '--per-channel/--per-tensor', help='Per-output-row (default) vs per-tensor weight scaling.'), activation=typer.Option('dynamic', '--activation', help="Activation scaling: 'dynamic' (per forward; default, no calibration needed) or 'static' (from calibration)."), calib_data=typer.Option(None, '--calib-data', help='Path to raw text file (one sample per line). Requires --tokenizer. Needed only for static activation.'), calib_data_tokens=typer.Option(None, '--calib-data-tokens', help='Path to pre-tokenized .pt file (tensor or list of tensors). Needed only for static activation.'), tokenizer=typer.Option(None, '--tokenizer', help='Path to HF tokenizer (required when --calib-data is set).'), target_modules=typer.Option(None, '--target-modules', help='Comma-separated layer names to quantize (default: all nn.Linear).'))

Quantize a model's Linear weights + activations with FP8 (E4M3/E5M2).

Stores real float8 weights (1 byte/weight) and simulates the fp8 matmul in fp32. Static activation scaling captures per-layer abs-max over the calibration batches; dynamic computes it per forward and needs no calibration data.

源代码位于： src/llm/cli/quantize.py

@app.command()
def fp8(
    model: Path = typer.Option(
        ...,
        "--model",
        help="Path to model checkpoint (.pt with DecoderModel state_dict).",
    ),
    output: Path = typer.Option(
        ...,
        "--output",
        help="Output path for quantized model (torch.save blob).",
    ),
    weight_dtype: str = typer.Option(
        "e4m3",
        "--weight-dtype",
        help="FP8 weight format: 'e4m3' (E4M3FN, default) or 'e5m2' (wider range).",
    ),
    per_channel: bool = typer.Option(
        True,
        "--per-channel/--per-tensor",
        help="Per-output-row (default) vs per-tensor weight scaling.",
    ),
    activation: str = typer.Option(
        "dynamic",
        "--activation",
        help="Activation scaling: 'dynamic' (per forward; default, no calibration needed) or 'static' (from calibration).",
    ),
    calib_data: Path | None = typer.Option(
        None,
        "--calib-data",
        help="Path to raw text file (one sample per line). Requires --tokenizer. Needed only for static activation.",
    ),
    calib_data_tokens: Path | None = typer.Option(
        None,
        "--calib-data-tokens",
        help="Path to pre-tokenized .pt file (tensor or list of tensors). Needed only for static activation.",
    ),
    tokenizer: Path | None = typer.Option(
        None,
        "--tokenizer",
        help="Path to HF tokenizer (required when --calib-data is set).",
    ),
    target_modules: str | None = typer.Option(
        None,
        "--target-modules",
        help="Comma-separated layer names to quantize (default: all nn.Linear).",
    ),
) -> None:
    """Quantize a model's Linear weights + activations with FP8 (E4M3/E5M2).

    Stores real float8 weights (1 byte/weight) and simulates the fp8 matmul
    in fp32. Static activation scaling captures per-layer abs-max over the
    calibration batches; dynamic computes it per forward and needs no
    calibration data.
    """
    if weight_dtype not in ("e4m3", "e5m2"):
        _die(f"--weight-dtype must be 'e4m3' or 'e5m2'; got {weight_dtype}.")
    if activation not in ("static", "dynamic"):
        _die(f"--activation must be 'static' or 'dynamic'; got {activation}.")

    if activation == "static":
        if calib_data is None and calib_data_tokens is None:
            # Static is opt-in now (dynamic is the default), so when the user
            # asks for it without data the message should name the only thing
            # they need to add (a calib source) AND the escape hatch (RIL
            # ISS-330): ``--activation dynamic`` needs no calibration at all.
            _die(
                "activation='static' requires calibration data via --calib-data/"
                "--calib-data-tokens; use --activation dynamic to quantize "
                "without calibration."
            )
        _validate_calib_inputs(calib_data, calib_data_tokens, tokenizer)
    elif calib_data is not None or calib_data_tokens is not None or tokenizer is not None:
        # Calibration is ignored under dynamic activation — refuse rather than
        # silently drop the user's data.
        _die("activation='dynamic' needs no calibration; drop --calib-data / --calib-data-tokens / --tokenizer.")
    _validate_model_path(model)
    _reject_clobbering_input(output, model)

    import torch

    from llm.quantization.fp8 import quantize_model_fp8
    from llm.utils.serialization import register_framework_safe_globals

    try:
        register_framework_safe_globals()
        model_obj = torch.load(model, map_location="cpu", weights_only=True)
    except Exception as exc:
        typer.echo(f"Error: failed to load model {model}: {exc}", err=True)
        raise typer.Exit(code=2) from exc

    batches = None
    if activation == "static":
        try:
            batches = _load_calibration_batches(calib_data, calib_data_tokens, tokenizer)
        except Exception as exc:
            typer.echo(f"Error: failed to load calibration data: {exc}", err=True)
            raise typer.Exit(code=2) from exc
    else:
        # Dynamic activation: quantize_model_fp8 accepts None calib_iter.
        batches = None

    target_list = _resolve_target_modules(target_modules)
    typer.echo(
        f"Quantizing model with FP8 (weight_dtype={weight_dtype}, "
        f"per_channel={per_channel}, activation={activation}, "
        f"target_modules={target_list or 'all'})..."
    )
    try:
        from llm.quantization.fp8 import Fp8Config

        fp8_cfg = Fp8Config(
            weight_dtype=cast(Literal["e4m3", "e5m2"], weight_dtype),
            per_channel=per_channel,
            activation=cast(Literal["static", "dynamic"], activation),
        )
        quantized = quantize_model_fp8(
            model_obj,
            iter(batches) if batches is not None else None,
            fp8_cfg,
            target_modules=target_list,
        )
    except Exception as exc:
        typer.echo(f"Error: quantization failed: {exc}", err=True)
        raise typer.Exit(code=2) from exc

    try:
        _atomic_save_quantized(quantized, output)
    except Exception as exc:
        typer.echo(f"Error: failed to save quantized model to {output}: {exc}", err=True)
        raise typer.Exit(code=2) from exc

    typer.echo(f"Quantized model saved to {output}")

main

main()

Entry point for llm-quantize = llm.cli.quantize:main in pyproject.toml.

源代码位于： src/llm/cli/quantize.py

def main() -> None:
    """Entry point for ``llm-quantize = llm.cli.quantize:main`` in pyproject.toml."""
    app()

Pruning

prune

llm-prune — weight-prune a pretrained model checkpoint.

Thin wrapper over :func:llm.quantization.prune.prune_model (the sibling of the llm-quantize CLI). Loads a bare torch.save model blob (a DecoderModel, plain or already quantized), zeroes a fraction of each nn.Linear's weights via a persistent weight_mask, and writes a new pruned blob atomically.

Exit codes:

0 — success (sparsity reported)
1 — argument validation failed (bad ratio / method / clobbered output)
2 — runtime failure (model load, pruning, save)
(Note: typer itself exits 2 — the same as a runtime failure — for
argparse-level usage errors like a missing required --model/--output,
so a bare mis-typed invocation is not distinguishable from an in-run
crash by exit code alone; the stderr message differs.)

prune

prune(model=typer.Option(..., '--model', help='Path to model blob (.pt torch.save of a DecoderModel).'), output=typer.Option(..., '--output', help='Output path for the pruned model blob.'), ratio=typer.Option(0.5, '--ratio', help="Fraction of each Linear's weights to zero (0 < ratio < 1)."), method=typer.Option('magnitude', '--method', help="'magnitude' (keep largest |W|) or 'random'."), target_modules=typer.Option(None, '--target-modules', help='Comma-separated module-name substrings to prune (default: all Linear).'), seed=typer.Option(None, '--seed', help="Seed for 'random' pruning (reproducibility)."))

Prune a pretrained model's linear weights and save a new blob.

源代码位于： src/llm/cli/prune.py

@app.command()
def prune(
    model: Path = typer.Option(..., "--model", help="Path to model blob (.pt torch.save of a DecoderModel)."),
    output: Path = typer.Option(..., "--output", help="Output path for the pruned model blob."),
    ratio: float = typer.Option(0.5, "--ratio", help="Fraction of each Linear's weights to zero (0 < ratio < 1)."),
    method: str = typer.Option("magnitude", "--method", help="'magnitude' (keep largest |W|) or 'random'."),
    target_modules: str | None = typer.Option(
        None, "--target-modules", help="Comma-separated module-name substrings to prune (default: all Linear)."
    ),
    seed: int | None = typer.Option(None, "--seed", help="Seed for 'random' pruning (reproducibility)."),
) -> None:
    """Prune a pretrained model's linear weights and save a new blob."""
    _validate_ratio(ratio)
    _validate_method(method)
    if method == "magnitude" and seed is not None:
        # Only "random" pruning consumes the seed (RIL ISS-335); passing one
        # with magnitude silently does nothing — surface it so the user isn't
        # misled into thinking the run is reproducible.
        typer.echo(
            f"Warning: --seed {seed} is ignored with --method magnitude (only 'random' pruning uses the seed).",
            err=True,
        )
    _validate_model_path(model)
    _reject_clobbering_input(output, model)

    import torch

    from llm.quantization.prune import PruningConfig, prune_model
    from llm.utils.serialization import register_framework_safe_globals

    try:
        typer.echo(f"Loading model from {model}...")
        register_framework_safe_globals()
        model_obj = torch.load(model, map_location="cpu", weights_only=True)
    except Exception as exc:
        typer.echo(f"Error: failed to load model {model}: {exc}", err=True)
        raise typer.Exit(code=2) from exc

    config = PruningConfig(
        ratio=ratio,
        method=method,
        target_modules=_resolve_target_modules(target_modules),
        random_seed=seed,
    )
    try:
        typer.echo(f"Pruning {method} ratio={ratio} target_modules={config.target_modules or 'all'}...")
        sparsity = prune_model(model_obj, config)
    except Exception as exc:
        typer.echo(f"Error: pruning failed: {exc}", err=True)
        raise typer.Exit(code=2) from exc

    try:
        _atomic_save_blob(model_obj, output)
    except Exception as exc:
        typer.echo(f"Error: failed to save pruned model to {output}: {exc}", err=True)
        raise typer.Exit(code=2) from exc

    typer.echo(f"Pruned model saved to {output}")
    typer.echo(f"Achieved sparsity: {sparsity:.2%}")

Low-rank decomposition

decompose

llm-decompose — low-rank (SVD U-V) factorize a pretrained model.

Thin wrapper over :func:llm.quantization.lowrank.decompose_model (the sibling of llm-prune in the compression family). Loads a bare torch.save model blob, factorizes each nn.Linear weight into u @ v at the requested rank, and writes a new blob atomically, reporting rank / compression ratio / Frobenius reconstruction error.

Exit codes:

0 — success
1 — argument validation failed (bad/conflicting rank, clobbered output)
2 — runtime failure (model load, decomposition, save)
(Note: typer itself exits 2 — the same as a runtime failure — for
argparse-level usage errors like a missing required --model/--output,
so a bare mis-typed invocation is not distinguishable from an in-run
crash by exit code alone; the stderr message differs.)

decompose

decompose(model=typer.Option(..., '--model', help='Path to model blob (.pt torch.save of a DecoderModel).'), output=typer.Option(..., '--output', help='Output path for the low-rank model blob.'), rank=typer.Option(None, '--rank', help='Explicit rank r (mutually exclusive with --rank-ratio).'), rank_ratio=typer.Option(None, '--rank-ratio', help='Auto rank = ratio * min(out, in) (mutually exclusive with --rank).'), target_modules=typer.Option(None, '--target-modules', help='Comma-separated module-name substrings to decompose (default: all Linear).'))

Low-rank factorize a pretrained model's linear weights.

源代码位于： src/llm/cli/decompose.py

@app.command()
def decompose(
    model: Path = typer.Option(..., "--model", help="Path to model blob (.pt torch.save of a DecoderModel)."),
    output: Path = typer.Option(..., "--output", help="Output path for the low-rank model blob."),
    rank: int | None = typer.Option(None, "--rank", help="Explicit rank r (mutually exclusive with --rank-ratio)."),
    rank_ratio: float | None = typer.Option(
        None, "--rank-ratio", help="Auto rank = ratio * min(out, in) (mutually exclusive with --rank)."
    ),
    target_modules: str | None = typer.Option(
        None, "--target-modules", help="Comma-separated module-name substrings to decompose (default: all Linear)."
    ),
) -> None:
    """Low-rank factorize a pretrained model's linear weights."""
    if (rank is None) == (rank_ratio is None):
        _die("must supply exactly one of --rank or --rank-ratio.")
    if rank is not None and rank <= 0:
        _die(f"--rank must be > 0; got {rank}.")
    if rank_ratio is not None and not 0.0 < rank_ratio <= 1.0:
        _die(f"--rank-ratio must be in (0, 1]; got {rank_ratio}.")
    _validate_model_path(model)
    _reject_clobbering_input(output, model)

    import torch

    from llm.quantization.lowrank import LowRankConfig, decompose_model
    from llm.utils.serialization import register_framework_safe_globals

    try:
        typer.echo(f"Loading model from {model}...")
        register_framework_safe_globals()
        model_obj = torch.load(model, map_location="cpu", weights_only=True)
    except Exception as exc:
        typer.echo(f"Error: failed to load model {model}: {exc}", err=True)
        raise typer.Exit(code=2) from exc

    config = LowRankConfig(
        rank=rank,
        rank_ratio=rank_ratio,
        target_modules=_resolve_target_modules(target_modules),
    )
    try:
        stats = decompose_model(model_obj, config)
    except Exception as exc:
        typer.echo(f"Error: decomposition failed: {exc}", err=True)
        raise typer.Exit(code=2) from exc

    try:
        _atomic_save_blob(model_obj, output)
    except Exception as exc:
        typer.echo(f"Error: failed to save low-rank model to {output}: {exc}", err=True)
        raise typer.Exit(code=2) from exc

    typer.echo(f"Low-rank model saved to {output}")
    typer.echo(f"Compression ratio: {stats['compression_ratio']:.3f}x")
    typer.echo(f"Mean reconstruction error: {stats['relative_error']:.4f}")
    typer.echo(f"Layers decomposed: {len(stats['layers'])}")