Telugu ISO single-speaker fine-tune (run LoRA)

.gitattributes

@@ -33,3 +33,4 @@ saved_model/**/* filter=lfs diff=lfs merge=lfs -text
 *.zip filter=lfs diff=lfs merge=lfs -text
 *.zst filter=lfs diff=lfs merge=lfs -text
 *tfevents* filter=lfs diff=lfs merge=lfs -text
+tokenizer.json filter=lfs diff=lfs merge=lfs -text

README.md

@@ -0,0 +1,33 @@
+---
+license: other
+language: [te, en]
+pipeline_tag: text-to-speech
+tags: [text-to-speech, telugu, higgs-audio, voice-avatar, code-switch]
+base_model: bosonai/higgs-audio-v3-tts-4b
+---
+
+# Higgs Audio v3 — Telugu single-speaker voice (ISO-romanized fine-tune)
+
+Expressive single-speaker **Telugu** (code-switch) TTS, fine-tuned from Higgs
+Audio v3 (4B). Generates this speaker's voice **from text with no reference clip**.
+
+**Frontend matters:** Telugu text must be ISO-15919 romanized with the *same*
+`frontend.py` used in training (Telugu tokenizes to byte-fragments otherwise).
+`generate_speech` handles the rest.
+
+```python
+import torch, torchaudio
+from transformers import AutoModelForCausalLM, AutoTokenizer
+from frontend import romanize          # shipped in this repo
+
+repo = "BNarayanaReddy/higgs-telugu-e4-lora"
+tok = AutoTokenizer.from_pretrained(repo)
+model = AutoModelForCausalLM.from_pretrained(repo, trust_remote_code=True,
+                                             dtype=torch.bfloat16).to("cuda").eval()
+
+text = "హలో, ఈ రోజు ఎలా ఉన్నారు?"
+wav = model.generate_speech(romanize(text, "iso"), tok, temperature=0.7, top_p=0.95)
+torchaudio.save("out.wav", wav.unsqueeze(0), model.config.sample_rate)
+```
+
+Run #1 of an ISO-romanized single-speaker adaptation. Research use.

chat_template.jinja

@@ -0,0 +1,54 @@
+{%- if tools %}
+    {{- '<|im_start|>system\n' }}
+    {%- if messages[0]['role'] == 'system' %}
+        {{- messages[0]['content'] }}
+    {%- else %}
+        {{- 'You are a helpful assistant.' }}
+    {%- endif %}
+    {{- "\n\n# Tools\n\nYou may call one or more functions to assist with the user query.\n\nYou are provided with function signatures within <tools></tools> XML tags:\n<tools>" }}
+    {%- for tool in tools %}
+        {{- "\n" }}
+        {{- tool | tojson }}
+    {%- endfor %}
+    {{- "\n</tools>\n\nFor each function call, return a json object with function name and arguments within <tool_call></tool_call> XML tags:\n<tool_call>\n{\"name\": <function-name>, \"arguments\": <args-json-object>}\n</tool_call><|im_end|>\n" }}
+{%- else %}
+    {%- if messages[0]['role'] == 'system' %}
+        {{- '<|im_start|>system\n' + messages[0]['content'] + '<|im_end|>\n' }}
+    {%- else %}
+        {{- '<|im_start|>system\nYou are a helpful assistant.<|im_end|>\n' }}
+    {%- endif %}
+{%- endif %}
+{%- for message in messages %}
+    {%- if (message.role == "user") or (message.role == "system" and not loop.first) or (message.role == "assistant" and not message.tool_calls) %}
+        {{- '<|im_start|>' + message.role + '\n' + message.content + '<|im_end|>' + '\n' }}
+    {%- elif message.role == "assistant" %}
+        {{- '<|im_start|>' + message.role }}
+        {%- if message.content %}
+            {{- '\n' + message.content }}
+        {%- endif %}
+        {%- for tool_call in message.tool_calls %}
+            {%- if tool_call.function is defined %}
+                {%- set tool_call = tool_call.function %}
+            {%- endif %}
+            {{- '\n<tool_call>\n{"name": "' }}
+            {{- tool_call.name }}
+            {{- '", "arguments": ' }}
+            {{- tool_call.arguments | tojson }}
+            {{- '}\n</tool_call>' }}
+        {%- endfor %}
+        {{- '<|im_end|>\n' }}
+    {%- elif message.role == "tool" %}
+        {%- if (loop.index0 == 0) or (messages[loop.index0 - 1].role != "tool") %}
+            {{- '<|im_start|>user' }}
+        {%- endif %}
+        {{- '\n<tool_response>\n' }}
+        {{- message.content }}
+        {{- '\n</tool_response>' }}
+        {%- if loop.last or (messages[loop.index0 + 1].role != "tool") %}
+            {{- '<|im_end|>\n' }}
+        {%- endif %}
+    {%- endif %}
+{%- endfor %}
+{%- if add_generation_prompt %}
+    {{- '<|im_start|>assistant\n' }}
+{%- endif %}

