chat_utils.py

from __future__ import annotations
import numpy as np
import torch
import torch.nn.functional as F
from torch.nn.attention.flex_attention import create_block_mask
from transformers.utils import ModelOutput
from dataclasses import dataclass
from transformers.cache_utils import Cache

@dataclass
class SimpleOutputWithPast(ModelOutput):
    loss: torch.FloatTensor | None = None
    logits: torch.FloatTensor | None = None
    causal_logits: torch.FloatTensor | None = None
    past_key_values: Cache | None = None
    hidden_states: tuple[torch.FloatTensor, ...] | None = None
    attentions: tuple[torch.FloatTensor, ...] | None = None
from .utils import maybe_truncate_last_dim, pad_along_last_dim
from .configuration_nlduni import get_ministral_shift_sbd_logits
from .sbd_attention import build_image_attention_doc_ids

def _resolve_shift_sbd_logits(model, shift_logits):
    model_module = model.module if hasattr(model, 'module') else model
    config = getattr(model_module, 'config', None)
    configured = get_ministral_shift_sbd_logits(config) if config is not None else None
    has_alignment_contract = config is not None
    if shift_logits is None:
        return bool(configured) if configured is not None else False
    requested = bool(shift_logits)
    if has_alignment_contract and requested != configured:
        raise ValueError('Inference shift_logits must match config.shift_sbd_logits to avoid a one-token alignment error. Update the model config explicitly if this override is intentional.')
    return requested

def wte(model, x, t2i_inference=False, gen_shape=None, x_gen=None, inputs_embeds_curr=None, new_token_mask=None, x_gen_child_embeddings=None, gen_parent_mask=None, gen_source_null_mask=None, gen_role=None):
    if t2i_inference:
        if (x_gen is None) == (x_gen_child_embeddings is None):
            raise ValueError('provide exactly one of x_gen or x_gen_child_embeddings')
        if new_token_mask is None:
            new_token_mask = x >= INT_MAX
        if gen_parent_mask is not None or gen_source_null_mask is not None or gen_role is not None:
            raise ValueError('structured generation masks and roles require Infinity BSQ')
        gen_latents_comp_embeds = model.call_gen_embedding(x_gen, gen_shape, child_embeddings=x_gen_child_embeddings)
        if inputs_embeds_curr is None:
            x_txt_only = x.clone()
            x_txt_only[new_token_mask] = 0
            inputs_embeds_curr = model.embed_tokens(x_txt_only)
        inputs_embeds_curr[new_token_mask] = pad_along_last_dim(gen_latents_comp_embeds, inputs_embeds_curr.shape[-1]).view(-1, inputs_embeds_curr.shape[-1])
    else:
        inputs_embeds_curr = model.embed_tokens(x)
        new_token_mask = None
    return (inputs_embeds_curr, new_token_mask)
INT_MAX = 1000000
SBD_FLEX_ATTENTION_PAD_MULTIPLE = 512

def _round_up_to_multiple(length, multiple):
    if isinstance(multiple, bool) or not isinstance(multiple, int) or multiple < 1:
        raise ValueError('padding multiple must be a positive integer')
    return (length + multiple - 1) // multiple * multiple

def single_stream_sbd_sampling_mask(block_size, past_length, b, h, q_idx, kv_idx, query_image_doc_id, key_value_image_doc_id):
    del block_size, h
    absolute_q = q_idx + past_length
    token_causal = absolute_q >= kv_idx
    image_query = query_image_doc_id[b, q_idx] > 0
    same_image_doc = image_query & (query_image_doc_id[b, q_idx] == key_value_image_doc_id[b, kv_idx])
    return token_causal | same_image_doc

