"""Discrete diffusion generation shared by DPLM and DPLM2. The implementation keeps model-specific vocabulary rules at the public entry points and shares only the categorical sampling and confidence-based remasking mechanism. It has no dependency on the pinned upstream checkout. """ from __future__ import annotations import math import torch from collections.abc import Iterable, Iterator, Mapping from contextlib import contextmanager from typing import Any, Protocol from tqdm.auto import tqdm class _MaskedLanguageModel(Protocol): """Structural type used by the two generation entry points.""" config: Any def eval(self) -> Any: ... def modules(self) -> Iterable[torch.nn.Module]: ... def __call__(self, **kwargs: Any) -> Any: ... _DPLM2_AA_BOUNDARY = 33 _DPLM2_AA_BOS = 0 _DPLM2_PAD = 1 _DPLM2_AA_EOS = 2 _DPLM2_AA_UNK = 3 _DPLM2_AA_X = 24 _DPLM2_AA_B = 25 _DPLM2_AA_U = 26 _DPLM2_AA_Z = 27 _DPLM2_AA_O = 28 _DPLM2_AA_MASK = 32 _DPLM2_STRUCT_BOS = 33 _DPLM2_STRUCT_EOS = 34 _DPLM2_STRUCT_UNK = 35 @contextmanager def _temporary_eval(model: _MaskedLanguageModel) -> Iterator[None]: """Run one generation forward in eval mode and restore every module flag.""" training_states = tuple((module, module.training) for module in model.modules()) model.eval() try: yield finally: for module, training in training_states: module.training = training def _resolve_max_iter(model: _MaskedLanguageModel, max_iter: int | None) -> int: if max_iter is None: max_iter = int(getattr(model.config, "num_diffusion_timesteps", 500)) if isinstance(max_iter, bool) or not isinstance(max_iter, int) or max_iter <= 0: raise ValueError("max_iter must be a positive integer") return max_iter def _validate_inputs( input_tokens: torch.Tensor, partial_masks: torch.Tensor | None, ) -> torch.Tensor | None: # input_tokens: (b, l); partial_masks: (b, l) or None if input_tokens.ndim != 2 or input_tokens.dtype not in { torch.int8, torch.int16, torch.int32, torch.int64, torch.uint8, }: raise ValueError("input_tokens must be an integer tensor with shape (b, l)") if input_tokens.shape[-1] == 0: raise ValueError("input_tokens must contain at least one token") if partial_masks is None: return None if partial_masks.shape != input_tokens.shape or partial_masks.dtype != torch.bool: raise ValueError("partial_masks must be boolean with the same shape as input_tokens") if partial_masks.device != input_tokens.device: raise ValueError("partial_masks and input_tokens must be on the same device") return partial_masks # (b, l) def _validate_temperature(temperature: float | None, *, default: float = 1.0) -> float: if temperature is None: temperature = default temperature = float(temperature) if not math.isfinite(temperature) or temperature < 0: raise ValueError("temperature must be finite and non-negative") return temperature def _steps(max_iter: int, *, show_progress: bool) -> Iterable[int]: return tqdm(range(max_iter), desc="Decoding", disable=not show_progress) def _categorical( logits: torch.Tensor, *, temperature: float, ) -> tuple[torch.Tensor, torch.Tensor]: # logits: (..., c) if temperature == 0: scores, tokens = logits.log_softmax(dim=-1).max(dim=-1) # (...), (...) return tokens, scores # (...), (...) distribution = torch.distributions.Categorical(logits=logits.div(temperature)) tokens = distribution.sample() # (...) return tokens, distribution.log_prob(tokens) # (...), (...) def _gumbel_argmax( logits: torch.Tensor, *, noise_scale: float, ) -> tuple[torch.Tensor, torch.Tensor]: # logits: (..., c) uniform = torch.rand_like(logits) # (..., c) noise = -torch.log(-torch.log(uniform + 1e-8) + 1e-8) # (..., c) return _categorical(logits + noise_scale * noise, temperature=0.0) # (...), (...) def _top_p(logits: torch.Tensor, probability: float = 0.95) -> torch.Tensor: """Apply the nucleus filter used by the official DPLM samplers.""" # logits: (..., c) original_shape = logits.shape flattened = logits.reshape(-1, original_shape[-1]) # (n, c) sorted_logits, sorted_indices = flattened.sort(dim=-1, descending=True) # (n, c), (n, c) cumulative = sorted_logits.softmax(dim=-1).cumsum(dim=-1) # (n, c) remove = cumulative > probability # (n, c) remove[..., 1:] = remove[..., :-1].clone() remove[..., 0] = False sorted_logits.masked_fill_(remove, -math.inf) return sorted_logits.gather(1, sorted_indices.argsort(dim=-1)).reshape( # (..., c) original_shape ) def _lowest_confidence_mask( scores: torch.Tensor, eligible: torch.Tensor, *, rate: float, stochastic_temperature: float | None = None, ) -> torch.Tensor: # scores, eligible: (b, l) selection_scores = scores.masked_fill(~eligible, 1000.0) # (b, l) if stochastic_temperature is not None: uniform = torch.rand_like(selection_scores) # (b, l) noise = -torch.log(-torch.log(uniform + 1e-8) + 1e-8) # (b, l) selection_scores = selection_scores + stochastic_temperature * rate * noise # (b, l) cutoff_index = ( # (b, 1) eligible.sum(dim=-1, keepdim=True).to(scores.dtype) * rate ).long() cutoff_index.clamp_(min=0, max=scores.shape[-1] - 1) sorted_scores = selection_scores.sort(dim=-1).values # (b, l) cutoff = sorted_scores.gather(dim=-1, index=cutoff_index) # (b, 1) return (selection_scores < cutoff) & eligible # (b, l) def _reparameterize( output_tokens: torch.Tensor, output_scores: torch.Tensor, candidate_tokens: torch.Tensor, candidate_scores: torch.Tensor, active_mask: torch.Tensor, eligible: torch.Tensor, *, mask_token_id: int, rate: float, stochastic_temperature: float | None = None, ) -> tuple[torch.Tensor, torch.Tensor, torch.Tensor]: # All tensor inputs except vocabulary-bearing candidate logits: (b, l). remask = _lowest_confidence_mask( # (b, l) candidate_scores, eligible, rate=rate, stochastic_temperature=stochastic_temperature, ) output_tokens.masked_fill_(remask, mask_token_id) output_scores.masked_fill_(remask, -math.inf) accept = active_mask & eligible & ~remask # (b, l) output_tokens.masked_scatter_(accept, candidate_tokens[accept]) output_scores.masked_scatter_(accept, candidate_scores[accept]) return remask, output_tokens, output_scores # (b, l), (b, l), (b, l) def _logits(output: object) -> torch.Tensor: value = output.get("logits") if isinstance(output, Mapping) else getattr(output, "logits", None) if not torch.is_tensor(value): raise RuntimeError("The masked-language model did not return logits") return value # (b, l, c) def _suppress_token_ids(logits: torch.Tensor, token_ids: Iterable[int]) -> None: vocabulary_size = logits.shape[-1] for token_id in token_ids: if 0 <= token_id < vocabulary_size: logits[..., token_id] = -math.inf def _dplm_special_id( model: _MaskedLanguageModel, tokenizer: object | None, name: str, default: int, ) -> int: value = getattr(model.config, name, None) if value is None: if tokenizer is None: tokenizer = getattr(model, "tokenizer", None) value = getattr(tokenizer, name, None) return default if value is None else int(value) def _dplm_resample_repeats( model: _MaskedLanguageModel, candidate_tokens: torch.Tensor, candidate_scores: torch.Tensor, *, invalid_token_ids: tuple[int, ...], mask_token_id: int, ratio: float, ) -> None: # candidate_tokens, candidate_scores: (b, l) selected_rows: list[int] = [] resample_tokens: list[torch.Tensor] = [] resample_scores: list[torch.Tensor] = [] resample_masks: list[torch.Tensor] = [] for row_index, row in enumerate(candidate_tokens): positions: dict[int, list[int]] = {} for position, token in enumerate(row.tolist()): positions.setdefault(int(token), []).append(position) repeated = [indices for indices in positions.values() if len(indices) > row.numel() * ratio] if not repeated: continue M = torch.zeros_like(row, dtype=torch.bool) # (l,) for indices in repeated: M[indices] = True selected_rows.append(row_index) resample_masks.append(M) # (l,) resample_tokens.append(row.masked_fill(M, mask_token_id)) # (l,) resample_scores.append(candidate_scores[row_index]) # (l,) if not selected_rows: return X = torch.stack(resample_tokens) # (r, l) S = torch.stack(resample_scores) # (r, l) M = torch.stack(resample_masks) # (r, l) with _temporary_eval(model), torch.no_grad(): logits = _logits(model(input_ids=X, return_dict=True)) # (r, l, c) if logits.dtype != S.dtype: logits = logits.to(S.dtype) # (r, l, c) _suppress_token_ids(logits, invalid_token_ids) logits = _top_p(logits) # (r, l, c) sampled_tokens, sampled_scores = _gumbel_argmax(logits, noise_scale=1.0) # (r, l), (r, l) X.masked_scatter_(M, sampled_tokens[M]) S.masked_scatter_(M, sampled_scores[M]) candidate_tokens[selected_rows] = X candidate_scores[selected_rows] = S def generate_dplm( model: _MaskedLanguageModel, input_tokens: torch.Tensor, *, tokenizer: object | None = None, max_iter: int | None = None, temperature: float | None = None, partial_masks: torch.Tensor | None = None, sampling_strategy: str = "gumbel_argmax", disable_resample: bool = False, resample_ratio: float = 0.25, show_progress: bool = False, ) -> torch.Tensor: """Generate DPLM sequences with the official iterative unmasking process. ``input_tokens`` is X with shape (b, l). ``partial_masks=True`` marks fixed positions. The return value is the generated token tensor X with shape (b, l), matching the official DPLM public API. """ partial_masks = _validate_inputs(input_tokens, partial_masks) max_iter = _resolve_max_iter(model, max_iter) # Upstream treats ``None`` as the falsey, zero-temperature branch for # vanilla categorical sampling. Gumbel and argmax strategies ignore it. temperature = _validate_temperature(temperature, default=0.0) if sampling_strategy not in {"vanilla", "argmax", "gumbel_argmax"}: raise ValueError(f"Unsupported DPLM sampling strategy: {sampling_strategy!r}") if not 0 < float(resample_ratio) <= 1: raise ValueError("resample_ratio must be in (0, 1]") pad_id = _dplm_special_id(model, tokenizer, "pad_token_id", 1) bos_id = _dplm_special_id(model, tokenizer, "bos_token_id", 0) eos_id = _dplm_special_id(model, tokenizer, "eos_token_id", 2) mask_id = _dplm_special_id(model, tokenizer, "mask_token_id", 32) x_id = 24 X = input_tokens.clone() # (b, l) mutable = X.ne(pad_id) & X.ne(bos_id) & X.ne(eos_id) # (b, l) if partial_masks is not None: mutable &= ~partial_masks X.masked_fill_(mutable, mask_id) S = torch.zeros_like(X, dtype=torch.float32) # (b, l) active = mutable.clone() # (b, l) invalid_ids = (mask_id, x_id, pad_id, bos_id, eos_id) for step in _steps(max_iter, show_progress=show_progress): with _temporary_eval(model), torch.no_grad(): logits = _logits(model(input_ids=X, return_dict=True)) # (b, l, c) if logits.dtype != S.dtype: logits = logits.to(S.dtype) # (b, l, c) _suppress_token_ids(logits, invalid_ids) if sampling_strategy == "vanilla": candidate_tokens, candidate_scores = _categorical( # (b, l), (b, l) logits, temperature=temperature, ) elif sampling_strategy == "argmax": candidate_scores, candidate_tokens = logits.max(dim=-1) # (b, l), (b, l) else: candidate_tokens, candidate_scores = _gumbel_argmax( # (b, l), (b, l) logits, noise_scale=1.0, ) if not disable_resample: _dplm_resample_repeats( model, candidate_tokens, candidate_scores, invalid_token_ids=invalid_ids, mask_token_id=mask_id, ratio=float(resample_ratio), ) eligible = X.ne(pad_id) & X.ne(bos_id) & X.ne(eos_id) # (b, l) if partial_masks is not None: eligible &= ~partial_masks rate = 1.0 - (step + 1) / max_iter active, X, S = _reparameterize( # (b, l), (b, l), (b, l) X.clone(), S.clone(), candidate_tokens, candidate_scores, active, eligible, mask_token_id=mask_id, rate=rate, ) return X # (b, l) def _normalize_dplm2_special_ids(X: torch.Tensor, vocabulary_size: int) -> torch.Tensor: # X: (b, l) normalized = X.clone() # (b, l) replacements = { vocabulary_size: _DPLM2_AA_EOS, vocabulary_size + 1: _DPLM2_AA_UNK, vocabulary_size + 2: _DPLM2_AA_BOS, vocabulary_size + 3: _DPLM2_AA_MASK, } for generic_id, native_id in replacements.items(): normalized.masked_fill_(X.eq(generic_id), native_id) return normalized # (b, l) def _dplm2_types(X: torch.Tensor) -> torch.Tensor: # X: (b, l) valid = X.ne(_DPLM2_PAD) # (b, l) types = ((X < _DPLM2_AA_BOUNDARY) & valid).to(torch.int64) # (b, l) types.masked_fill_(~valid, 2) return types # (b, l) def _dplm2_mutable(X: torch.Tensor, partial_masks: torch.Tensor | None) -> torch.Tensor: # X, partial_masks: (b, l) mutable = ( # (b, l) X.ne(_DPLM2_PAD) & X.ne(_DPLM2_AA_BOS) & X.ne(_DPLM2_AA_EOS) & X.ne(_DPLM2_STRUCT_BOS) & X.ne(_DPLM2_STRUCT_EOS) ) if partial_masks is not None: mutable &= ~partial_masks return mutable # (b, l) def _dplm2_unmasking_temperature(strategy: str) -> float | None: if strategy == "deterministic": return None if strategy.startswith("stochastic"): suffix = strategy.removeprefix("stochastic") value = 1.0 if not suffix else float(suffix) if not math.isfinite(value) or value < 0: raise ValueError("The stochastic unmasking temperature must be non-negative") return value raise ValueError(f"Unsupported DPLM2 unmasking strategy: {strategy!r}") def _annealing_temperature(strategy: str, step: int, max_iter: int) -> float | None: if not strategy.startswith("annealing"): return None try: maximum, minimum = map(float, strategy.split("@", maxsplit=1)[1].split(":")) except (IndexError, ValueError) as error: raise ValueError("Annealing must use the form 'annealing@maximum:minimum'") from error if not all(math.isfinite(value) and value >= 0 for value in (maximum, minimum)): raise ValueError("Annealing temperatures must be finite and non-negative") rate = 1.0 - step / max_iter return minimum + (maximum - minimum) * rate def generate_dplm2( model: _MaskedLanguageModel, input_tokens: torch.Tensor, *, max_iter: int | None = None, temperature: float = 1.0, partial_masks: torch.Tensor | None = None, unmasking_strategy: str = "stochastic1.0", sampling_strategy: str = "annealing@2.0:0.1", show_progress: bool = False, ) -> dict[str, torch.Tensor]: """Generate packed DPLM2 sequence and structure tracks. ``input_tokens`` is X with shape (b, l). A packed co-generation input has two equal-length modality tracks. ``partial_masks=True`` marks fixed positions. The output mapping matches the official DPLM2 public API. """ partial_masks = _validate_inputs(input_tokens, partial_masks) max_iter = _resolve_max_iter(model, max_iter) temperature = _validate_temperature(temperature) unmasking_temperature = _dplm2_unmasking_temperature(unmasking_strategy) if sampling_strategy.startswith("annealing"): _annealing_temperature(sampling_strategy, 0, max_iter) elif sampling_strategy not in {"argmax", "gumbel_argmax"}: raise ValueError(f"Unsupported DPLM2 sampling strategy: {sampling_strategy!r}") vocabulary_size = int(model.config.vocab_size) if vocabulary_size <= _DPLM2_STRUCT_UNK + 1: raise ValueError("DPLM2 generation requires the multimodal vocabulary") struct_mask_id = vocabulary_size - 1 X = _normalize_dplm2_special_ids(input_tokens, vocabulary_size) # (b, l) if X.numel() and (X.min() < 0 or X.max() >= vocabulary_size): raise ValueError("input_tokens contains an ID outside the DPLM2 vocabulary") mutable = _dplm2_mutable(X, partial_masks) # (b, l) types = _dplm2_types(X) # (b, l) X.masked_fill_(mutable & types.eq(1), _DPLM2_AA_MASK) X.masked_fill_(mutable & types.eq(0), struct_mask_id) S = torch.zeros_like(X, dtype=torch.float32) # (b, l) active = mutable.clone() # (b, l) invalid_ids = ( _DPLM2_AA_BOS, _DPLM2_AA_EOS, _DPLM2_AA_MASK, _DPLM2_STRUCT_BOS, _DPLM2_STRUCT_EOS, struct_mask_id, _DPLM2_PAD, _DPLM2_AA_UNK, _DPLM2_STRUCT_UNK, _DPLM2_AA_X, _DPLM2_AA_B, _DPLM2_AA_U, _DPLM2_AA_Z, _DPLM2_AA_O, ) for step in _steps(max_iter, show_progress=show_progress): eligible = _dplm2_mutable(X, partial_masks) # (b, l) types = _dplm2_types(X) # (b, l) with _temporary_eval(model), torch.no_grad(): logits = _logits( # (b, l, c) model(input_ids=X, return_dict=True) ).log_softmax(dim=-1) if logits.dtype != S.dtype: logits = logits.to(S.dtype) # (b, l, c) aa_rows, aa_columns = torch.where(types.eq(1) & eligible) # (n_aa,), (n_aa,) struct_rows, struct_columns = torch.where( # (n_struct,), (n_struct,) types.eq(0) & eligible ) logits[aa_rows, aa_columns, _DPLM2_AA_BOUNDARY:] = -math.inf logits[struct_rows, struct_columns, :_DPLM2_AA_BOUNDARY] = -math.inf _suppress_token_ids(logits, invalid_ids) logits = _top_p(logits) # (b, l, c) if sampling_strategy == "argmax": candidate_scores, candidate_tokens = logits.max(dim=-1) # (b, l), (b, l) elif sampling_strategy == "gumbel_argmax": candidate_tokens, candidate_scores = _gumbel_argmax( # (b, l), (b, l) logits, noise_scale=temperature, ) candidate_tokens.masked_scatter_(~eligible, X[~eligible]) else: annealed = _annealing_temperature(sampling_strategy, step, max_iter) sample_temperature = temperature if annealed is None else annealed candidate_tokens, candidate_scores = _categorical( # (b, l), (b, l) logits, temperature=sample_temperature, ) rate = 1.0 - (step + 1) / max_iter new_active = torch.zeros_like(active) # (b, l) for modality, mask_id in ((1, _DPLM2_AA_MASK), (0, struct_mask_id)): modality_positions = types.eq(modality) & eligible # (b, l) if not bool(modality_positions.any()): continue modality_active, X, S = _reparameterize( # (b, l), (b, l), (b, l) X, S, candidate_tokens, candidate_scores, active, modality_positions, mask_token_id=mask_id, rate=rate, stochastic_temperature=unmasking_temperature, ) new_active |= modality_active active = new_active # (b, l) return {"output_tokens": X} # (b, l) __all__ = ["generate_dplm", "generate_dplm2"]