config.json

@@ -0,0 +1,99 @@
+{
+  "architectures": [
+    "HiggsMultimodalQwen3ForConditionalGeneration"
+  ],
+  "audio_encoder_config": {
+    "encoder_type": "discrete",
+    "mel_per_sample": 8,
+    "model_type": "higgs_audio_encoder",
+    "num_codebooks": 8,
+    "out_dim": 2560,
+    "tie_word_embeddings": true,
+    "use_delay_pattern": true,
+    "vocab_size": 1026
+  },
+  "audio_token_id": -100,
+  "audio_tokenizer_id": "bosonai/higgs-audio-v2-tokenizer",
+  "auto_map": {
+    "AutoConfig": "configuration_higgs_multimodal_qwen3.HiggsMultimodalQwen3Config",
+    "AutoModel": "modeling_higgs_multimodal_qwen3.HiggsMultimodalQwen3ForConditionalGeneration",
+    "AutoModelForCausalLM": "modeling_higgs_multimodal_qwen3.HiggsMultimodalQwen3ForConditionalGeneration"
+  },
+  "dtype": "bfloat16",
+  "ignore_index": -100,
+  "mel_per_sample": 8,
+  "model_type": "higgs_multimodal_qwen3",
+  "sample_rate": 24000,
+  "text_config": {
+    "architectures": [
+      "Qwen3ForCausalLM"
+    ],
+    "attention_bias": false,
+    "attention_dropout": 0.0,
+    "bos_token_id": 151643,
+    "dtype": "bfloat16",
+    "eos_token_id": 151643,
+    "head_dim": 128,
+    "hidden_act": "silu",
+    "hidden_size": 2560,
+    "initializer_range": 0.02,
+    "intermediate_size": 9728,
+    "layer_types": [
+      "full_attention",
+      "full_attention",
+      "full_attention",
+      "full_attention",
+      "full_attention",
+      "full_attention",
+      "full_attention",
+      "full_attention",
+      "full_attention",
+      "full_attention",
+      "full_attention",
+      "full_attention",
+      "full_attention",
+      "full_attention",
+      "full_attention",
+      "full_attention",
+      "full_attention",
+      "full_attention",
+      "full_attention",
+      "full_attention",
+      "full_attention",
+      "full_attention",
+      "full_attention",
+      "full_attention",
+      "full_attention",
+      "full_attention",
+      "full_attention",
+      "full_attention",
+      "full_attention",
+      "full_attention",
+      "full_attention",
+      "full_attention",
+      "full_attention",
+      "full_attention",
+      "full_attention",
+      "full_attention"
+    ],
+    "max_position_embeddings": 32768,
+    "max_window_layers": 36,
+    "model_type": "qwen3",
+    "num_attention_heads": 32,
+    "num_hidden_layers": 36,
+    "num_key_value_heads": 8,
+    "pad_token_id": null,
+    "rms_norm_eps": 1e-06,
+    "rope_parameters": {
+      "rope_theta": 1000000,
+      "rope_type": "default"
+    },
+    "sliding_window": null,
+    "tie_word_embeddings": true,
+    "use_cache": true,
+    "use_sliding_window": false,
+    "vocab_size": 151936
+  },
+  "transformers_version": "5.16.1",
+  "use_cache": false
+}

configuration_higgs_multimodal_qwen3.py