def _create_single_stream_sbd_sampling_block_mask(input_embeddings, *, past_key_values, block_size, input_modality_indices, generation_token_mask=None, pad_to_multiple=None):
    if isinstance(block_size, bool) or not isinstance(block_size, int) or block_size < 1:
        raise ValueError('SBD diffusion sampling requires a positive integer block_size')
    query_length = input_embeddings.shape[1]
    past_length = past_key_values.get_seq_length() if past_key_values is not None else 0
    key_value_length = past_length + query_length
    padded_query_length = _round_up_to_multiple(query_length, pad_to_multiple) if pad_to_multiple is not None else query_length
    padded_key_value_length = _round_up_to_multiple(key_value_length, pad_to_multiple) if pad_to_multiple is not None else key_value_length
    if input_modality_indices is None:
        raise ValueError('SBD diffusion sampling requires input_modality_indices to route same-block generation attention')
    if tuple(input_modality_indices.shape) != tuple(input_embeddings.shape[:2]):
        raise ValueError('SBD diffusion sampling requires input_modality_indices to align with the current input embeddings')
    query_transformer_routing = input_modality_indices.to(dtype=torch.bool)
    if generation_token_mask is None:
        generation_token_mask = query_transformer_routing
    elif tuple(generation_token_mask.shape) != tuple(input_embeddings.shape[:2]):
        raise ValueError('SBD diffusion sampling requires generation_token_mask to align with the current input embeddings')
    query_target_generation_routing = generation_token_mask.to(dtype=torch.bool)
    cached_generation_routing = torch.zeros(query_transformer_routing.shape[0], past_length, dtype=torch.bool, device=query_transformer_routing.device)
    padded_generation_routing = torch.zeros(query_transformer_routing.shape[0], padded_key_value_length - key_value_length, dtype=torch.bool, device=query_transformer_routing.device)
    key_value_target_generation_routing = torch.cat((cached_generation_routing, query_target_generation_routing, padded_generation_routing), dim=1)
    key_value_valid = (torch.arange(padded_key_value_length, device=query_transformer_routing.device).unsqueeze(0) < key_value_length).expand(query_transformer_routing.shape[0], -1).contiguous()
    key_value_image_doc_id = build_image_attention_doc_ids(key_value_target_generation_routing, valid_tokens=key_value_valid)
    query_image_doc_id = torch.cat((key_value_image_doc_id[:, past_length:past_length + query_length], torch.zeros(query_transformer_routing.shape[0], padded_query_length - query_length, dtype=key_value_image_doc_id.dtype, device=query_transformer_routing.device)), dim=1)
    past_length_tensor = torch.tensor(past_length, dtype=torch.long, device=query_transformer_routing.device)
    query_valid = (torch.arange(padded_query_length, device=query_transformer_routing.device).unsqueeze(0) < query_length).expand(query_transformer_routing.shape[0], -1).contiguous()

    def mask_mod(b, h, q_idx, kv_idx):
        logical_attention = single_stream_sbd_sampling_mask(block_size, past_length_tensor, b, h, q_idx, kv_idx, query_image_doc_id, key_value_image_doc_id) & query_valid[b, q_idx] & key_value_valid[b, kv_idx]
        padded_query_attention = ~query_valid[b, q_idx] & (kv_idx == 0)
        return logical_attention | padded_query_attention
    return create_block_mask(mask_mod, B=input_embeddings.shape[0], H=None, Q_LEN=padded_query_length, KV_LEN=padded_key_value_length, device=input_embeddings.device)

def get_logits(model, input_emnbeddings, modality_indices=None, t2i_inference=False, past_key_values=None, gen_shape=None, timesteps=None, input_modality_indices=None, generation_attention_mask=None, sbd_diffusion_sampling=False, full_attention_approx=False, loop_k=None):
    if t2i_inference:
        resolved_loop_k = None
        if input_modality_indices is None:
            input_modality_indices = modality_indices
        attention_kwargs = {}
        if full_attention_approx:
            attention_kwargs['overwrite_block_mask'] = 'full'
            attention_kwargs['overwrite_attn_impl'] = 'flash_attn'
        elif sbd_diffusion_sampling:
            if generation_attention_mask is None:
                generation_attention_mask = modality_indices
            attention_kwargs['overwrite_block_mask'] = _create_single_stream_sbd_sampling_block_mask(input_emnbeddings, past_key_values=past_key_values, block_size=getattr(model.config, 'block_size', None), input_modality_indices=input_modality_indices, generation_token_mask=generation_attention_mask, pad_to_multiple=SBD_FLEX_ATTENTION_PAD_MULTIPLE)
        else:
            attention_kwargs['overwrite_attn_impl'] = 'flash_attn'
        output = model(None, input_embeddings=input_emnbeddings, modality_indices=input_modality_indices, output_hidden_states=True, past_key_values=past_key_values, use_cache=None, single_stream_image_diffusion=True, loop_k=None, **attention_kwargs)
        hidden_states = getattr(output, 'last_hidden_state', None)
        if hidden_states is None:
            hidden_states = output.hidden_states[-1]
        gen_hidden_states = hidden_states[modality_indices]
        gen_hidden_states = maybe_truncate_last_dim(gen_hidden_states, model.config.d_model_gen)
        gen_logits = model.call_gen_predictor(gen_hidden_states, gen_shape, timesteps=timesteps)
        seq_len_per_img = np.prod(gen_shape)
        if len(gen_logits.shape) == 2:
            gen_logits = gen_logits.view(-1, seq_len_per_img, gen_logits.shape[-1])
        else:
            gen_logits = gen_logits.view(-1, seq_len_per_img, *gen_logits.shape[-2:])
        return gen_logits
    else:
        raise ValueError('get_logits is the image sampler helper')

