# inference.py: Decoding helpers shared with the generator: prefix construction, constrained decoding, token extraction. from __future__ import annotations import logging from typing import List, Optional import torch logger = logging.getLogger(__name__) def build_inference_prefix( tokenizer, registry, prompt: str, stage: int, ) -> torch.Tensor: if stage not in (1, 2): raise ValueError(f"Unsupported stage {stage}; must be 1 or 2") prompt_ids = tokenizer.encode(prompt, add_special_tokens=False) if stage == 1: prefix = ( [registry.bos_id, registry.user_prompt_id] + prompt_ids + [registry.audio_start_id] ) else: prefix = ( [registry.bos_id, registry.user_prompt_id] + prompt_ids + [registry.think_start_id] ) return torch.tensor([prefix], dtype=torch.long) def extract_speech_token_ids( generated_ids: List[int], registry, prefix_len: int, ) -> List[int]: if prefix_len > len(generated_ids): return [] tail = generated_ids[prefix_len:] speech_ids: List[int] = [] started = False for tid in tail: if registry.is_speech_token(tid): speech_ids.append(tid) started = True elif started: break elif tid == registry.audio_end_id or tid == registry.eos_id: break return speech_ids def _flatten_generation_output(output) -> List[int]: if hasattr(output, "sequences"): tensor = output.sequences else: tensor = output if isinstance(tensor, torch.Tensor): if tensor.dim() == 2: return tensor[0].tolist() return tensor.tolist() return list(tensor) def _sample_from_logits( logits: torch.Tensor, *, temperature: float, top_p: float, top_k: int, ) -> int: eff_temperature = max(temperature, 1e-4) if top_k and top_k > 0: top_vals, top_ix = torch.topk(logits, k=min(top_k, logits.numel())) masked = torch.full_like(logits, float("-inf")) masked[top_ix] = top_vals logits = masked probs = torch.softmax(logits / eff_temperature, dim=-1) if top_p and top_p < 1.0: sorted_probs, sorted_ix = torch.sort(probs, descending=True) cumulative = torch.cumsum(sorted_probs, dim=-1) cutoff = cumulative > top_p cutoff[..., 0] = False sorted_probs[cutoff] = 0.0 sorted_probs = sorted_probs / sorted_probs.sum().clamp(min=1e-12) next_ix = torch.multinomial(sorted_probs, 1) return int(sorted_ix.gather(-1, next_ix).item()) return int(torch.multinomial(probs, 1).item()) def _build_speech_mask(model, device) -> torch.Tensor: reg = model.token_registry vocab_size = model.backbone.config.vocab_size allowed = torch.zeros(vocab_size, dtype=torch.bool, device=device) allowed[reg.speech_token_id_min : reg.speech_token_id_max + 1] = True return allowed def _speech_constrained_decode( model, prefix: torch.Tensor, *, max_new_tokens: int, temperature: float, top_p: float, top_k: int, level_cycle: int = 0, level_temps: Optional[List[float]] = None, stop_ids: Optional[List[int]] = None, ) -> List[int]: device = prefix.device allowed = _build_speech_mask(model, device) neg_inf = torch.tensor(float("-inf"), device=device) level_masks = None if level_cycle > 0: reg = model.token_registry id_min = reg.speech_token_id_min n_speech = reg.speech_token_id_max - id_min + 1 csize = n_speech // level_cycle level_masks = [] for lvl in range(level_cycle): m = torch.zeros_like(allowed) m[id_min + lvl * csize: id_min + (lvl + 1) * csize] = True level_masks.append(m & allowed) stop_mask = None if stop_ids: stop_mask = torch.zeros_like(allowed) for sid in stop_ids: stop_mask[sid] = True seq = prefix.clone() speech_ids: List[int] = [] with torch.no_grad(): for i in range(max_new_tokens): out = model(input_ids=seq) logits = out["logits"][0, -1].float() step_mask = ( level_masks[i % level_cycle] if level_masks is not None else allowed ) if stop_mask is not None and (level_cycle == 0 or i % level_cycle == 0): step_mask = step_mask | stop_mask logits = torch.where(step_mask, logits, neg_inf) step_temp = temperature if level_masks is not None and level_temps: step_temp = level_temps[(i % level_cycle) % len(level_temps)] next_id = _sample_from_logits( logits, temperature=step_temp, top_p=top_p, top_k=top_k, ) if stop_ids and next_id in stop_ids: break speech_ids.append(next_id) seq = torch.cat( [seq, torch.tensor([[next_id]], device=device, dtype=seq.dtype)], dim=-1, ) return speech_ids def speculative_generate( model, prefix: torch.Tensor, *, max_new_tokens: int, temperature: float = 0.0, speech_only: bool = False, ) -> tuple[List[int], dict]: if temperature > 0: raise NotImplementedError( "speculative_generate currently supports greedy decoding only " "(temperature=0). Sampling-mode speculative requires rejection " "sampling (Leviathan et al.) and is deferred to a later phase." ) reg = model.token_registry device = prefix.device if speech_only: speech_mask = _build_speech_mask(model, device) neg_inf = torch.tensor(float("-inf"), device=device) def constrain(logits: torch.Tensor) -> torch.Tensor: return torch.where(speech_mask, logits, neg_inf) else: def constrain(logits: torch.Tensor) -> torch.Tensor: return logits seq = prefix.clone() generated: List[int] = [] iterations = 0 drafts_accepted = 0 with torch.no_grad(): while len(generated) < max_new_tokens: iterations += 1 out = model(input_ids=seq) main_logits = constrain(out["logits"][0, -1].float()) t = int(main_logits.argmax().item()) generated.append(t) if len(generated) >= max_new_tokens: seq = torch.cat( [seq, torch.tensor([[t]], device=device, dtype=seq.dtype)], dim=-1, ) break hidden_T = out["hidden_states"][0, -1] draft = int(model.mtp_module.predict_offset_2(hidden_T).item()) if speech_only: h_normed = model.mtp_module.spec_shared_norm(hidden_T.unsqueeze(0)) spec_logits = constrain( model.mtp_module.spec_shared_head(h_normed)[0].float() ) draft = int(spec_logits.argmax().item()) seq = torch.cat( [seq, torch.tensor([[t, draft]], device=device, dtype=seq.dtype)], dim=-1, ) out_v = model(input_ids=seq) verify_logits = constrain(out_v["logits"][0, -2].float()) verify_t1 = int(verify_logits.argmax().item()) if verify_t1 == draft: generated.append(draft) drafts_accepted += 1 else: seq = seq[:, :-1] seq[0, -1] = t generated.append(verify_t1) seq = torch.cat( [seq, torch.tensor([[verify_t1]], device=device, dtype=seq.dtype)], dim=-1, ) if len(generated) > max_new_tokens: generated = generated[:max_new_tokens] meta = { "iterations": iterations, "drafts_proposed": iterations, "drafts_accepted": drafts_accepted, "acceptance_rate": drafts_accepted / max(iterations, 1), "tokens_emitted": len(generated), "forward_passes": 2 * iterations, } return generated, meta def _cot_constrained_decode( model, prefix: torch.Tensor, *, max_new_tokens: int, max_think_tokens: int, temperature: float, top_p: float, top_k: int, ) -> tuple[List[int], List[int], dict]: reg = model.token_registry device = prefix.device speech_mask = _build_speech_mask(model, device) neg_inf = torch.tensor(float("-inf"), device=device) seq = prefix.clone() think_ids: List[int] = [] speech_ids: List[int] = [] meta = {"think_truncated": False, "transition": None} THINKING, AUDIO = 0, 1 state = THINKING think_count = 0 tokens_used = 0 def emit(token_id: int): nonlocal seq, tokens_used seq = torch.cat( [seq, torch.tensor([[token_id]], device=device, dtype=seq.dtype)], dim=-1, ) tokens_used += 1 with torch.no_grad(): while tokens_used < max_new_tokens: if state == THINKING: if think_count >= max_think_tokens: meta["think_truncated"] = True meta["transition"] = "forced" emit(reg.think_end_id) if tokens_used < max_new_tokens: emit(reg.audio_start_id) state = AUDIO continue out = model(input_ids=seq) logits = out["logits"][0, -1].float() next_id = _sample_from_logits( logits, temperature=temperature, top_p=top_p, top_k=top_k, ) if next_id == reg.think_end_id: meta["transition"] = "think_end" emit(next_id) if tokens_used < max_new_tokens: emit(reg.audio_start_id) state = AUDIO elif next_id == reg.audio_start_id: meta["transition"] = "audio_start_direct" emit(reg.think_end_id) if tokens_used < max_new_tokens: emit(next_id) state = AUDIO else: think_ids.append(next_id) think_count += 1 emit(next_id) else: out = model(input_ids=seq) logits = out["logits"][0, -1].float() logits = torch.where(speech_mask, logits, neg_inf) next_id = _sample_from_logits( logits, temperature=temperature, top_p=top_p, top_k=top_k, ) speech_ids.append(next_id) emit(next_id) return think_ids, speech_ids, meta def generate_speech( model, prompt: str, output_path: str, *, stage: int = 1, max_new_tokens: int = 1500, max_think_tokens: int = 200, temperature: float = 1.0, top_p: float = 1.0, top_k: int = 0, constrained: bool = False, speculative: bool = False, codec=None, codec_name: str = "xcodec2", device: Optional[str] = None, ) -> dict: import soundfile as sf from codec import load_codec if device is None: device = getattr(model, "device", "cuda") tokenizer = model.tokenizer registry = model.token_registry input_ids = build_inference_prefix(tokenizer, registry, prompt, stage) prefix_len = input_ids.shape[1] input_ids = input_ids.to(device) cot_meta = None spec_meta = None think_ids: List[int] = [] if speculative and stage == 2: raise NotImplementedError( "speculative=True with stage=2 (CoT-aware decode) is not " "supported in Phase 11 — the THINK→AUDIO state machine and " "the draft-and-verify loop have multiple interaction points " "that need careful design. Use stage=1 + speculative for now." ) if speculative: gen_tokens, spec_meta = speculative_generate( model, input_ids, max_new_tokens=max_new_tokens, temperature=0.0, speech_only=constrained, ) generated_ids = input_ids[0].tolist() + gen_tokens if constrained: speech_ids = list(gen_tokens) else: speech_ids = extract_speech_token_ids( generated_ids, registry, prefix_len, ) elif constrained and stage == 2: think_ids, speech_ids, cot_meta = _cot_constrained_decode( model, input_ids, max_new_tokens=max_new_tokens, max_think_tokens=max_think_tokens, temperature=temperature, top_p=top_p, top_k=top_k, ) boundary_pair = [registry.think_end_id, registry.audio_start_id] if cot_meta and cot_meta.get("transition") == "audio_start_direct": pass generated_ids = ( input_ids[0].tolist() + think_ids + boundary_pair + speech_ids ) elif constrained: speech_ids = _speech_constrained_decode( model, input_ids, max_new_tokens=max_new_tokens, temperature=temperature, top_p=top_p, top_k=top_k, ) generated_ids = input_ids[0].tolist() + speech_ids else: max_length = prefix_len + max_new_tokens with torch.no_grad(): output = model.generate( input_ids, max_length=max_length, temperature=temperature, top_p=top_p, top_k=top_k, ) generated_ids = _flatten_generation_output(output) speech_ids = extract_speech_token_ids(generated_ids, registry, prefix_len) if not speech_ids: raise ValueError( f"No speech tokens generated (prefix_len={prefix_len}, " f"total_generated={len(generated_ids)})" ) if codec is None: codec = load_codec(codec_name, device=device) codec_values = [registry.speech_id_to_value(tid) for tid in speech_ids] waveform, sr = codec.decode(codec_values) sf.write(output_path, waveform, sr) duration = len(waveform) / sr if sr else 0.0 logger.info( "Generated %d speech tokens -> %s (%.2fs)", len(speech_ids), output_path, duration, ) result = { "output_path": output_path, "num_speech_tokens": len(speech_ids), "duration_seconds": duration, "sample_rate": sr, "token_ids": generated_ids, } if cot_meta is not None: result["cot_meta"] = cot_meta result["think_token_count"] = len(think_ids) if spec_meta is not None: result["spec_meta"] = spec_meta return result