@@ -0,0 +1,90 @@
+# coding=utf-8
+"""Config for HiggsMultimodalQwen3 — the Higgs Audio v3 TTS model.
+
+A standard Qwen3 text backbone plus a fused multi-codebook audio embedding /
+head. Audio is encoded/decoded by the separately-loaded
+``bosonai/higgs-audio-v2-tokenizer`` (``higgs_audio_v2_tokenizer``), which is
+native to transformers >= 5.5.
+"""
+
+from __future__ import annotations
+
+from typing import Any
+
+from transformers import CONFIG_MAPPING, PretrainedConfig
+
+# Higgs Qwen3 sub-configs ship ``rope_theta=null``; transformers' default of
+# 10000 is wrong for Qwen3 (trained at 1e6). Patch before instantiation.
+_QWEN3_ROPE_THETA = 1_000_000
+
+
+def _build_text_config(raw: Any) -> PretrainedConfig:
+    """Realise a text-backbone sub-config into a concrete ``PretrainedConfig``."""
+    if isinstance(raw, PretrainedConfig):
+        return raw
+    cfg = dict(raw or {})
+    model_type = cfg.get("model_type", "qwen3")
+    if model_type == "qwen3":
+        rope = cfg.get("rope_parameters") or {}
+        if cfg.get("rope_theta") is None and rope.get("rope_theta") is None:
+            cfg["rope_theta"] = _QWEN3_ROPE_THETA
+    cfg_cls = CONFIG_MAPPING[model_type]
+    return cfg_cls(**cfg)
+
+
+_DEFAULT_AUDIO_ENCODER_CONFIG: dict[str, Any] = {
+    "encoder_type": "discrete",
+    "num_codebooks": 8,
+    "vocab_size": 1026,
+    "out_dim": 2560,
+    "tie_word_embeddings": True,
+    "use_delay_pattern": True,
+    "model_type": "higgs_audio_encoder",
+}
+
+
+class HiggsMultimodalQwen3Config(PretrainedConfig):
+    """Config for ``HiggsMultimodalQwen3ForConditionalGeneration``.
+
+    Args:
+        audio_encoder_config: discrete-codec descriptor. ``num_codebooks`` /
+            ``vocab_size`` (incl. BOC/EOC specials) / ``out_dim`` /
+            ``tie_word_embeddings`` drive the fused embedding + head.
+        text_config: Qwen3 backbone config, eagerly realised so
+            ``config.text_config.num_attention_heads`` works directly.
+        audio_token_id: placeholder id (``-100``) marking reference-audio slots
+            in ``input_ids`` that the fused audio embedding fills.
+        audio_tokenizer_id: repo id of the codec used to encode reference audio
+            and decode generated codes back to a waveform.
+        sample_rate: codec sample rate, Hz.
+    """
+
+    model_type = "higgs_multimodal_qwen3"
+    is_composition = True
+
+    def __init__(
+        self,
+        audio_encoder_config: dict[str, Any] | None = None,
+        text_config: dict[str, Any] | PretrainedConfig | None = None,
+        audio_token_id: int = -100,
+        mel_per_sample: int = 8,
+        audio_tokenizer_id: str = "bosonai/higgs-audio-v2-tokenizer",
+        sample_rate: int = 24_000,
+        **kwargs,
+    ):
+        self.audio_token_id = audio_token_id
+        self.mel_per_sample = mel_per_sample
+        self.audio_tokenizer_id = audio_tokenizer_id
+        self.sample_rate = sample_rate
+        self.audio_encoder_config = audio_encoder_config or dict(
+            _DEFAULT_AUDIO_ENCODER_CONFIG
+        )
+        self.text_config = _build_text_config(text_config)
+        super().__init__(**kwargs)
+
+    def get_text_config(self, decoder: bool = False) -> PretrainedConfig:
+        del decoder
+        return self.text_config
+
+
+__all__ = ["HiggsMultimodalQwen3Config"]

frontend.py

@@ -0,0 +1,42 @@
+"""Frozen Telugu text frontend — the ONE romanizer used at train AND serve.
+
+Never reimplement this on one side. A one-token drift between training and
+inference silently destroys quality. Run #1 uses scheme="iso".
+
+  iso    ISO-15919, diacritics kept (halō, mīru viṁṭunnāru) — phonemically faithful.
+  ascii  ISO-15919 diacritics stripped — ablation only (bakes in English accent).
+  native passthrough (byte-soup baseline) — ablation only.
+
+Only Telugu Unicode runs are transliterated; English/code-switch spans, digits
+and punctuation pass through untouched.
+"""
+from __future__ import annotations
+
+import re
+import unicodedata
+
+from indic_transliteration import sanscript
+
+_TELUGU_RUN = re.compile(r"[ఀ-౿]+")
+
+
+def _iso_run(m: "re.Match") -> str:
+    return sanscript.transliterate(m.group(0), sanscript.TELUGU, sanscript.ISO)
+
+
+def romanize(text: str, scheme: str = "iso") -> str:
+    if scheme == "native":
+        return text
+    iso = _TELUGU_RUN.sub(_iso_run, text)
+    if scheme == "iso":
+        return iso
+    if scheme == "ascii":
+        d = unicodedata.normalize("NFKD", iso)
+        return "".join(c for c in d if not unicodedata.combining(c))
+    raise ValueError(f"unknown scheme: {scheme!r}")
+
+
+if __name__ == "__main__":
+    for s in ["హలో my dear Telugu people, మీరు వింటున్నారు.", "పవనితో కబుర్లు."]:
+        print("NATIVE:", s)
+        print("  ISO :", romanize(s, "iso"))