def add_gumbel_noise(logits, temperature):
    if temperature == 0:
        return logits
    logits = logits.to(torch.float64)
    noise = torch.rand_like(logits, dtype=torch.float64)
    gumbel_noise = (-torch.log(noise)) ** temperature
    return logits.exp() / gumbel_noise

def get_transfer_index(logits, temperature, remasking, mask_index, x, num_transfer_tokens, threshold=None, neg_entropy=False):
    logits_with_noise = add_gumbel_noise(logits, temperature=temperature)
    x0 = torch.argmax(logits_with_noise, dim=-1)
    if remasking == 'low_confidence':
        p = F.softmax(logits, dim=-1)
        x0_p = torch.squeeze(torch.gather(p, dim=-1, index=torch.unsqueeze(x0, -1)), -1)
    elif remasking == 'top_p_margin':
        p = F.softmax(logits, dim=-1)
        top2 = torch.topk(p, k=2, dim=-1).values
        margin = top2[..., 0] - top2[..., 1]
        plus_inf = torch.full_like(margin, 1e309)
        minus_inf = torch.full_like(margin, -1e309)
        masked_for_min = torch.where(mask_index, margin, plus_inf)
        masked_for_max = torch.where(mask_index, margin, minus_inf)
        row_min = masked_for_min.amin(dim=1, keepdim=True)
        row_max = masked_for_max.amax(dim=1, keepdim=True)
        denom = row_max - row_min
        normalized = torch.zeros_like(margin)
        nonzero = denom > 0
        normalized = torch.where(mask_index & nonzero, (margin - row_min) / (denom + 1e-12), normalized)
        normalized = torch.where(mask_index & ~nonzero, torch.ones_like(normalized), normalized)
        x0_p = normalized
    elif remasking == 'random':
        x0_p = torch.rand((x0.shape[0], x0.shape[1]), device=x0.device)
    else:
        raise NotImplementedError(remasking)
    if neg_entropy:
        p = F.softmax(logits, dim=-1)
        epsilon = 1e-10
        log_probs = torch.log(p + epsilon)
        confidence_scores = torch.sum(p * log_probs, dim=-1)
    else:
        confidence_scores = x0_p
    x0 = torch.where(mask_index, x0, x)
    confidence = torch.where(mask_index, confidence_scores, -np.inf)
    transfer_index = torch.zeros_like(x0, dtype=torch.bool, device=x0.device)
    if threshold is not None:
        num_transfer_tokens = mask_index.sum(dim=1, keepdim=True)
    for j in range(confidence.shape[0]):
        (_, select_index) = torch.topk(confidence[j], k=num_transfer_tokens[j])
        transfer_index[j, select_index] = True
        if threshold is not None:
            for k in range(1, num_transfer_tokens[j]):
                if confidence[j, select_index[k]] < threshold:
                    transfer_index[j, select_index[k]] = False
    return (x0, transfer_index)

