# dataset.py: Text + think block + speech tokens -> one training sequence with per-region loss weights; collator. import json import logging import torch from torch.utils.data import Dataset from dataclasses import dataclass from typing import List, Dict, Any from tokenizer import TokenRegistry, NUM_SPEECH_TOKENS logger = logging.getLogger(__name__) IGNORE_INDEX = -100 class MambaCoTTTSDataset(Dataset): def __init__( self, data_path: str, tokenizer, registry: TokenRegistry, stage: int = 1, max_seq_len: int = 2048, cfg_dropout: float = 0.0, cot_mode: str = "prose", ): assert stage in (1, 2, "auto"), f"stage must be 1, 2 or 'auto', got {stage}" self.tokenizer = tokenizer self.registry = registry self.stage = stage self.max_seq_len = max_seq_len self.cfg_dropout = cfg_dropout self.data: List[dict] = [] with open(data_path, "r") as f: for line in f: line = line.strip() if line: self.data.append(json.loads(line)) assert cot_mode in ("prose", "tags", "none"), ( f"cot_mode={cot_mode!r} is not supported by the flat " f"MambaCoTTTSDataset. The P²-CoT plan tiers (plan / tags+plan / " f"full) carry a per-position injection schedule and run on the " f"delay/depth path only — launch with --depth_module (or " f"--delay_pattern), which routes to delay_dataset.DelayMimiDataset." if cot_mode in ("plan", "tags+plan", "full") else cot_mode) self.cot_mode = cot_mode if cot_mode != "prose": for row in self.data: if cot_mode == "none": row["reasoning"] = "" else: row["reasoning"] = ( f"EMO={row.get('emotion_label') or 'neutral'} " f"PACE={row.get('pace') or 'normal'} " f"PITCH={row.get('pitch') or 'normal'}") self.bos_id = registry.bos_id self.eos_id = registry.eos_id self.user_prompt_id = registry.user_prompt_id self.think_start_id = registry.think_start_id self.think_end_id = registry.think_end_id self.audio_start_id = registry.audio_start_id self.audio_end_id = registry.audio_end_id logger.info( f"Loaded {len(self.data)} samples from {data_path} (stage={stage}, " f"cot_mode={cot_mode}, max_seq_len={max_seq_len})" ) def __len__(self) -> int: return len(self.data) def __getitem__(self, idx: int) -> dict: item = self.data[idx] if self.stage == "auto": row_stage = 2 if str(item.get("reasoning", "") or "").strip() else 1 else: row_stage = self.stage prompt_text: str = item["prompt"] speech_values: List[int] = item["speech_tokens"] residual_codes = item.get("residual_codes", None) speaker_id = item.get("speaker_id", None) clamped = False for i, v in enumerate(speech_values): if v < 0 or v >= NUM_SPEECH_TOKENS: speech_values[i] = max(0, min(v, NUM_SPEECH_TOKENS - 1)) clamped = True if clamped: logger.warning(f"Sample {idx}: speech_tokens contained out-of-range values, clamped to [0, {NUM_SPEECH_TOKENS - 1}]") speech_ids = [self.registry.speech_value_to_id(v) for v in speech_values] if self.cfg_dropout > 0.0 and torch.rand(1).item() < self.cfg_dropout: prompt_ids = [] else: prompt_ids = self.tokenizer.encode(prompt_text, add_special_tokens=False) reasoning_ids = [] if row_stage == 2: reasoning_text = item.get("reasoning", "") if reasoning_text: reasoning_clean = reasoning_text.strip() if reasoning_clean.startswith(""): reasoning_clean = reasoning_clean[len(""):].strip() if reasoning_clean.endswith(""): reasoning_clean = reasoning_clean[:-len("")].strip() if reasoning_clean: reasoning_ids = self.tokenizer.encode(reasoning_clean, add_special_tokens=False) if row_stage == 1: sequence = ( [self.bos_id, self.user_prompt_id] + prompt_ids + [self.audio_start_id] + speech_ids + [self.audio_end_id, self.eos_id] ) prompt_end = 2 + len(prompt_ids) reasoning_start = prompt_end reasoning_end = prompt_end speech_start = prompt_end + 1 speech_end = speech_start + len(speech_ids) else: sequence = ( [self.bos_id, self.user_prompt_id] + prompt_ids + [self.think_start_id] + reasoning_ids + [self.think_end_id] + [self.audio_start_id] + speech_ids + [self.audio_end_id, self.eos_id] ) prompt_end = 2 + len(prompt_ids) reasoning_start = prompt_end + 1 reasoning_end = reasoning_start + len(reasoning_ids) speech_start = reasoning_end + 2 speech_end = speech_start + len(speech_ids) if len(sequence) > self.max_seq_len: overflow = len(sequence) - self.max_seq_len max_speech_trim = len(speech_ids) speech_trim = min(overflow, max_speech_trim) speech_ids_trimmed = speech_ids[:len(speech_ids) - speech_trim] overflow -= speech_trim reasoning_ids_trimmed = reasoning_ids if overflow > 0 and row_stage == 2: max_reasoning_trim = len(reasoning_ids) reasoning_trim = min(overflow, max_reasoning_trim) reasoning_ids_trimmed = reasoning_ids[:len(reasoning_ids) - reasoning_trim] overflow -= reasoning_trim if overflow > 0: logger.warning( f"Sample {idx}: sequence still {overflow} tokens over max_seq_len " f"after truncating speech and reasoning" ) if row_stage == 1: sequence = ( [self.bos_id, self.user_prompt_id] + prompt_ids + [self.audio_start_id] + speech_ids_trimmed + [self.audio_end_id, self.eos_id] ) prompt_end = 2 + len(prompt_ids) reasoning_start = prompt_end reasoning_end = prompt_end speech_start = prompt_end + 1 speech_end = speech_start + len(speech_ids_trimmed) else: sequence = ( [self.bos_id, self.user_prompt_id] + prompt_ids + [self.think_start_id] + reasoning_ids_trimmed + [self.think_end_id] + [self.audio_start_id] + speech_ids_trimmed + [self.audio_end_id, self.eos_id] ) prompt_end = 2 + len(prompt_ids) reasoning_start = prompt_end + 1 reasoning_end = reasoning_start + len(reasoning_ids_trimmed) speech_start = reasoning_end + 2 speech_end = speech_start + len(speech_ids_trimmed) return { "input_ids": torch.tensor(sequence, dtype=torch.long), "prompt_end": prompt_end, "reasoning_start": reasoning_start, "reasoning_end": reasoning_end, "speech_start": speech_start, "speech_end": speech_end, "residual_codes": residual_codes, "speaker_id": speaker_id, } @dataclass class MambaCoTDataCollator: pad_token_id: int registry: TokenRegistry stage: int = 1 lambda_r: float = 0.3 lambda_s: float = 1.0 semantic_weight: float = 1.0 eos_weight: float = 1.0 level_cycle: int = 0 def __call__(self, batch: List[Dict[str, Any]]) -> Dict[str, torch.Tensor]: input_ids_list = [item["input_ids"] for item in batch] input_ids = torch.nn.utils.rnn.pad_sequence( input_ids_list, batch_first=True, padding_value=self.pad_token_id ) B, T_max = input_ids.shape labels = input_ids.clone() loss_weights = torch.ones(B, T_max, dtype=torch.float32) for i, item in enumerate(batch): seq_len = len(item["input_ids"]) prompt_end = item["prompt_end"] labels[i, :prompt_end] = IGNORE_INDEX if seq_len < T_max: labels[i, seq_len:] = IGNORE_INDEX loss_weights[i, :prompt_end] = 0.0 if seq_len < T_max: loss_weights[i, seq_len:] = 0.0 if self.stage == 2 or self.stage == "auto": r_start = item["reasoning_start"] r_end = item["reasoning_end"] s_start = item["speech_start"] s_end = item["speech_end"] think_delim_start = max(prompt_end, r_start - 1) think_delim_end = min(seq_len, r_end + 1) loss_weights[i, think_delim_start:think_delim_end] = self.lambda_r loss_weights[i, s_start - 1:seq_len] = self.lambda_s if self.semantic_weight != 1.0 and self.level_cycle > 0: s_start = item["speech_start"] s_end = min(item["speech_end"], seq_len) for pos in range(s_start, s_end, self.level_cycle): loss_weights[i, pos] *= self.semantic_weight if self.eos_weight != 1.0: s_end = item["speech_end"] for pos in (s_end, s_end + 1): if pos < seq_len: loss_weights[i, pos] *= self.eos_weight speech_mask = self.registry.build_speech_mask(input_ids) & (labels != IGNORE_INDEX) attention_mask = (input_ids != self.pad_token_id).long() has_residual = any(item.get("residual_codes") is not None for item in batch) residual_codes_tensor = None if has_residual: num_layers = 0 for item in batch: if item.get("residual_codes") is not None: num_layers = len(item["residual_codes"]) break if num_layers > 0: residual_codes_tensor = torch.full((B, num_layers, T_max), -100, dtype=torch.long) for i, item in enumerate(batch): rc = item.get("residual_codes") if rc is not None: s_start = item["speech_start"] for layer_idx, layer_codes in enumerate(rc): for t_idx, code in enumerate(layer_codes): pos = s_start + t_idx if pos < T_max: residual_codes_tensor[i, layer_idx, pos] = code speaker_ids = None if any(item.get("speaker_id") is not None for item in batch): speaker_ids = torch.tensor( [item.get("speaker_id", 0) for item in batch], dtype=torch.long ) return { "input_ids": input_ids, "labels": labels, "attention_mask": attention_mask, "speech_mask": speech_mask, "loss_weights": loss_weights, "residual_codes": residual_codes_tensor, "speaker_ids": speaker_ids, }