model.safetensors

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:2b6fbb37661844298934269a3b27e35c715fd75c8aee54c09cf84bc8d0692a70
+size 8087006776

modeling_higgs_multimodal_qwen3.py

@@ -0,0 +1,391 @@
+# coding=utf-8
+"""HiggsMultimodalQwen3 — Higgs Audio v3 TTS, ported to plain transformers.
+
+Architecture:
+  * a standard Qwen3 backbone (``Qwen3Model``) for the autoregressive LM;
+  * one fused multi-codebook embedding ``[N*V, D]`` whose per-codebook lookups
+    are summed — it both embeds reference-audio codes into the prompt and
+    re-embeds each generated row during decoding;
+  * a tied fused head ``[L, D] -> [L, N, V]`` producing per-codebook logits.
+
+Audio I/O (waveform <-> discrete codes) is handled by the transformers-native
+``HiggsAudioV2TokenizerModel`` (``bosonai/higgs-audio-v2-tokenizer``), loaded
+lazily on first use.
+
+Generation follows Higgs' delay pattern: codebook ``c`` is shifted by ``c``
+steps, padded with BOC (1024) before and EOC (1025) after its data span. A
+small per-request state machine drives the delay ramp-up, EOC detection and
+wind-down; the produced rows are de-delayed and decoded to a 24 kHz waveform.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import Any
+
+import torch
+import torch.nn.functional as F
+from torch import nn
+from transformers import AutoModel, PreTrainedModel
+from transformers.models.qwen3.modeling_qwen3 import Qwen3Model
+
+from .configuration_higgs_multimodal_qwen3 import HiggsMultimodalQwen3Config
+
+# Codec-vocab specials, inside the per-codebook [0, V) space (NOT the text vocab).
+BOC_ID = 1024
+EOC_ID = 1025
+# Placeholder id marking reference-audio slots in ``input_ids``.
+AUDIO_PLACEHOLDER_ID = -100
+
+_REQUIRED_SPECIALS = ("<|tts|>", "<|ref_audio|>", "<|text|>", "<|audio|>")
+
+
+# --------------------------------------------------------------------------- #
+# Delay pattern
+# --------------------------------------------------------------------------- #
+def apply_delay_pattern(codes_TN: torch.Tensor) -> torch.Tensor:
+    """``[T, N]`` raw codes -> ``[T + N - 1, N]`` delayed, BOC/EOC padded."""
+    T, N = codes_TN.shape
+    out = torch.full(
+        (T + N - 1, N), EOC_ID, device=codes_TN.device, dtype=codes_TN.dtype
+    )
+    t_idx = torch.arange(T + N - 1, device=codes_TN.device)
+    for c in range(N):
+        out[t_idx < c, c] = BOC_ID
+        out[c : c + T, c] = codes_TN[:, c]
+    return out
+
+
+def reverse_delay_pattern(delayed_LN: torch.Tensor) -> torch.Tensor:
+    """``[L, N]`` delayed (L >= N) -> ``[L - (N - 1), N]`` raw codes."""
+    L, N = delayed_LN.shape
+    T = L - (N - 1)
+    if T <= 0:
+        raise ValueError(f"delayed rows L={L} < num_codebooks N={N}")
+    out = torch.empty((T, N), device=delayed_LN.device, dtype=delayed_LN.dtype)
+    for c in range(N):
+        out[:, c] = delayed_LN[c : c + T, c]
+    return out
+
+
+# --------------------------------------------------------------------------- #
+# Fused multi-codebook modules
+# --------------------------------------------------------------------------- #
+class HiggsFusedMultiTextEmbedding(nn.Module):
+    """Fused multi-codebook embedding: one ``[N*V, D]`` weight + offset lookup.
+
+    ``codes_LN[..., N]`` -> ``[..., D]`` summed across the codebook axis.
+    """
+
+    def __init__(self, num_codebooks: int, vocab_size: int, hidden_size: int):
+        super().__init__()
+        self.weight = nn.Parameter(torch.empty(num_codebooks * vocab_size, hidden_size))
+        self.num_codebooks = num_codebooks
+        self.vocab_size = vocab_size
+
+    def forward(self, codes_LN: torch.Tensor) -> torch.Tensor:
+        offsets = (
+            torch.arange(self.num_codebooks, device=codes_LN.device, dtype=codes_LN.dtype)
+            * self.vocab_size
+        )
+        return F.embedding(codes_LN + offsets, self.weight).sum(dim=-2)
+
+
+class HiggsFusedMultiTextHead(nn.Module):
+    """Fused multi-codebook head: ``[L, D]`` -> ``[L, N, V]`` via one linear."""
+
+    def __init__(self, num_codebooks: int, vocab_size: int, hidden_size: int):
+        super().__init__()
+        self.weight = nn.Parameter(torch.empty(num_codebooks * vocab_size, hidden_size))
+        self.num_codebooks = num_codebooks
+        self.vocab_size = vocab_size
+
+    def forward(self, hidden_LD: torch.Tensor) -> torch.Tensor:
+        logits = F.linear(hidden_LD, self.weight)
+        return logits.reshape(hidden_LD.shape[0], self.num_codebooks, self.vocab_size)
+
+
+# --------------------------------------------------------------------------- #
+# Per-request delay/EOC sampler state machine (reference oracle, pure torch)
+# --------------------------------------------------------------------------- #
+@dataclass
+class _SamplerState:
+    num_codebooks: int
+    delay_count: int = 0
+    eoc_countdown: int | None = None
+    generation_done: bool = False
+    last_codes: torch.Tensor | None = None
+
+
+def _sample(logits_NV: torch.Tensor, temperature: float, top_p: float | None,
+            top_k: int | None) -> torch.Tensor:
+    if temperature <= 1e-5:
+        return logits_NV.argmax(dim=-1)
+    logits = logits_NV / temperature
+    if top_k is not None and top_k > 0:
+        k = min(top_k, logits.size(-1))
+        kth = logits.topk(k, dim=-1).values[:, -1:]
+        logits = torch.where(logits < kth, float("-inf"), logits)
+    if top_p is not None and top_p < 1.0:
+        sorted_logits, sorted_idx = torch.sort(logits, descending=True, dim=-1)
+        cum = sorted_logits.softmax(dim=-1).cumsum(dim=-1)
+        remove = cum > top_p
+        remove[..., 1:] = remove[..., :-1].clone()
+        remove[..., 0] = False
+        scatter = torch.zeros_like(remove)
+        scatter.scatter_(-1, sorted_idx, remove)
+        logits = torch.where(scatter, float("-inf"), logits)
+    return logits.softmax(dim=-1).multinomial(num_samples=1).squeeze(-1)
+
+
+def _sampler_step(logits_NV: torch.Tensor, state: _SamplerState, *,
+                  temperature: float, top_p: float | None,
+                  top_k: int | None) -> torch.Tensor:
+    """One AR step of the multi-codebook delay sampler. Mutates ``state``."""
+    N = state.num_codebooks
+    codes_N = _sample(logits_NV, temperature, top_p, top_k).to(torch.long)
+
+    if state.delay_count < N:
+        next_cb = state.delay_count + 1
+        if next_cb < N:
+            codes_N[next_cb:] = BOC_ID
+        state.delay_count += 1
+    elif state.eoc_countdown is not None:
+        state.eoc_countdown -= 1
+        if state.eoc_countdown <= 0:
+            state.generation_done = True
+    elif int(codes_N[0].item()) == EOC_ID:
+        if N <= 2:
+            state.generation_done = True
+        else:
+            state.eoc_countdown = N - 2
+
+    if not state.generation_done:
+        state.last_codes = codes_N.clone()
+    return codes_N
+
+
+class HiggsMultimodalQwen3PreTrainedModel(PreTrainedModel):
+    config_class = HiggsMultimodalQwen3Config
+    base_model_prefix = "model"
+    _supports_cache_class = True
+    _supports_sdpa = True
+
+
+class HiggsMultimodalQwen3ForConditionalGeneration(HiggsMultimodalQwen3PreTrainedModel):
+    """Higgs Audio v3 TTS model.
+
+    Loadable via ``AutoModelForCausalLM.from_pretrained(..., trust_remote_code=True)``.
+    Use :meth:`generate_speech` for end-to-end text -> waveform synthesis.
+    """
+
+    # Higgs ckpt names -> this module's parameter tree. transformers 5.x applies
+    # this via the ``key_mapping`` weight-conversion path (see ``from_pretrained``).
+    _HIGGS_KEY_MAPPING = {
+        r"tied\.embedding\.text_embedding\.": "model.embed_tokens.",
+        r"tied\.embedding\.modality_embeddings\.0\.embedding\.": "audio_embedding.",
+        r"body\.": "model.",
+    }
+    # Codec weights (bundled in the ckpt) and the tied text head are not part of
+    # the AR graph; the codec is loaded separately from ``audio_tokenizer_id``.
+    _keys_to_ignore_on_load_unexpected = [
+        r"tied\.embedding\.modality_embeddings\.0\.model\.",
+        r"tied\.head\.",
+    ]
+    _tied_weights_keys = ["audio_head.weight"]
+
+    @classmethod
+    def from_pretrained(cls, *args, **kwargs):
+        kwargs.setdefault("key_mapping", dict(cls._HIGGS_KEY_MAPPING))
+        return super().from_pretrained(*args, **kwargs)
+
+    def __init__(self, config: HiggsMultimodalQwen3Config):
+        super().__init__(config)
+        text_config = config.get_text_config()
+        self.model = Qwen3Model(text_config)
+
+        enc = config.audio_encoder_config or {}
+        self.num_codebooks = int(enc["num_codebooks"])
+        self.codebook_vocab_size = int(enc["vocab_size"])
+        hidden = int(enc.get("out_dim", text_config.hidden_size))
+        self._tie_audio_head = bool(enc.get("tie_word_embeddings", True))
+
+        self.audio_embedding = HiggsFusedMultiTextEmbedding(
+            self.num_codebooks, self.codebook_vocab_size, hidden
+        )
+        self.audio_head = HiggsFusedMultiTextHead(
+            self.num_codebooks, self.codebook_vocab_size, hidden
+        )
+        self._audio_codec = None  # lazily loaded
+        self.post_init()
+
+    def tie_weights(self, *args, **kwargs):
+        super().tie_weights(*args, **kwargs)
+        if self._tie_audio_head:
+            self.audio_head.weight = self.audio_embedding.weight
+
+    def get_input_embeddings(self):
+        return self.model.embed_tokens
+
+    def set_input_embeddings(self, value):
+        self.model.embed_tokens = value
+
+    # ----- audio codec (lazy) --------------------------------------------- #
+    def get_audio_codec(self):
+        """Load + cache the ``higgs_audio_v2_tokenizer`` codec (fp32, eval)."""
+        if self._audio_codec is None:
+            codec = AutoModel.from_pretrained(
+                self.config.audio_tokenizer_id, dtype=torch.float32,
+                trust_remote_code=True,
+            )
+            codec = codec.to(self.device).eval()
+            for p in codec.parameters():
+                p.requires_grad_(False)
+            self._audio_codec = codec
+        return self._audio_codec
+
+    @torch.no_grad()
+    def _encode_reference(self, waveform: torch.Tensor, sample_rate: int) -> torch.Tensor:
+        """Reference waveform -> ``[T, N]`` int64 codes (on model device)."""
+        import torchaudio
+
+        codec = self.get_audio_codec()
+        wav = waveform.float()
+        while wav.ndim < 3:
+            wav = wav.unsqueeze(0)
+        if sample_rate != self.config.sample_rate:
+            wav = torchaudio.functional.resample(wav, sample_rate, self.config.sample_rate)
+        if wav.shape[-1] < self.config.sample_rate:
+            wav = F.pad(wav, (0, self.config.sample_rate - wav.shape[-1]))
+        wav = wav.to(self.device, dtype=torch.float32)
+        codes_BNT = codec.encode(wav).audio_codes
+        return codes_BNT.squeeze(0).transpose(0, 1).to(torch.long)
+
+    @torch.no_grad()
+    def _decode_codes(self, codes_TN: torch.Tensor) -> torch.Tensor:
+        """``[T, N]`` raw codes -> mono waveform ``[L]`` (CPU float32)."""
+        codec = self.get_audio_codec()
+        codec_vocab = self.codebook_vocab_size - 2  # drop BOC/EOC
+        codes_TN = torch.where(codes_TN >= codec_vocab, torch.zeros_like(codes_TN), codes_TN)
+        codes_BNT = codes_TN.transpose(0, 1).unsqueeze(0).to(self.device, torch.long)
+        audio = codec.decode(codes_BNT).audio_values
+        return audio.squeeze(0).squeeze(0).detach().cpu().float()
+
+    # ----- prompt assembly ------------------------------------------------ #
+    @staticmethod
+    def _special_ids(tokenizer) -> dict[str, int | None]:
+        vocab = dict(tokenizer.get_added_vocab())
+        missing = [t for t in _REQUIRED_SPECIALS if t not in vocab]
+        if missing:
+            raise ValueError(f"Tokenizer is missing Higgs TTS specials: {missing}")
+        ids = {t: vocab[t] for t in _REQUIRED_SPECIALS}
+        ids["<|ref_text|>"] = vocab.get("<|ref_text|>")
+        return ids
+
+    def _build_prompt_ids(self, tokenizer, text: str, *, num_ref_tokens: int,
+                          reference_text: str | None) -> list[int]:
+        sp = self._special_ids(tokenizer)
+        ids: list[int] = [sp["<|tts|>"]]
+        if reference_text and num_ref_tokens > 0 and sp["<|ref_text|>"] is not None:
+            ids.append(sp["<|ref_text|>"])
+            ids.extend(tokenizer.encode(reference_text, add_special_tokens=False))
+        if num_ref_tokens > 0:
+            ids.append(sp["<|ref_audio|>"])
+            ids.extend([AUDIO_PLACEHOLDER_ID] * num_ref_tokens)
+        ids.append(sp["<|text|>"])
+        ids.extend(tokenizer.encode(text, add_special_tokens=False))
+        ids.append(sp["<|audio|>"])
+        return ids
+
+    def _prefill_embeds(self, prompt_ids: list[int],
+                        delayed_ref: torch.Tensor | None) -> torch.Tensor:
+        """Embed the prompt; overlay fused audio embedding at ``-100`` slots."""
+        ids = torch.tensor(prompt_ids, dtype=torch.long, device=self.device)
+        mask = ids == AUDIO_PLACEHOLDER_ID
+        safe = torch.where(mask, torch.zeros_like(ids), ids)
+        embeds = self.model.embed_tokens(safe)
+        if delayed_ref is not None and mask.any():
+            n = int(mask.sum().item())
+            audio = self.audio_embedding(delayed_ref[:n].to(self.device))
+            embeds[mask] = audio.to(embeds.dtype)
+        return embeds.unsqueeze(0)  # [1, S, D]
+
+    # ----- generation ----------------------------------------------------- #
+    @torch.no_grad()
+    def generate_speech(
+        self,
+        text: str,
+        tokenizer,
+        *,
+        reference_audio: torch.Tensor | None = None,
+        reference_sample_rate: int | None = None,
+        reference_codes: torch.Tensor | None = None,
+        reference_text: str | None = None,
+        max_new_tokens: int = 2048,
+        temperature: float = 1.0,
+        top_p: float | None = None,
+        top_k: int | None = None,
+    ) -> torch.Tensor:
+        """Synthesize ``text`` into a mono 24 kHz waveform (CPU float32 ``[L]``).
+
+        Voice cloning: pass ``reference_audio`` (``[L]`` / ``[C, L]`` tensor with
+        ``reference_sample_rate``) or pre-encoded ``reference_codes`` (``[T, N]``).
+        ``reference_text`` (the transcript of the reference) improves cloning.
+        """
+        N = self.num_codebooks
+
+        delayed_ref = None
+        if reference_codes is not None:
+            delayed_ref = apply_delay_pattern(reference_codes.to(torch.long))
+        elif reference_audio is not None:
+            sr = reference_sample_rate or self.config.sample_rate
+            codes_TN = self._encode_reference(reference_audio, sr)
+            delayed_ref = apply_delay_pattern(codes_TN.cpu())
+
+        prompt_ids = self._build_prompt_ids(
+            tokenizer, text,
+            num_ref_tokens=0 if delayed_ref is None else delayed_ref.shape[0],
+            reference_text=reference_text,
+        )
+
+        inputs_embeds = self._prefill_embeds(prompt_ids, delayed_ref)
+        out = self.model(inputs_embeds=inputs_embeds, use_cache=True)
+        past = out.past_key_values
+        hidden_last = out.last_hidden_state[:, -1, :]
+        position = inputs_embeds.shape[1]
+
+        state = _SamplerState(num_codebooks=N)
+        rows: list[torch.Tensor] = []
+        for _ in range(max_new_tokens):
+            logits_NV = self.audio_head(hidden_last).to(torch.float32)[0]  # [N, V]
+            codes_N = _sampler_step(
+                logits_NV, state,
+                temperature=temperature, top_p=top_p, top_k=top_k,
+            )
+            if state.generation_done:
+                break
+            rows.append(codes_N.cpu())
+
+            step_embed = self.audio_embedding(codes_N.unsqueeze(0)).unsqueeze(1)
+            cache_pos = torch.tensor([position], device=self.device)
+            out = self.model(
+                inputs_embeds=step_embed.to(inputs_embeds.dtype),
+                past_key_values=past,
+                use_cache=True,
+                cache_position=cache_pos,
+            )
+            past = out.past_key_values
+            hidden_last = out.last_hidden_state[:, -1, :]
+            position += 1
+
+        if len(rows) < N:
+            return torch.zeros(0, dtype=torch.float32)
+        delayed_LN = torch.stack(rows, dim=0)
+        codes_TN = reverse_delay_pattern(delayed_LN)
+        return self._decode_codes(codes_TN)
+
+
+__all__ = [
+    "HiggsMultimodalQwen3ForConditionalGeneration",
+    "HiggsMultimodalQwen3PreTrainedModel",
+]

speaker_embedding.pt

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:e54562976d9f6334e391072d6289e2fcf685e8bfbe0f162091884f5bfc1ba2bb
+size 11951

tags_spec.json

@@ -0,0 +1 @@
+{"speakers": ["pavani"], "serve_tags": ["<|prosody:expressive_high|>"]}

tokenizer.json

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:eb883de2de5adc5113f1f02b54830a0ea7cd6ef191cde65c41aceb3737d4d1c1
+size 11433924

tokenizer_config.json

@@ -0,0 +1,79 @@
+{
+  "add_prefix_space": false,
+  "backend": "tokenizers",
+  "bos_token": null,
+  "clean_up_tokenization_spaces": false,
+  "eos_token": "<|endoftext|>",
+  "errors": "replace",
+  "extra_special_tokens": [
+    "<|asr|>",
+    "<|streaming_asr|>",
+    "<|tts|>",
+    "<|streaming_tts|>",
+    "<|audio_cont_txt|>",
+    "<|audio|>",
+    "<|audio_end|>",
+    "<|text|>",
+    "<|text_end|>",
+    "<|eoc|>",
+    "<|user|>",
+    "<|assistant|>",
+    "<|system|>",
+    "<|await_audio|>",
+    "<|ref_audio|>",
+    "<|ref_text|>",
+    "<|emotion:elation|>",
+    "<|emotion:amusement|>",
+    "<|emotion:enthusiasm|>",
+    "<|emotion:determination|>",
+    "<|emotion:pride|>",
+    "<|emotion:contentment|>",
+    "<|emotion:affection|>",
+    "<|emotion:relief|>",
+    "<|emotion:contemplation|>",
+    "<|emotion:confusion|>",
+    "<|emotion:surprise|>",
+    "<|emotion:awe|>",
+    "<|emotion:longing|>",
+    "<|emotion:arousal|>",
+    "<|emotion:anger|>",
+    "<|emotion:fear|>",
+    "<|emotion:disgust|>",
+    "<|emotion:bitterness|>",
+    "<|emotion:sadness|>",
+    "<|emotion:shame|>",
+    "<|emotion:helplessness|>",
+    "<|env:music|>",
+    "<|env:noise|>",
+    "<|style:singing|>",
+    "<|style:shouting|>",
+    "<|style:whispering|>",
+    "<|sfx:cough|>",
+    "<|sfx:laughter|>",
+    "<|sfx:crying|>",
+    "<|sfx:screaming|>",
+    "<|sfx:burping|>",
+    "<|sfx:humming|>",
+    "<|sfx:sigh|>",
+    "<|sfx:sniff|>",
+    "<|sfx:sneeze|>",
+    "<|prosody:speed_very_slow|>",
+    "<|prosody:speed_slow|>",
+    "<|prosody:speed_fast|>",
+    "<|prosody:speed_very_fast|>",
+    "<|prosody:pitch_low|>",
+    "<|prosody:pitch_high|>",
+    "<|prosody:pause|>",
+    "<|prosody:long_pause|>",
+    "<|chatml|>",
+    "<|prosody:expressive_high|>",
+    "<|prosody:expressive_low|>"
+  ],
+  "is_local": false,
+  "local_files_only": false,
+  "model_max_length": 131072,
+  "pad_token": "<|endoftext|>",
+  "split_special_tokens": false,
+  "tokenizer_class": "Qwen2Tokenizer",
+  "unk_token": null
+}