def get_num_transfer_tokens(mask_index, steps: int):
    mask_num = mask_index.sum(dim=1, keepdim=True)
    base = mask_num // steps
    remainder = mask_num % steps
    num_transfer_tokens = torch.zeros(mask_num.size(0), steps, device=mask_index.device, dtype=torch.int64) + base
    for i in range(mask_num.size(0)):
        num_transfer_tokens[i, :int(remainder[i])] += 1
    return num_transfer_tokens

def simple_fwd(model, input_ids=None, inputs_embeds=None, attention_mask=None, position_ids=None, past_key_values=None, **kwargs):
    enc_out = model.encoder(past_key_values=past_key_values, input_ids=input_ids, inputs_embeds=inputs_embeds, attention_mask=attention_mask, position_ids=position_ids, overwrite_attn_impl='flash_attn', **kwargs)
    logits = model.diffusion_head(enc_out.last_hidden_state)
    return SimpleOutputWithPast(loss=logits, logits=logits, causal_logits=None, past_key_values=enc_out.past_key_values, hidden_states=None, attentions=None)

@torch.no_grad()
def generate_with_prefix_cache_block_diff(model, prompt=None, prompt_embeds=None, steps=128, gen_length=128, block_length=128, temperature=0.0, remasking='low_confidence', mask_id=126336, threshold=None, factor=None, shift_logits=None, neg_entropy=False, causal_context=False, eos_token_id=None, max_thinking_tokens=None, end_think_token_id=None):
    shift_logits = _resolve_shift_sbd_logits(model, shift_logits)
    dream_style = shift_logits
    if (prompt is None) == (prompt_embeds is None):
        raise ValueError('Exactly one of `prompt` or `prompt_embeds` must be provided.')
    if prompt is not None:
        prompt_ids = prompt
        prompt_len = prompt_ids.shape[1]
        x_accum = prompt_ids.clone()
        B = prompt_ids.shape[0]
        token_device = prompt_ids.device
        token_dtype = prompt_ids.dtype
    else:
        prompt_ids = None
        prompt_len = prompt_embeds.shape[1]
        B = prompt_embeds.shape[0]
        token_device = prompt_embeds.device
        token_dtype = torch.long
        x_accum = torch.full((B, prompt_len), mask_id, dtype=token_dtype, device=token_device)
    if gen_length % block_length != 0:
        gen_length = (gen_length // block_length + 1) * block_length
    num_blocks = gen_length // block_length
    assert steps % num_blocks == 0
    steps_per_block = steps // num_blocks
    nfe = 0
    model_module = model.module if hasattr(model, 'module') else model
    for layer in model_module.encoder.layers:
        layer.self_attn.mode = 'bidirectional'
    if causal_context:
        for layer in model_module.encoder.layers:
            if hasattr(layer.self_attn, 'diffusion_lm'):
                layer.self_attn.diffusion_lm = False
    output = simple_fwd(model, input_ids=prompt_ids, inputs_embeds=prompt_embeds, use_cache=True, use_causal_mask=causal_context)
    past_key_values = output.past_key_values
    if causal_context:
        for layer in model_module.encoder.layers:
            if hasattr(layer.self_attn, 'diffusion_lm'):
                layer.self_attn.diffusion_lm = True
    next_token = None
    if causal_context:
        last_logit = output.logits[:, -1, :]
        if temperature > 0:
            probs = torch.softmax(last_logit / temperature, dim=-1)
            next_token = torch.multinomial(probs, num_samples=1)
        else:
            next_token = torch.argmax(last_logit, dim=-1, keepdim=True)
    next_logits_context = None
    if dream_style:
        next_logits_context = output.logits[:, -1:, :]
    for num_block in range(num_blocks):
        mask_block = torch.ones((B, block_length), dtype=token_dtype, device=token_device) * mask_id
        if causal_context:
            mask_block[:, 0] = next_token[:, 0]
        x_accum = torch.cat([x_accum, mask_block], dim=1)
        current_block_start = prompt_len + num_block * block_length
        block_slice = slice(current_block_start, current_block_start + block_length)
        if end_think_token_id is not None and max_thinking_tokens is not None:
            tokens_before_block = num_block * block_length
            tokens_after_block = tokens_before_block + block_length
            if tokens_after_block > max_thinking_tokens:
                gen_so_far = x_accum[:, prompt_len:current_block_start]
                has_end_think = (gen_so_far == end_think_token_id).any(dim=1) if gen_so_far.size(1) > 0 else torch.zeros(B, dtype=torch.bool, device=token_device)
                if not has_end_think.all():
                    if tokens_before_block < max_thinking_tokens:
                        offset = max_thinking_tokens - tokens_before_block
                    else:
                        offset = 0
                    inject_pos = current_block_start + offset
                    for b in range(B):
                        if not has_end_think[b]:
                            x_accum[b, inject_pos] = end_think_token_id
        mask_block_idx0 = x_accum[:, block_slice] == mask_id
        if dream_style:
            schedule_mask = mask_block_idx0
        else:
            schedule_mask = mask_block_idx0
        num_transfer_tokens = get_num_transfer_tokens(schedule_mask, steps_per_block)
        for i in range(steps_per_block):
            mask_block_idx = x_accum[:, block_slice] == mask_id
            if mask_block_idx.sum() == 0:
                break
            nfe += 1
            logits_block = simple_fwd(model, x_accum[:, block_slice], past_key_values=past_key_values, use_cache=False).logits
            if dream_style:
                if block_length == 1:
                    logits_use = next_logits_context
                else:
                    logits_use = torch.cat([next_logits_context, logits_block[:, :-1, :]], dim=1)
                mask_use = mask_block_idx
                x_use = x_accum[:, block_slice]
                (x0, transfer_idx) = get_transfer_index(logits_use, temperature, remasking, mask_use, x_use, num_transfer_tokens=num_transfer_tokens[:, i], threshold=threshold, neg_entropy=neg_entropy)
                cur = x_accum[:, block_slice].clone()
                cur[transfer_idx] = x0[transfer_idx]
                x_accum[:, block_slice] = cur
            else:
                (x0, transfer_idx) = get_transfer_index(logits_block, temperature, remasking, mask_block_idx, x_accum[:, block_slice], num_transfer_tokens=num_transfer_tokens[:, i], threshold=threshold, neg_entropy=neg_entropy)
                cur = x_accum[:, block_slice].clone()
                cur[transfer_idx] = x0[transfer_idx]
                x_accum[:, block_slice] = cur
            if eos_token_id is not None:
                block_tokens = x_accum[:, block_slice]
                eos_mask = block_tokens == eos_token_id
                any_eos = eos_mask.any(dim=1)
                if any_eos.any():
                    after_eos = eos_mask.cumsum(dim=1).bool()
                    mask_before = (block_tokens == mask_id) & ~after_eos
                    if (any_eos & ~mask_before.any(dim=1)).any():
                        break
        if causal_context:
            for layer in model_module.encoder.layers:
                if hasattr(layer.self_attn, 'diffusion_lm'):
                    layer.self_attn.diffusion_lm = False
        output = simple_fwd(model, x_accum[:, block_slice], past_key_values=past_key_values, use_cache=True, use_causal_mask=causal_context)
        past_key_values = output.past_key_values
        nfe += 1
        if causal_context:
            for layer in model_module.encoder.layers:
                if hasattr(layer.self_attn, 'diffusion_lm'):
                    layer.self_attn.diffusion_lm = True
            last_logit = output.logits[:, -1, :]
            if temperature > 0:
                probs = torch.softmax(last_logit / temperature, dim=-1)
                next_token = torch.multinomial(probs, num_samples=1)
            else:
                next_token = torch.argmax(last_logit, dim=-1, keepdim=True)
        if dream_style and num_block < num_blocks - 1:
            next_logits_context = output.logits[:, -1:, :]
        if eos_token_id is not None:
            gen_so_far = x_accum[:, prompt_len:]
            is_eos = gen_so_far == eos_token_id
            has_eos = is_eos.any(dim=1)
            if has_eos.all():
                first_eos_pos = is_eos.to(torch.int64).argmax(dim=1)
                max_eos = first_eos_pos.max().item()
                if prompt_ids is None:
                    return (x_accum[:, prompt_len:prompt_len + max_eos + 1], nfe)
                return (x_accum[:, :prompt_len + max_eos + 1], nfe)
    if prompt_ids is None:
        return (x_accum[:, prompt_len:], nfe)
    return (x_accum, nfe)