src/runtime/yue2/sampling.py

"""Request-local sampling, preserving mode-specific historical arithmetic."""
from __future__ import annotations
import time
import torch
from .protocol import EOD, ABC_END, MUSIC_END, CODEC_OFFSET, CODEC_SIZE, CONTEXT


def synchronize(device):
    device = torch.device(device)
    if device.type == "cuda":
        torch.cuda.synchronize(device)
    elif device.type == "mps":
        torch.mps.synchronize()


def window_penalty(logits, recent_ids, penalty):
    if penalty == 1.0 or len(recent_ids) == 0:
        return logits
    recent = torch.as_tensor(recent_ids, dtype=torch.long, device=logits.device).reshape(1, -1)
    freq = torch.zeros_like(logits)
    freq.scatter_add_(-1, recent, torch.ones_like(recent, dtype=logits.dtype))
    alpha = penalty ** freq
    return torch.where(logits < 0, logits * alpha, logits / alpha)


def distribution(logits, sampling, history, step, phase, legacy_off=False):
    # vLLM's symbolic processor receives FP32 logits; historical off uses BF16.
    scores = logits.clone() if legacy_off else logits.float().clone()
    end = ABC_END if phase == "abc" else MUSIC_END
    allowed = torch.full_like(scores, float("-inf"))
    if phase == "abc":
        allowed[..., :EOD] = 0
    else:
        allowed[..., CODEC_OFFSET:CODEC_OFFSET + CODEC_SIZE] = 0
    allowed[..., end] = 0
    scores = scores + allowed
    if step < sampling.min_tokens:
        scores[..., end] = -torch.inf
    scores = window_penalty(scores, history[-sampling.penalty_window:], sampling.repetition_penalty)
    if sampling.temperature == 0:
        return scores
    if sampling.temperature != 1:
        scores = scores / sampling.temperature
    threshold = scores.topk(min(sampling.top_k, scores.shape[-1])).values[..., -1, None]
    scores = scores.masked_fill(scores < threshold, -torch.inf)
    if sampling.top_p < 1:
        values, indices = scores.sort(descending=True)
        probabilities = values.softmax(-1)
        removed = probabilities.cumsum(-1) - probabilities > sampling.top_p
        removed[..., :3 if legacy_off else 1] = False
        values = values.masked_fill(removed, -torch.inf)
        scores = values.scatter(-1, indices, values)
    return scores


