Ruurd's picture
Deploy BYOD-Llama-3.1-8B full-precision demo
3329d00 verified
Raw
History Blame Contribute Delete
56.1 kB
"""Interactive iterative denoising inference for saved LoRA adapters."""
from __future__ import annotations
import gc
from html import escape
import json
import os
from dataclasses import dataclass
from pathlib import Path
from typing import Any, Callable
import torch
import torch.nn.functional as F
from peft import PeftModel
from transformers import AutoModel, AutoModelForCausalLM, AutoTokenizer
from .data import validate_mask_token
from .legacy_compat import install_legacy_pickle_modules, patch_legacy_lora_modules, restore_legacy_pickle_modules
from .modeling import forward_bidirectional
def find_adapters(outputs_dir: str | Path = "outputs") -> list[str]:
"""Return adapter directories relative to outputs_dir, newest first."""
root = Path(outputs_dir).resolve()
if not root.exists():
return []
paths = [path for path in root.rglob("adapter_config.json") if path.parent.is_dir()]
return [str(path.parent.relative_to(root)) for path in sorted(paths, key=lambda p: p.stat().st_mtime, reverse=True)]
def _safe_adapter_path(outputs_dir: str | Path, selection: str) -> Path:
"""Resolve a selected adapter while preventing paths outside outputs_dir."""
root = Path(outputs_dir).resolve()
path = (root / selection).resolve()
if root not in path.parents or not (path / "adapter_config.json").is_file():
raise ValueError("Select a valid adapter directory below outputs/.")
return path
def _precision_dtype(precision: str, device: torch.device) -> torch.dtype:
"""Map configured precision to a safe dtype for the selected device."""
requested = {"fp16": torch.float16, "bf16": torch.bfloat16, "fp32": torch.float32}.get(precision, torch.float32)
# CPU inference with low-precision weights is not generally supported; MPS
# has better float32 compatibility for interactive single-request inference.
if device.type == "cpu":
return torch.float32
# T4-class CUDA GPUs have no native BF16 Tensor Core support. BF16
# quantized compute there is substantially slower than FP16, so retain the
# saved run's preference only where the hardware can execute it natively.
if device.type == "cuda" and requested == torch.bfloat16 and not torch.cuda.is_bf16_supported():
return torch.float16
return requested
def select_device(requested: str = "auto") -> torch.device:
"""Choose an available CUDA, MPS, or CPU device from a UI selection."""
if requested == "auto":
requested = "cuda" if torch.cuda.is_available() else "mps" if torch.backends.mps.is_available() else "cpu"
if requested == "cuda" and not torch.cuda.is_available():
raise ValueError("CUDA was requested but is unavailable.")
if requested == "mps" and not torch.backends.mps.is_available():
raise ValueError("MPS was requested but is unavailable.")
return torch.device(requested)
@dataclass
class InferenceSession:
model: torch.nn.Module
tokenizer: Any
device: torch.device
adapter_path: Path
config: dict[str, Any]
mask_token_id: int
quantization: str = "none"
compute_dtype: str = "unknown"
legacy_wrapper: bool = False
prompt_format: str = "chat_template"
llada: bool = False
def _load_adapter_path(
adapter_path: str | Path,
device_name: str = "auto",
quantization: str | None = None,
preflight: bool = True,
) -> InferenceSession:
"""Load an adapter directory that has already been resolved and validated."""
adapter_path = Path(adapter_path).expanduser().resolve()
if not (adapter_path / "adapter_config.json").is_file():
raise ValueError(f"Adapter directory has no adapter_config.json: {adapter_path}")
config_candidates = (
adapter_path / "resolved_config.json",
adapter_path / "lad_run_config.json",
adapter_path.parent / "resolved_config.json",
)
run_config_path = next((path for path in config_candidates if path.is_file()), None)
run_config = json.loads(run_config_path.read_text()) if run_config_path else {}
adapter_config = json.loads((adapter_path / "adapter_config.json").read_text())
base_model = adapter_config["base_model_name_or_path"]
device = select_device(device_name)
dtype = _precision_dtype(run_config.get("precision", "fp32"), device)
requested_quantization = str(quantization or "auto").lower()
resolved_quantization = str(run_config.get("quantization", "none") if requested_quantization == "auto" else requested_quantization).lower()
if resolved_quantization in {"4-bit", "qlora"}:
resolved_quantization = "4bit"
if resolved_quantization not in {"none", "off", "false", "4bit"}:
raise ValueError("Inference quantization must be 'auto', 'none', or '4bit'.")
use_4bit = resolved_quantization == "4bit"
compute_dtype = dtype
cache_dir = run_config.get("base_model_cache_dir", "base_models")
token = os.getenv("HF_TOKEN")
tokenizer_name = run_config.get("tokenizer_name_or_path", base_model)
tokenizer_kwargs: dict[str, Any] = {}
if "mistral" in str(tokenizer_name).lower():
tokenizer_kwargs["fix_mistral_regex"] = True
tokenizer = AutoTokenizer.from_pretrained(
tokenizer_name,
use_fast=True,
token=token,
cache_dir=cache_dir,
clean_up_tokenization_spaces=False,
**tokenizer_kwargs,
)
if tokenizer.pad_token_id is None:
tokenizer.pad_token = tokenizer.eos_token
load_kwargs: dict[str, Any] = dict(torch_dtype=dtype, token=token, cache_dir=cache_dir, trust_remote_code=False)
if use_4bit:
if device.type != "cuda":
raise RuntimeError("4-bit inference requires an NVIDIA CUDA device.")
try:
from transformers import BitsAndBytesConfig
import bitsandbytes # noqa: F401
except ImportError as exc:
raise ImportError("4-bit inference requires bitsandbytes; install with `pip install -e '.[cuda]'`.") from exc
compute_dtype = _precision_dtype(run_config.get("compute_dtype", run_config.get("precision", "bf16")), device)
load_kwargs["quantization_config"] = BitsAndBytesConfig(
load_in_4bit=True,
bnb_4bit_quant_type=str(run_config.get("quantization_type", "nf4")),
bnb_4bit_compute_dtype=compute_dtype,
bnb_4bit_use_double_quant=bool(run_config.get("double_quant", True)),
)
# Quantized modules cannot subsequently be moved with model.to().
load_kwargs["device_map"] = {"": device.index if device.index is not None else 0}
base = AutoModelForCausalLM.from_pretrained(base_model, **load_kwargs)
base.config.use_cache = False
base.config.is_causal = False
if hasattr(base.config, "use_bidirectional_attention"):
base.config.use_bidirectional_attention = True
adapter_load_kwargs: dict[str, Any] = {}
if not use_4bit:
# Stage adapter tensors on CPU before moving the assembled root module.
# This is required by ZeroGPU, where CUDA is represented by a proxy at
# module-import time but no physical GPU has been allocated yet.
adapter_load_kwargs["device_map"] = {"": "cpu"}
adapter_load_kwargs["torch_device"] = "cpu"
model = PeftModel.from_pretrained(
base,
adapter_path,
is_trainable=False,
**adapter_load_kwargs,
)
norm_path = adapter_path / "normalization_state.pt"
if norm_path.is_file():
model.load_state_dict(torch.load(norm_path, map_location="cpu", weights_only=True), strict=False)
if not use_4bit:
model.to(device)
model.eval()
mask_info = validate_mask_token(tokenizer, str(run_config.get("mask_token", "MASK")))
session = InferenceSession(model, tokenizer, device, adapter_path, run_config, mask_info["mask_token_id"], "4bit" if use_4bit else "none", str(compute_dtype).removeprefix("torch."))
if preflight:
preflight_session(session)
return session
def load_session(adapter_selection: str, outputs_dir: str | Path = "outputs", device_name: str = "auto", quantization: str | None = None) -> InferenceSession:
"""Load a base model, saved LoRA adapter, tokenizer, and norm state."""
adapter_path = _safe_adapter_path(outputs_dir, adapter_selection)
return _load_adapter_path(adapter_path, device_name, quantization)
def load_hub_adapter_session(
repo_id: str,
device_name: str = "auto",
quantization: str | None = None,
revision: str | None = None,
cache_dir: str | Path | None = None,
preflight: bool = True,
) -> InferenceSession:
"""Download and load a BYOD adapter from the Hugging Face Hub."""
try:
from huggingface_hub import snapshot_download
except ImportError as exc:
raise ImportError(
"Hub inference requires huggingface_hub; install it with `pip install huggingface-hub`."
) from exc
adapter_path = snapshot_download(
repo_id=repo_id,
repo_type="model",
revision=revision,
cache_dir=str(cache_dir) if cache_dir is not None else None,
token=os.getenv("HF_TOKEN"),
)
return _load_adapter_path(adapter_path, device_name, quantization, preflight=preflight)
def load_merged_session(
model_path: str | Path,
device_name: str = "auto",
quantization: str | None = None,
source_config: dict[str, Any] | None = None,
) -> InferenceSession:
"""Load a standalone model produced by ``merge_adapter.py``."""
model_path = Path(model_path).expanduser().resolve()
if not (model_path / "config.json").is_file():
raise ValueError(f"Merged model directory has no config.json: {model_path}")
run_config = dict(source_config or {})
saved_run_config = model_path / "lad_run_config.json"
if not run_config and saved_run_config.is_file():
run_config = json.loads(saved_run_config.read_text())
device = select_device(device_name)
dtype = _precision_dtype(run_config.get("precision", "bf16"), device)
requested_quantization = str(quantization or "none").lower()
if requested_quantization == "auto":
requested_quantization = "none"
if requested_quantization in {"4-bit", "qlora"}:
requested_quantization = "4bit"
if requested_quantization not in {"none", "off", "false", "4bit"}:
raise ValueError("Merged-model quantization must be 'none' or '4bit'.")
token = os.getenv("HF_TOKEN")
tokenizer = AutoTokenizer.from_pretrained(
model_path,
use_fast=True,
token=token,
clean_up_tokenization_spaces=False,
)
if tokenizer.pad_token_id is None:
tokenizer.pad_token = tokenizer.eos_token
use_4bit = requested_quantization == "4bit"
compute_dtype = dtype
load_kwargs: dict[str, Any] = {
"torch_dtype": dtype,
"token": token,
"trust_remote_code": False,
}
if use_4bit:
if device.type != "cuda":
raise RuntimeError("4-bit merged-model inference requires an NVIDIA CUDA device.")
try:
from transformers import BitsAndBytesConfig
import bitsandbytes # noqa: F401
except ImportError as exc:
raise ImportError("4-bit inference requires bitsandbytes; install with `pip install -e '.[cuda]'`.") from exc
compute_dtype = _precision_dtype(run_config.get("compute_dtype", run_config.get("precision", "bf16")), device)
load_kwargs["quantization_config"] = BitsAndBytesConfig(
load_in_4bit=True,
bnb_4bit_quant_type=str(run_config.get("quantization_type", "nf4")),
bnb_4bit_compute_dtype=compute_dtype,
bnb_4bit_use_double_quant=bool(run_config.get("double_quant", True)),
)
load_kwargs["device_map"] = {"": device.index if device.index is not None else 0}
model = AutoModelForCausalLM.from_pretrained(model_path, **load_kwargs)
model.config.use_cache = False
model.config.is_causal = False
if hasattr(model.config, "use_bidirectional_attention"):
model.config.use_bidirectional_attention = True
if not use_4bit:
model.to(device)
model.eval()
mask_info = validate_mask_token(tokenizer, str(run_config.get("mask_token", "MASK")))
session = InferenceSession(
model=model,
tokenizer=tokenizer,
device=device,
adapter_path=model_path,
config=run_config,
mask_token_id=mask_info["mask_token_id"],
quantization="4bit" if use_4bit else "none",
compute_dtype=str(compute_dtype).removeprefix("torch."),
)
preflight_session(session)
return session
def _load_legacy_checkpoint_session(checkpoint: str | Path, tokenizer_name_or_path: str, device_name: str = "auto", source_config: dict[str, Any] | None = None) -> InferenceSession:
"""Load one trusted legacy full-object checkpoint from a local path."""
checkpoint = Path(checkpoint).expanduser().resolve()
if not checkpoint.is_file():
raise ValueError(f"Legacy checkpoint does not exist: {checkpoint}")
if not tokenizer_name_or_path.strip():
raise ValueError("Legacy loading requires a tokenizer name or local tokenizer path.")
# A full-object checkpoint can execute pickle code. This loader is for
# checkpoints the user trusts, including their locally archived model.
previous_modules = install_legacy_pickle_modules()
try:
model = torch.load(checkpoint, map_location="cpu", weights_only=False)
finally:
restore_legacy_pickle_modules(previous_modules)
if not isinstance(model, torch.nn.Module):
raise ValueError(f"{checkpoint} is not a full torch.nn.Module checkpoint.")
patch_legacy_lora_modules(model)
token = os.getenv("HF_TOKEN")
device = select_device(device_name)
tokenizer = AutoTokenizer.from_pretrained(tokenizer_name_or_path.strip(), use_fast=True, token=token, clean_up_tokenization_spaces=False)
if tokenizer.pad_token_id is None:
tokenizer.pad_token = tokenizer.eos_token
model.to(device).eval()
mask_info = validate_mask_token(tokenizer)
session = InferenceSession(
model=model,
tokenizer=tokenizer,
device=device,
adapter_path=checkpoint,
config=source_config or {"model_source": "local_legacy", "checkpoint": str(checkpoint), "tokenizer_name_or_path": tokenizer_name_or_path.strip()},
mask_token_id=mask_info["mask_token_id"],
quantization="none",
compute_dtype=str(next((parameter.dtype for parameter in model.parameters() if parameter.is_floating_point()), torch.float32)).removeprefix("torch."),
legacy_wrapper=True,
prompt_format="legacy_llama",
)
preflight_session(session)
return session
def load_local_legacy_session(checkpoint_path: str | Path, tokenizer_name_or_path: str, device_name: str = "auto") -> InferenceSession:
"""Load and preflight a trusted local legacy full-model checkpoint."""
return _load_legacy_checkpoint_session(checkpoint_path, tokenizer_name_or_path, device_name)
def load_hosted_legacy_session(repo_id: str, filename: str, tokenizer_name_or_path: str, device_name: str = "auto") -> InferenceSession:
"""Load the trusted legacy full-model checkpoint hosted on Hugging Face.
This exists for controlled comparisons: the checkpoint uses its original
wrapper to construct full bidirectional attention, while decoding uses the
current project's prompt and denoising loop. Pickled checkpoints are only
safe to load from a repository you trust.
"""
if not repo_id.strip() or not filename.strip() or not tokenizer_name_or_path.strip():
raise ValueError("Hosted legacy loading requires a repository ID, checkpoint filename, and tokenizer name.")
try:
from huggingface_hub import hf_hub_download
except ImportError as exc:
raise ImportError("Hosted model loading requires huggingface_hub, installed with transformers.") from exc
token = os.getenv("HF_TOKEN")
checkpoint = hf_hub_download(repo_id=repo_id.strip(), filename=filename.strip(), token=token)
return _load_legacy_checkpoint_session(
checkpoint,
tokenizer_name_or_path,
device_name,
{"model_source": "huggingface_legacy", "repo_id": repo_id.strip(), "filename": filename.strip(), "tokenizer_name_or_path": tokenizer_name_or_path.strip()},
)
def load_llada_session(repo_id: str = "GSAI-ML/LLaDA-8B-Instruct", device_name: str = "auto") -> InferenceSession:
"""Load LLaDA Instruct as a mask predictor for this app's denoising loop."""
if not repo_id.strip():
raise ValueError("LLaDA loading requires a Hugging Face repository ID.")
device = select_device(device_name)
if device.type not in {"cuda", "mps"}:
raise ValueError("LLaDA-8B-Instruct requires CUDA or MPS inference; select a GPU-capable runtime.")
# Apple MPS does not reliably support BF16 inference for this remote model.
# FP16 is the practical MPS format; CUDA retains BF16 where available.
dtype = torch.float16 if device.type == "mps" else _precision_dtype("bf16", device)
token = os.getenv("HF_TOKEN")
cache_dir = "base_models"
try:
tokenizer = AutoTokenizer.from_pretrained(
repo_id.strip(), trust_remote_code=True, token=token, cache_dir=cache_dir,
)
model = AutoModel.from_pretrained(
repo_id.strip(), trust_remote_code=True, torch_dtype=dtype, token=token, cache_dir=cache_dir,
)
except Exception as exc:
raise RuntimeError(
"Could not load LLaDA. Its official implementation requires the remote model code "
"and is tested with transformers==4.38.2."
) from exc
if tokenizer.pad_token_id == 126336:
raise ValueError("LLaDA's pad token must differ from its fixed mask token (126336).")
tokenizer.padding_side = "left"
model.to(device).eval()
session = InferenceSession(
model=model,
tokenizer=tokenizer,
device=device,
adapter_path=Path(repo_id.strip()),
config={"model_source": "huggingface_llada", "repo_id": repo_id.strip()},
mask_token_id=126336,
quantization="none",
compute_dtype=str(dtype).removeprefix("torch."),
prompt_format="llada",
llada=True,
)
preflight_session(session)
return session
def _sample(
logits: torch.Tensor,
temperature: float,
top_k: int,
generator: torch.Generator | None,
) -> tuple[torch.Tensor, torch.Tensor]:
"""Top-k sample token IDs and return their normalized sampling confidence."""
logits = logits / max(temperature, 1e-5)
vocab_size = logits.shape[-1]
k = min(max(int(top_k), 1), vocab_size)
values, indices = torch.topk(logits, k, dim=-1)
probabilities = F.softmax(values, dim=-1)
picked_local = torch.multinomial(probabilities, 1, generator=generator)
picked = indices.gather(-1, picked_local).squeeze(-1)
confidence = probabilities.gather(-1, picked_local).squeeze(-1)
return picked, confidence
def _llada_gumbel_noise(logits: torch.Tensor, temperature: float) -> torch.Tensor:
"""Apply the float64 Gumbel-max transform used by official LLaDA decoding."""
if float(temperature) == 0.0:
return logits
logits = logits.to(torch.float64)
noise = torch.rand_like(logits, dtype=torch.float64)
return logits.exp() / (-torch.log(noise)).pow(float(temperature))
def _apply_repetition_penalty(
logits: torch.Tensor,
answer_ids: torch.Tensor,
penalty: float,
mask_token_id: int,
*,
exclude_self: bool = False,
excluded_token_ids: set[int] | None = None,
) -> torch.Tensor:
"""Reduce repeated-token probability weights by ``penalty ** count``.
Only tokens already present in the generated answer are penalized; prompt
tokens, MASK, and configured special tokens never contribute. Subtracting
``count * log(penalty)`` from a logit divides its unnormalized softmax
probability by ``penalty ** count``. For the revisable denoise stream, a
position's current token is excluded from its count, avoiding needless
churn of unique predictions.
"""
penalty = float(penalty)
if penalty < 1.0:
raise ValueError("repetition_penalty must be at least 1.0")
if penalty == 1.0:
return logits
if logits.ndim != 3 or answer_ids.ndim != 2 or logits.shape[:2] != answer_ids.shape:
raise ValueError("repetition penalty expects logits [batch, length, vocab] matching answer IDs")
adjusted = logits.clone()
vocabulary_size = adjusted.shape[-1]
for batch_index in range(answer_ids.shape[0]):
valid = answer_ids[batch_index]
valid_mask = (
(valid != int(mask_token_id))
& (valid >= 0)
& (valid < vocabulary_size)
)
excluded = set(excluded_token_ids or ())
excluded.add(int(mask_token_id))
if excluded:
excluded_tensor = torch.tensor(
sorted(excluded), device=valid.device, dtype=valid.dtype
)
valid_mask &= ~torch.isin(valid, excluded_tensor)
valid = valid[valid_mask]
if not len(valid):
continue
token_ids, counts = torch.unique(valid, return_counts=True)
current = answer_ids[batch_index]
scores = adjusted[batch_index, :, token_ids]
exponents = counts[None, :].expand(logits.shape[1], -1)
if exclude_self:
exponents = exponents - (current[:, None] == token_ids[None, :]).to(
dtype=exponents.dtype
)
log_penalty = torch.log(
torch.tensor(penalty, device=logits.device, dtype=torch.float32)
)
penalized = scores.float() - exponents.float() * log_penalty
# Penalty arithmetic stays in FP32 for numerical stability, then returns
# to the model's native BF16/FP16 dtype for indexed assignment.
adjusted[batch_index, :, token_ids] = penalized.to(dtype=adjusted.dtype)
return adjusted
def _apply_eos_eot_prediction_penalty(
logits: torch.Tensor,
penalty: float,
eos_token_id: int,
eot_token_id: int | None = None,
) -> torch.Tensor:
"""Reduce EOS/EoT sampling weights without changing retention confidence.
Subtracting ``log(penalty)`` from the selected logits divides their
unnormalized probability weight by ``penalty``. This is deliberately
independent of the LLaDA-style delayed-retention option, which changes
which sampled positions are retained or re-masked rather than what token
is sampled in the first place.
"""
penalty = float(penalty)
if penalty < 1.0:
raise ValueError("EOS/EOT prediction penalty must be at least 1.0")
if penalty == 1.0:
return logits
token_ids = {int(eos_token_id)}
if eot_token_id is not None:
token_ids.add(int(eot_token_id))
token_ids = {token_id for token_id in token_ids if 0 <= token_id < logits.shape[-1]}
if not token_ids:
return logits
adjusted = logits.clone()
log_penalty = torch.log(
torch.tensor(penalty, device=logits.device, dtype=torch.float32)
)
indices = torch.tensor(sorted(token_ids), device=logits.device, dtype=torch.long)
adjusted[..., indices] = (
adjusted[..., indices].float() - log_penalty
).to(dtype=adjusted.dtype)
return adjusted
def _llada_transfer_schedule(mask_count: int, steps: int) -> list[int]:
"""Distribute a linear-noise transfer budget uniformly across steps."""
if mask_count < 0 or steps < 1:
raise ValueError("mask_count must be non-negative and steps must be positive")
base, remainder = divmod(mask_count, steps)
return [base + int(index < remainder) for index in range(steps)]
def _block_step_plan(
generation_length: int,
steps: int,
block_length: int | None,
) -> list[tuple[int, int, int, int, int]]:
"""Allocate a fixed total step budget across sequential answer blocks."""
generation_length, steps = int(generation_length), int(steps)
requested_block_length = generation_length if block_length is None else int(block_length)
if generation_length < 1 or steps < 1 or requested_block_length < 1:
raise ValueError("generation length, steps, and block length must be positive")
effective_block_length = min(requested_block_length, generation_length)
num_blocks = (generation_length + effective_block_length - 1) // effective_block_length
if steps < num_blocks:
raise ValueError(
f"Denoising steps ({steps}) must be at least the number of blocks ({num_blocks})"
)
base, remainder = divmod(steps, num_blocks)
plan = []
for block_index in range(num_blocks):
block_start = block_index * effective_block_length
block_end = min(generation_length, block_start + effective_block_length)
block_steps = base + int(block_index < remainder)
for block_step in range(block_steps):
plan.append((block_index, block_start, block_end, block_step, block_steps))
return plan
def _remask_offsets(confidence: torch.Tensor, mask_probability: float, confidence_guided: bool) -> torch.Tensor:
"""Choose answer offsets to re-mask, preferring uncertain tokens when guided."""
probability = max(0.0, min(1.0, float(mask_probability)))
if confidence_guided:
count = round(probability * len(confidence))
return torch.argsort(confidence)[:count]
return torch.where(torch.rand(len(confidence), device=confidence.device) < probability)[0]
def _native_eot_token_id(tokenizer: Any) -> int | None:
"""Return a tokenizer's native end-of-turn ID when it has one."""
convert = getattr(tokenizer, "convert_tokens_to_ids", None)
if convert is None:
return None
unknown = getattr(tokenizer, "unk_token_id", None)
for token in ("<|eot_id|>", "<end_of_turn>", "<|end_of_turn|>"):
token_id = convert(token)
if token_id is not None and token_id != unknown and int(token_id) >= 0:
return int(token_id)
return None
def forward_denoising(session: InferenceSession, input_ids: torch.Tensor, padding_mask: torch.Tensor) -> torch.Tensor:
"""Return denoising logits for either the current or legacy model wrapper."""
if session.llada:
# LLaDA caches rotary embeddings during preflight. Its remote model code
# requires all later uses of those cached inference tensors to remain in
# inference mode as well.
with torch.inference_mode():
outputs = session.model(input_ids, attention_mask=(~padding_mask).to(dtype=torch.long))
return outputs.logits
if session.legacy_wrapper:
# The archived CustomTransformerModel builds its own full-attention
# 4-D mask and passes use_cache=False to its inner Peft model. Passing
# either argument here would duplicate the wrapper's keyword.
outputs = session.model(input_ids=input_ids)
return outputs["logits"] if isinstance(outputs, dict) else outputs.logits
return forward_bidirectional(session.model, input_ids, padding_mask)
@torch.inference_mode()
def preflight_session(session: InferenceSession) -> tuple[int, int]:
"""Run one real forward pass and fail early if a loaded model is unusable."""
prefix = _prompt_ids(session.tokenizer, "Reply with OK.", "You are a helpful assistant.", session.prompt_format)
input_ids = torch.tensor([prefix + [session.mask_token_id]], device=session.device, dtype=torch.long)
padding = torch.zeros_like(input_ids, dtype=torch.bool)
try:
logits = forward_denoising(session, input_ids, padding)
except Exception as exc:
source = "LLaDA" if session.llada else "legacy hosted checkpoint" if session.legacy_wrapper else "saved adapter"
raise RuntimeError(f"Inference preflight failed for {source}; the model was not loaded for generation: {exc}") from exc
if logits.ndim != 3 or logits.shape[:2] != input_ids.shape:
raise RuntimeError(f"Inference preflight returned invalid logits shape {tuple(logits.shape)} for input shape {tuple(input_ids.shape)}")
if not torch.isfinite(logits[:, -1]).all():
raise RuntimeError("Inference preflight produced non-finite final-token logits.")
return int(input_ids.shape[1]), int(logits.shape[-1])
def _prompt_ids(tokenizer: Any, question: str, system_prompt: str, prompt_format: str = "chat_template") -> list[int]:
"""Render system/user messages through a tokenizer’s native chat template."""
from .data import apply_neutral_chat_template
if not question.strip():
raise ValueError("Enter a question or prompt.")
if prompt_format == "legacy_llama":
# The hosted historical checkpoint used a base Llama tokenizer with no
# chat_template. Match the prompt layout from its original app while
# still running the current project's denoising/sampling loop.
prompt = (
"<|begin_of_text|>\n"
"<|start_header_id|>system<|end_header_id|>\n"
f"{system_prompt}\n"
"<|start_header_id|>user<|end_header_id|>\n"
f"{question.strip()}\n"
"<|start_header_id|>assistant<|end_header_id|>\n"
)
return list(tokenizer.encode(prompt, add_special_tokens=False))
if prompt_format == "llada":
content = f"{system_prompt}\n\n{question.strip()}" if system_prompt.strip() else question.strip()
rendered = apply_neutral_chat_template(
tokenizer,
[{"role": "user", "content": content}],
tokenize=True,
add_generation_prompt=True,
)
if isinstance(rendered, str):
rendered = tokenizer.encode(rendered, add_special_tokens=False)
elif hasattr(rendered, "input_ids"):
rendered = rendered.input_ids
if rendered and isinstance(rendered[0], list):
rendered = rendered[0]
return list(rendered)
if prompt_format != "chat_template":
raise ValueError(f"Unknown prompt format: {prompt_format}")
if not getattr(tokenizer, "chat_template", None):
raise ValueError(f"{tokenizer.name_or_path} has no chat template; inference needs one to identify the answer boundary.")
messages = [
{"role": "system", "content": system_prompt},
{"role": "user", "content": question},
]
try:
rendered = apply_neutral_chat_template(tokenizer, messages, tokenize=True, add_generation_prompt=True)
except Exception as exc:
# Gemma's official template rejects a separate system role; preserve
# the prompt by folding it into the user message, as training does.
if exc.__class__.__name__ != "TemplateError" or "System role not supported" not in str(exc):
raise
rendered = apply_neutral_chat_template(tokenizer, [
{"role": "user", "content": f"{system_prompt}\n\n{question}"},
], tokenize=True, add_generation_prompt=True)
if isinstance(rendered, str):
rendered = tokenizer.encode(rendered, add_special_tokens=False)
elif hasattr(rendered, "input_ids"):
rendered = rendered.input_ids
if rendered and isinstance(rendered[0], list):
rendered = rendered[0]
if not all(isinstance(token, int) for token in rendered):
raise ValueError(f"Tokenizer returned a non-integer chat-template encoding: {type(rendered).__name__}")
return list(rendered)
@torch.inference_mode()
def llada_generate(
session: InferenceSession,
question: str,
*,
gen_length: int,
steps: int,
block_length: int | None = None,
temperature: float = 0.0,
cfg_scale: float = 0.0,
remasking: str = "low_confidence",
logits_eos_inf: bool = False,
confidence_eos_eot_inf: bool = False,
eot_token_id: int | None = None,
system_prompt: str = "",
seed: int = 1234,
repetition_penalty: float = 1.0,
eos_eot_prediction_penalty: float = 1.0,
) -> str:
"""Generate with the official LLaDA fixed-budget transfer algorithm.
This intentionally bypasses ``denoise_stream``: official LLaDA predicts
only still-masked positions, permanently transfers a fixed number per
reverse step, and uses neither proportional unmasking nor a mask-ratio
heuristic. The sampler is model-agnostic, so mask-only LAD adapters can use
it with their native tokenizer and prompt format as well.
"""
gen_length, steps = int(gen_length), int(steps)
block_length = int(block_length or gen_length)
if gen_length < 1 or steps < 1 or block_length < 1:
raise ValueError("gen_length, steps, and block_length must be positive")
if gen_length % block_length:
raise ValueError("LLaDA gen_length must be divisible by block_length")
num_blocks = gen_length // block_length
if steps % num_blocks:
raise ValueError("LLaDA steps must be divisible by the number of blocks")
if remasking not in {"low_confidence", "random"}:
raise ValueError("LLaDA remasking must be 'low_confidence' or 'random'")
torch.manual_seed(int(seed))
if session.device.type == "cuda":
torch.cuda.manual_seed_all(int(seed))
prefix = _prompt_ids(session.tokenizer, question, system_prompt, session.prompt_format)
prompt_length = len(prefix)
x = torch.full((1, prompt_length + gen_length), session.mask_token_id, dtype=torch.long, device=session.device)
x[0, :prompt_length] = torch.tensor(prefix, dtype=torch.long, device=session.device)
padding = torch.zeros_like(x, dtype=torch.bool)
prompt_index = x != session.mask_token_id
steps_per_block = steps // num_blocks
eos_token_id = int(session.tokenizer.eos_token_id)
for block in range(num_blocks):
block_start = prompt_length + block * block_length
block_end = block_start + block_length
transfer_schedule = _llada_transfer_schedule(int((x[:, block_start:block_end] == session.mask_token_id).sum()), steps_per_block)
for transfer_count in transfer_schedule:
mask_index = x == session.mask_token_id
if cfg_scale > 0.0:
unconditional = x.clone()
unconditional[prompt_index] = session.mask_token_id
model_input = torch.cat([x, unconditional], dim=0)
model_padding = torch.cat([padding, padding], dim=0)
conditional_logits, unconditional_logits = forward_denoising(session, model_input, model_padding).chunk(2, dim=0)
logits = unconditional_logits + (float(cfg_scale) + 1.0) * (conditional_logits - unconditional_logits)
else:
logits = forward_denoising(session, x, padding)
logits[:, prompt_length:] = _apply_repetition_penalty(
logits[:, prompt_length:],
x[:, prompt_length:],
repetition_penalty,
session.mask_token_id,
excluded_token_ids=set(getattr(session.tokenizer, "all_special_ids", [])),
)
logits[:, prompt_length:] = _apply_eos_eot_prediction_penalty(
logits[:, prompt_length:],
eos_eot_prediction_penalty,
eos_token_id,
eot_token_id,
)
if logits_eos_inf:
logits = logits.clone()
logits[..., eos_token_id] = -torch.inf
predictions = torch.argmax(_llada_gumbel_noise(logits, temperature), dim=-1)
if remasking == "low_confidence":
probabilities = F.softmax(logits, dim=-1)
confidence = probabilities.gather(-1, predictions.unsqueeze(-1)).squeeze(-1)
if confidence_eos_eot_inf:
# Appendix B.4 delays EOS/EoT predictions by assigning
# them the lowest transfer confidence; they remain valid
# predictions and can still transfer in later steps.
special_prediction = predictions == eos_token_id
if eot_token_id is not None and 0 <= int(eot_token_id) < logits.shape[-1]:
special_prediction |= predictions == int(eot_token_id)
confidence = confidence.masked_fill(special_prediction, torch.finfo(confidence.dtype).min)
else:
confidence = torch.rand(predictions.shape, device=session.device)
candidate = mask_index.clone()
candidate[:, :block_start] = False
candidate[:, block_end:] = False
confidence = confidence.masked_fill(~candidate, -torch.inf)
if transfer_count:
transfer = torch.topk(confidence[0], k=int(transfer_count)).indices
x[0, transfer] = predictions[0, transfer]
answer = x[0, prompt_length:].tolist()
return session.tokenizer.decode(answer, skip_special_tokens=True).strip()
def render_denoising_step(
tokens: list[int],
confidences: list[float],
answer_start: int,
tokenizer: Any,
mask_token_id: int,
step: int,
total_steps: int,
retained: set[int] | None = None,
frozen: dict[int, int] | None = None,
frozen_confidences: dict[int, float] | None = None,
frozen_steps: dict[int, int] | None = None,
color_mode: str = "Prediction probability",
) -> str:
"""Render one denoising state with optional token coloring."""
eos_id = tokenizer.eos_token_id
pieces = []
answer = tokens[answer_start:]
output_token_count = 0
for offset, token in enumerate(answer):
if token == eos_id:
break
output_token_count += 1
token_text = escape(tokenizer.decode([token], skip_special_tokens=False)).replace("\n", "↵ ")
if token == mask_token_id:
style, token_text = (
"display:inline-block;background:#d1d5db;color:#4b5563;"
"border:1px solid #9ca3af;border-radius:4px;padding:0 4px;"
"font-size:.78em;line-height:1.45;margin:0 1px;vertical-align:baseline",
"mask",
)
title = f"token position {offset} · masked at iteration {step}"
else:
confidence = (
frozen_confidences[offset]
if frozen and offset in frozen and frozen_confidences and offset in frozen_confidences
else float(confidences[offset]) if offset < len(confidences) else 0.0
)
confidence = max(0.0, min(1.0, confidence))
predicted_step = (
frozen_steps[offset]
if frozen and offset in frozen and frozen_steps and offset in frozen_steps
else step
)
if color_mode == "Prediction iteration":
iteration_fraction = max(0.0, min(1.0, predicted_step / max(total_steps, 1)))
lightness = 72 - round(42 * iteration_fraction)
style = f"color:hsl(210,90%,{lightness}%);font-weight:500"
elif color_mode == "Prediction probability":
hue = int(confidence * 120)
style = f"color:hsl({hue},90%,30%);font-weight:{'600' if confidence > .8 else '400'}"
else:
style = "color:inherit;font-weight:400"
title = (
f"token position {offset} · predicted at iteration {predicted_step} · "
f"sampling probability {confidence:.1%}"
)
pieces.append(f"<span style='{style}' title='{title}'>{token_text}</span>")
pct = int(100 * step / max(total_steps, 1))
if color_mode == "Prediction iteration":
legend = "Light-to-dark blue indicates earlier-to-later prediction iterations"
elif color_mode == "Prediction probability":
legend = "Red-to-green indicates lower-to-higher sampling probability"
else:
legend = "Token coloring is disabled"
return (f"<div style='font-family:system-ui;padding:14px;border:1px solid #d1d5db;border-radius:9px;background:#fafafa'>"
f"<div style='font-weight:700;color:#2563eb;margin-bottom:7px'>Denoising step {step}/{total_steps} · {output_token_count} output tokens</div>"
f"<div style='background:#e5e7eb;border-radius:4px;height:7px;margin-bottom:10px'><div style='background:#2563eb;width:{pct}%;height:7px;border-radius:4px'></div></div>"
f"<div style='line-height:2;font-size:15px;white-space:pre-wrap'>{''.join(pieces)}</div>"
f"<div style='font-size:11px;color:#6b7280;margin-top:8px'>{legend}; gray chips are masks. Hover over a token for its position, prediction iteration, and probability.</div></div>")
def decode_denoising_state(
tokens: list[int],
tokenizer: Any,
mask_token_id: int,
*,
show_eos_tokens: bool = False,
) -> str:
"""Decode one answer state with unresolved positions and optional EOS shown."""
eos_id = tokenizer.eos_token_id
eot_id = _native_eot_token_id(tokenizer) if show_eos_tokens else None
visible_end_ids = {int(eos_id)}
if eot_id is not None:
visible_end_ids.add(int(eot_id))
if not show_eos_tokens and eos_id in tokens:
tokens = tokens[:tokens.index(eos_id)]
# Decode contiguous resolved spans so subword spacing remains natural, but
# make adjacent mask tokens unambiguous and independent of the configured
# mask vocabulary item (some model configs use markers such as `<?>`).
pieces: list[str] = []
resolved: list[int] = []
def flush_resolved() -> None:
if resolved:
text = tokenizer.decode(
resolved,
skip_special_tokens=True,
clean_up_tokenization_spaces=False,
).strip()
if text:
pieces.append(text)
resolved.clear()
for token in tokens:
if token == mask_token_id:
flush_resolved()
pieces.append("MASK")
elif show_eos_tokens and token in visible_end_ids:
flush_resolved()
marker = tokenizer.decode(
[token],
skip_special_tokens=False,
clean_up_tokenization_spaces=False,
).strip()
fallback = "<EOS>" if token == eos_id else "<EOT>"
pieces.append(marker or fallback)
else:
resolved.append(token)
flush_resolved()
return " ".join(pieces)
def denoise_stream(session: InferenceSession, question: str, system_prompt: str, max_new_tokens: int, num_steps: int, noise_level: float, temperature: float, top_k: int, seed: int, permanent_unmask: bool = False, confidence_guided: bool = False, proportional_unmask: bool = True, early_stopping: bool = False, confidence_eos_eot_inf: bool = False, freeze_retained_tokens: bool = True, repetition_penalty: float = 1.0, eos_eot_prediction_penalty: float = 1.0, include_pre_remask_prediction: bool = False, block_length: int | None = None, trajectory_color_mode: str = "Prediction probability"):
"""Yield denoising states with optionally retained positions and locked values."""
prefix = _prompt_ids(session.tokenizer, question, system_prompt, session.prompt_format)
max_new_tokens, num_steps = int(max_new_tokens), int(num_steps)
if max_new_tokens < 1 or num_steps < 1:
raise ValueError("max_new_tokens and num_steps must both be at least 1.")
step_plan = _block_step_plan(max_new_tokens, num_steps, block_length)
num_blocks = step_plan[-1][0] + 1
ids = prefix + [session.mask_token_id] * max_new_tokens
answer_start = len(prefix)
# Use the device's default RNG so this works consistently on CUDA, MPS, and
# CPU; seed it once per request for reproducible interactive runs.
torch.manual_seed(int(seed))
if session.device.type == "cuda":
torch.cuda.manual_seed_all(int(seed))
padding = torch.zeros((1, len(ids)), device=session.device, dtype=torch.bool)
last_confidence = 0.0
retained: set[int] = set()
frozen: dict[int, int] = {}
frozen_confidences: dict[int, float] = {}
frozen_steps: dict[int, int] = {}
last_predictions: list[tuple[int, ...]] = []
eot_token_id = _native_eot_token_id(session.tokenizer) if confidence_eos_eot_inf or float(eos_eot_prediction_penalty) > 1.0 else None
# EOS/EOT delaying and confidence-guided retention are independent. In
# random mode, endings are still retained last, while ordinary tokens are
# selected randomly.
guided_retention = confidence_guided
skip_block_index: int | None = None
for step, (block_index, block_start, block_end, block_step, block_steps) in enumerate(step_plan):
if block_index == skip_block_index:
continue
if block_step == 0:
last_predictions.clear()
tokens = torch.tensor([ids], device=session.device, dtype=torch.long)
with torch.inference_mode():
answer_ids = tokens[:, answer_start:]
logits = forward_denoising(session, tokens, padding)[:, answer_start:]
logits = _apply_repetition_penalty(
logits,
answer_ids,
repetition_penalty,
session.mask_token_id,
exclude_self=True,
excluded_token_ids=set(getattr(session.tokenizer, "all_special_ids", [])),
)[0]
logits = _apply_eos_eot_prediction_penalty(
logits,
eos_eot_prediction_penalty,
session.tokenizer.eos_token_id,
eot_token_id,
)
sampled, confidence = _sample(logits, float(temperature), int(top_k), None)
retention_confidence = confidence
special_prediction = torch.zeros_like(sampled, dtype=torch.bool)
if confidence_eos_eot_inf:
special_prediction = sampled == session.tokenizer.eos_token_id
if eot_token_id is not None:
special_prediction |= sampled == eot_token_id
retention_confidence = confidence.masked_fill(
special_prediction, torch.finfo(confidence.dtype).min
)
ids[answer_start + block_start : answer_start + block_end] = sampled[block_start:block_end].tolist()
if freeze_retained_tokens:
for offset, token in frozen.items():
ids[answer_start + offset] = token
predicted_text = decode_denoising_state(
ids[answer_start:],
session.tokenizer,
session.mask_token_id,
show_eos_tokens=include_pre_remask_prediction,
)
last_confidence = float(confidence[block_start:block_end].mean().cpu())
# Compare the visible sampled answer before the next iteration's
# re-masking. Tokens after the first EOS are not part of the answer and
# must not prevent convergence. Excluding EOS itself still preserves
# its position through the tuple length: moving EOS changes the prefix.
prediction = ids[answer_start:]
if session.tokenizer.eos_token_id in prediction:
prediction = prediction[:prediction.index(session.tokenizer.eos_token_id)]
last_predictions.append(tuple(prediction))
if len(last_predictions) > 2:
last_predictions.pop(0)
stopped_early = early_stopping and len(last_predictions) == 2 and len(set(last_predictions)) == 1
# Progressively reduce corruption. Re-mask independently, retaining the
# legacy schedule's initial noise_level and ending with a clean sample.
if block_step + 1 < block_steps and not stopped_early:
block_size = block_end - block_start
mask_probability = max(0.0, min(1.0, float(noise_level) * (1.0 - (block_step + 1) / block_steps)))
if permanent_unmask:
keep_count = min(block_size, max(0, round((1.0 - mask_probability) * block_size)))
retained_in_block = sum(block_start <= i < block_end for i in retained)
needed = keep_count - retained_in_block
candidates = [i for i in range(block_start, block_end) if i not in retained]
if needed > 0 and candidates:
def random_retention_order(pool: list[int]) -> list[int]:
order = torch.randperm(len(pool), device=session.device).tolist()
randomized = [pool[i] for i in order]
if not confidence_eos_eot_inf:
return randomized
ordinary = [i for i in randomized if not bool(special_prediction[i])]
endings = [i for i in randomized if bool(special_prediction[i])]
return ordinary + endings
if proportional_unmask:
eos_positions = [i for i in range(block_start, block_end) if ids[answer_start + i] == session.tokenizer.eos_token_id]
boundary = min(eos_positions) if eos_positions else block_end
pools = [[i for i in candidates if i < boundary], [i for i in candidates if i >= boundary]]
target_normal = round(keep_count * (boundary - block_start) / block_size)
target_counts = [
max(0, target_normal - sum(block_start <= i < boundary for i in retained)),
max(0, keep_count - target_normal - sum(boundary <= i < block_end for i in retained)),
]
chosen = []
for pool, target in zip(pools, target_counts):
if not pool or target <= 0:
continue
if guided_retention:
order = torch.argsort(retention_confidence, descending=True).tolist()
chosen.extend([i for i in order if i in pool][:target])
else:
chosen.extend(random_retention_order(pool)[:target])
if len(chosen) < needed:
remainder = [i for i in candidates if i not in chosen]
chosen.extend(remainder[: needed - len(chosen)])
elif guided_retention:
confidence_order = torch.argsort(retention_confidence, descending=True).tolist()
chosen = [i for i in confidence_order if i in candidates][:needed]
else:
chosen = random_retention_order(candidates)[:needed]
for offset in chosen:
retained.add(offset)
if freeze_retained_tokens:
frozen[offset] = ids[answer_start + offset]
frozen_confidences[offset] = float(confidence[offset].cpu())
frozen_steps[offset] = step + 1
for offset in range(block_start, block_end):
if offset not in retained:
ids[answer_start + offset] = session.mask_token_id
else:
# Confidence-guided refinement keeps every token revisable, but
# preferentially re-masks the least certain predictions. The
# unguided mode retains the original random re-masking policy.
remask_offsets = _remask_offsets(
retention_confidence[block_start:block_end],
mask_probability,
guided_retention,
)
if confidence_eos_eot_inf and not guided_retention:
ending_offsets = torch.where(special_prediction[block_start:block_end])[0]
remask_offsets = torch.unique(torch.cat((remask_offsets, ending_offsets)))
for offset in remask_offsets.tolist():
ids[answer_start + block_start + offset] = session.mask_token_id
current_answer = ids[answer_start:]
visible_answer = current_answer
if session.tokenizer.eos_token_id in visible_answer:
visible_answer = visible_answer[:visible_answer.index(session.tokenizer.eos_token_id)]
remasked_text = decode_denoising_state(
current_answer,
session.tokenizer,
session.mask_token_id,
show_eos_tokens=include_pre_remask_prediction,
)
current_text = remasked_text
if include_pre_remask_prediction:
remask_label = (
"State after re-mask"
if block_step + 1 < block_steps and not stopped_early
else "State after re-mask (unchanged; final state)"
)
current_text = (
f"Predicted (before re-mask):\n{predicted_text}\n"
f"{remask_label}:\n{remasked_text}"
)
status = f"Denoising step {step + 1}/{num_steps} · {len(visible_answer)} output tokens · mean confidence {last_confidence:.3f}"
if num_blocks > 1:
status += f" · block {block_index + 1}/{num_blocks}"
if permanent_unmask:
status += f" · retained {len(retained)} tokens"
if stopped_early:
status += " · block stopped early (same answer for 2 consecutive iterations)"
html = render_denoising_step(
ids,
confidence.tolist(),
answer_start,
session.tokenizer,
session.mask_token_id,
step + 1,
num_steps,
retained if permanent_unmask else None,
frozen if permanent_unmask and freeze_retained_tokens else None,
frozen_confidences if permanent_unmask and freeze_retained_tokens else None,
frozen_steps if permanent_unmask and freeze_retained_tokens else None,
trajectory_color_mode,
)
yield current_text, status, html
if stopped_early:
if num_blocks == 1:
break
skip_block_index = block_index
answer = ids[answer_start:]
if session.tokenizer.eos_token_id in answer:
answer = answer[:answer.index(session.tokenizer.eos_token_id)]
text = session.tokenizer.decode(answer, skip_special_tokens=True).strip()
return
def denoise(session: InferenceSession, question: str, system_prompt: str, max_new_tokens: int, num_steps: int, noise_level: float, temperature: float, top_k: int, seed: int, permanent_unmask: bool = False, confidence_guided: bool = False, proportional_unmask: bool = True, early_stopping: bool = False, progress: Callable[[float, str], None] | None = None, confidence_eos_eot_inf: bool = False, freeze_retained_tokens: bool = True, repetition_penalty: float = 1.0, eos_eot_prediction_penalty: float = 1.0, block_length: int | None = None) -> tuple[str, str]:
"""Run denoising to completion and return only the final text and status."""
result = ("", "")
for step, (text, status, _html) in enumerate(denoise_stream(session, question, system_prompt, max_new_tokens, num_steps, noise_level, temperature, top_k, seed, permanent_unmask, confidence_guided, proportional_unmask, early_stopping, confidence_eos_eot_inf, freeze_retained_tokens, repetition_penalty, eos_eot_prediction_penalty, False, block_length), start=1):
result = (text, status)
if progress:
progress(step / int(num_steps), status)
return result
def release_session(session: InferenceSession | None) -> None:
"""Free a loaded inference model and release backend allocator caches."""
if session is None:
return
del session.model
gc.collect()
if torch.cuda.is_available():
torch.cuda.empty_cache()
if torch.backends.mps.is_available():
torch.mps.empty_cache()