@torch.inference_mode()
def generate_tokens(model, prefix, sampling, seed, phase, negative=None, cfg_scale=1.0,
                    legacy_off=False, cancelled=None, on_token=None, use_cuda_graph=True):
    from .modeling_yue2 import StaticKVCache
    device = next(model.parameters()).device
    dtype = next(model.parameters()).dtype
    if (device.type == "cuda" and use_cuda_graph and sampling.temperature > 0 and not legacy_off
            and getattr(model, "_captured_sampling", False) and getattr(model, "_compact_sampling", False)
            and not getattr(model, "_yue2_fp8_originals", {})):
        from .combined_sampling import generate_combined_tokens
        return generate_combined_tokens(model, prefix, sampling, seed, phase, negative=negative,
            cfg_scale=cfg_scale, legacy_off=legacy_off, cancelled=cancelled, on_token=on_token,
            use_cuda_graph=use_cuda_graph)
    if len(prefix) + sampling.max_tokens > CONTEXT:
        raise ValueError("Prefix + requested generation budget exceeds 24576; no implicit truncation")
    if cfg_scale != 1 and negative is None:
        raise ValueError("CFG requires a negative prefix")
    if negative is not None and len(negative) + sampling.max_tokens > CONTEXT:
        raise ValueError("Negative prefix + generation budget exceeds context")
    if cancelled is not None and cancelled():
        raise InterruptedError("Cancelled before prefill")
    # The two stages deliberately reset their request-local seed, matching the preset.
    rng_device = device if device.type in {"cpu", "cuda"} else torch.device("cpu")
    generator = torch.Generator(device=rng_device).manual_seed(seed)
    config = model.config

    def prefill(ids):
        cache = StaticKVCache(num_layers=config.num_hidden_layers, batch_size=1,
                              num_kv_heads=config.num_key_value_heads,
                              max_seq_len=len(ids) + sampling.max_tokens,
                              head_dim=config.head_dim, dtype=dtype, device=device)
        output = model(torch.tensor([ids], device=device), past_key_values=cache,
                       use_cache=True, logits_to_keep=1)
        return output.logits[:, -1, :], output.past_key_values

    graph = None
    positive_cache = negative_cache = None
    graph_enabled = use_cuda_graph and device.type == "cuda" and not getattr(model, "_yue2_fp8_originals", {})
    synchronize(device)
    start = time.perf_counter()
    old_phase = getattr(model.lm_head, "_semantic_phase", False)
    model.lm_head._semantic_phase = phase == "semantic"
    try:
        if graph_enabled:
            from .cuda_graph import GraphAR
            graph = GraphAR(model, [prefix] if cfg_scale == 1 else [prefix, negative], sampling.max_tokens)
            logits = graph.prefill()
            conditional = logits[:1]
            unconditional = logits[1:] if cfg_scale != 1 else None
        else:
            conditional, positive_cache = prefill(prefix)
            unconditional = None
            if cfg_scale != 1.0:
                unconditional, negative_cache = prefill(negative)
        synchronize(device)
        prefill_seconds = time.perf_counter() - start
        history, first, eos = [], None, False
        end = ABC_END if phase == "abc" else MUSIC_END
        for step in range(sampling.max_tokens):
            if cancelled is not None and cancelled():
                raise InterruptedError(f"Cancelled during {phase}")
            # Preserve historical BF16 CFG subtraction/multiply/add before upcast.
            logits = conditional if cfg_scale == 1.0 else unconditional + cfg_scale * (conditional - unconditional)
            if getattr(model, "_compact_sampling", False) and not legacy_off:
                from .compact_sampling import sample_topk
                next_id = sample_topk(logits, sampling, history, step, phase, legacy_off, generator)
            else:
                scores = distribution(logits, sampling, history, step, phase, legacy_off)
                if sampling.temperature == 0:
                    next_id = scores.argmax(-1, keepdim=True)
                else:
                    probabilities = scores.softmax(-1)
                    if device.type == "mps":
                        next_id = torch.multinomial(probabilities.cpu(), 1, generator=generator).to(device)
                    else:
                        next_id = torch.multinomial(probabilities, 1, generator=generator)
            token = int(next_id.item())
            if first is None:
                first = time.perf_counter() - start
            if on_token is not None:
                on_token(phase, token)
            if token == end:
                eos = True
                break
            history.append(token)
            if step + 1 < sampling.max_tokens:
                if graph is not None:
                    branch_logits = graph.step(next_id)
                    conditional = branch_logits[:1]
                    unconditional = branch_logits[1:] if cfg_scale != 1 else None
                else:
                    conditional = model(next_id, past_key_values=positive_cache, use_cache=True,
                                        logits_to_keep=1).logits[:, -1, :]
                    if negative_cache is not None:
                        unconditional = model(next_id, past_key_values=negative_cache, use_cache=True,
                                              logits_to_keep=1).logits[:, -1, :]
        synchronize(device)
        seconds = time.perf_counter() - start
        count = len(history) + int(eos)
        timing = {"seconds": seconds, "prefill_seconds": prefill_seconds,
                  "ttft_seconds": first, "output_tokens": count, "content_tokens": len(history),
                  "output_tps": count / seconds, "prefix_tokens": len(prefix),
                  "cfg_branches": 1 if cfg_scale == 1 else 2,
                  "sampling": "eager",
                  "execution": "cuda_graph" if graph is not None else "eager",
                  "attention": graph.attention_backend if graph is not None else "sdpa"}
        if use_cuda_graph and not graph_enabled:
            timing["graph_fallback_reason"] = "fp8_not_graph_validated" if getattr(model, "_yue2_fp8_originals", {}) else "non_cuda_device"
        return history, timing, not eos
    finally:
        model.lm_head._semantic_phase = old_phase
        if graph is not None:
            graph.close()
        positive_cache = negative_cache = None