""" Self-speculative decoding for Speech LLMs. """ import argparse import math import os import torch import torch.nn.functional as F import evaluate from normalizer import data_utils import time from tqdm import tqdm from transformers import AutoProcessor, AutoModelForSpeechSeq2Seq, models assert hasattr(models, "granite_speech") wer_metric = evaluate.load("wer") torch.set_float32_matmul_precision('high') def main(args): device = f"cuda:{args.device}" if args.device >= 0 else "cpu" processor = AutoProcessor.from_pretrained(args.model_id) tokenizer = processor.tokenizer model = AutoModelForSpeechSeq2Seq.from_pretrained(args.model_id, torch_dtype=torch.bfloat16).to(device) model.eval() print(f"Model size: {sum(p.numel() for p in model.parameters()) / 1e9:.2f}B parameters") logits_scaling = getattr(model.language_model.config, 'logits_scaling', 1.0) # ========== Chat Template Setup ========== text_instruction = "<|audio|>can you transcribe the speech into a written format?" # Build chat message and apply template message = [ {"role": "user", "content": text_instruction}, ] text_prompt = tokenizer.apply_chat_template(message, tokenize=False, add_generation_prompt=True) # Derive prefix and suffix from the formatted prompt prompt_prefix, prompt_suffix = text_prompt.split("<|audio|>") # Cache prompt embeddings (suffix excludes <|audio|> since we insert audio embeds separately) embed_layer = model.language_model.get_input_embeddings() prefix_ids = tokenizer.encode(prompt_prefix, add_special_tokens=False) suffix_ids = tokenizer.encode(prompt_suffix, add_special_tokens=False) cached_prefix_embeds = embed_layer(torch.tensor([prefix_ids], device=device)) cached_suffix_embeds = embed_layer(torch.tensor([suffix_ids], device=device)) HOP_LENGTH = 160 confidence_threshold = args.confidence_threshold ctc_threshold = args.ctc_threshold @torch.no_grad() def ctc_decode(audios): """CTC decode with entropy-based confidence.""" texts = [text_prompt] * len(audios) model_inputs = processor(texts, audios, device=device, return_tensors="pt").to(device) with torch.amp.autocast(device_type='cuda', dtype=torch.bfloat16): encoder_output = model.encoder(model_inputs["input_features"]) embeddings = encoder_output.last_hidden_state if hasattr(encoder_output, 'last_hidden_state') else encoder_output ctc_logits = model.encoder.out(embeddings) ctc_probs = F.softmax(ctc_logits.float(), dim=-1) _, idx_batch = ctc_probs.max(dim=-1) entropy = -(ctc_probs * torch.log(ctc_probs + 1e-10)).sum(dim=-1) ctc_texts, ctc_entropies, embed_lengths = [], [], [] for i, idx in enumerate(idx_batch): dedup = torch.unique_consecutive(idx, dim=-1) non_blank = dedup[dedup > 0].tolist() ctc_texts.append(''.join(chr(c) for c in non_blank)) ctc_entropies.append(entropy[i].max().item() if non_blank else float('inf')) embed_lengths.append(len(audios[i]) // HOP_LENGTH // 2 + 1) return ctc_texts, ctc_entropies, embeddings, embed_lengths @torch.no_grad() def verify(ctc_texts, embeddings, embed_lengths): """Verify CTC outputs with LLM.""" batch_sz = len(ctc_texts) ctc_token_ids = [] for text in ctc_texts: text = text.strip() if text else "" ctc_token_ids.append(tokenizer.encode(text, add_special_tokens=False) if text else []) with torch.amp.autocast(device_type='cuda', dtype=torch.bfloat16): audio_embeds = model.projector(embeddings) max_proj_len = audio_embeds.shape[1] window_size, downsample_rate = model.config.window_size, model.config.downsample_rate num_queries = window_size // downsample_rate proj_lengths = [min(math.ceil(enc_len / window_size) * num_queries, max_proj_len) for enc_len in embed_lengths] if not any(ctc_token_ids): return [(False, ctc_texts[i]) for i in range(batch_sz)], audio_embeds, proj_lengths audio_token_id = model.config.audio_token_id all_input_ids, prompt_lens, audio_ranges = [], [], [] for i, proj_len in enumerate(proj_lengths): audio_start = len(prefix_ids) audio_ranges.append((audio_start, audio_start + proj_len)) prompt_part = prefix_ids + [audio_token_id] * proj_len + suffix_ids prompt_lens.append(len(prompt_part)) all_input_ids.append(prompt_part + ctc_token_ids[i]) max_len = max(len(ids) for ids in all_input_ids) padded_ids = torch.full((batch_sz, max_len), tokenizer.pad_token_id, dtype=torch.long, device=device) attn_mask = torch.zeros(batch_sz, max_len, dtype=torch.long, device=device) for i, ids in enumerate(all_input_ids): padded_ids[i, :len(ids)] = torch.tensor(ids, dtype=torch.long, device=device) attn_mask[i, :len(ids)] = 1 inputs_embeds = model.language_model.get_input_embeddings()(padded_ids) for i in range(batch_sz): s, e = audio_ranges[i] inputs_embeds[i, s:e, :] = audio_embeds[i, :e-s, :] with torch.amp.autocast(device_type='cuda', dtype=torch.bfloat16): hidden = model.language_model.model(attention_mask=attn_mask, inputs_embeds=inputs_embeds, use_cache=False).last_hidden_state # Gather hidden states at verification positions sample_idx, pos_idx, ctc_flat = [], [], [] sample_ranges, sample_valid = [], [] offset = 0 for i in range(batch_sz): ctc_tokens = ctc_token_ids[i] if not ctc_tokens or prompt_lens[i] - 1 + len(ctc_tokens) > hidden.shape[1]: sample_ranges.append((offset, offset)) sample_valid.append(False) continue verify_start = prompt_lens[i] - 1 for k in range(len(ctc_tokens)): sample_idx.append(i) pos_idx.append(verify_start + k) ctc_flat.append(ctc_tokens[k]) sample_ranges.append((offset, offset + len(ctc_tokens))) sample_valid.append(True) offset += len(ctc_tokens) if pos_idx: gathered = hidden[torch.tensor(sample_idx, device=device), torch.tensor(pos_idx, device=device), :] with torch.amp.autocast(device_type='cuda', dtype=torch.bfloat16): logits = model.language_model.lm_head(gathered) / logits_scaling probs = F.softmax(logits.float(), dim=-1) ctc_probs = probs[torch.arange(len(ctc_flat), device=device), torch.tensor(ctc_flat, device=device)] results = [] for i in range(batch_sz): s, e = sample_ranges[i] if not sample_valid[i]: results.append((False, ctc_texts[i])) continue token_probs = ctc_probs[s:e] accepted = (token_probs >= confidence_threshold).all().item() results.append((accepted, ctc_texts[i])) return results, audio_embeds, proj_lengths @torch.no_grad() def fallback(audio_embeds, indices, proj_lengths): """AR fallback for failed samples.""" if not indices: return [] batch_sz = len(indices) hidden_dim = audio_embeds.shape[-1] all_embeds, all_lengths = [], [] for i in indices: sample_embeds = audio_embeds[i, :proj_lengths[i], :].unsqueeze(0) combined = torch.cat([cached_prefix_embeds, sample_embeds, cached_suffix_embeds], dim=1) all_embeds.append(combined.squeeze(0)) all_lengths.append(combined.shape[1]) max_len = max(all_lengths) padded = torch.zeros(batch_sz, max_len, hidden_dim, device=device, dtype=audio_embeds.dtype) attn_mask = torch.zeros(batch_sz, max_len, dtype=torch.long, device=device) for i, (emb, length) in enumerate(zip(all_embeds, all_lengths)): padded[i, max_len - length:] = emb attn_mask[i, max_len - length:] = 1 outputs = model.language_model.generate( inputs_embeds=padded, attention_mask=attn_mask, bos_token_id=tokenizer.bos_token_id, pad_token_id=tokenizer.pad_token_id, eos_token_id=tokenizer.eos_token_id, max_new_tokens=args.max_new_tokens, num_beams=args.num_beams, early_stopping=args.num_beams > 1, do_sample=False, use_cache=True ) return [tokenizer.decode(outputs[i], skip_special_tokens=True) for i in range(batch_sz)] def benchmark(batch): audios = [audio["array"] for audio in batch["audio"]] batch_sz = len(audios) sampling_rate = batch["audio"][0]["sampling_rate"] batch["audio_length_s"] = [len(audio) / sampling_rate for audio in audios] batch["audio_filepath"] = data_utils.extract_audio_filepaths_from_batch(batch, batch_sz) start_time = time.time() # Step 1: CTC decode ctc_texts, ctc_entropies, embeddings, embed_lengths = ctc_decode(audios) # Step 2: Gate by CTC entropy predictions = [None] * batch_sz verify_idx = [] for i, (text, ent) in enumerate(zip(ctc_texts, ctc_entropies)): if ent <= ctc_threshold and text.strip(): predictions[i] = text.strip() else: verify_idx.append(i) # Step 3: Verify remaining if verify_idx: verify_emb = embeddings[verify_idx] verify_lens = [embed_lengths[i] for i in verify_idx] verify_texts = [ctc_texts[i] for i in verify_idx] results, audio_embeds, proj_lengths = verify(verify_texts, verify_emb, verify_lens) fail_idx = [] for j, (accepted, text) in enumerate(results): i = verify_idx[j] if accepted: predictions[i] = text.strip() else: fail_idx.append(j) # Step 4: Fallback if fail_idx: fallback_texts = fallback(audio_embeds, fail_idx, proj_lengths) for k, j in enumerate(fail_idx): predictions[verify_idx[j]] = fallback_texts[k] runtime = time.time() - start_time batch["transcription_time_s"] = [runtime / batch_sz] * batch_sz batch["predictions"] = predictions # raw; normalization applied at scoring time batch["references"] = batch["original_text"] # raw; normalization applied at scoring time return batch # Load and process dataset dataset = data_utils.load_data(args) if args.max_eval_samples is not None and args.max_eval_samples > 0: print(f"Subsampling to {args.max_eval_samples} samples") dataset = dataset.select(range(min(args.max_eval_samples, len(dataset)))) dataset = data_utils.prepare_data(dataset) dataset = dataset.map(benchmark, batch_size=args.batch_size, batched=True, remove_columns=["audio"], desc="Processing") all_results = {"audio_length_s": [], "transcription_time_s": [], "predictions": [], "references": [], "audio_filepath": []} for result in tqdm(dataset, desc="Samples"): for key in all_results: all_results[key].append(result[key]) # Write results manifest_path = data_utils.write_manifest( all_results["references"], all_results["predictions"], args.model_id, args.dataset_path, args.dataset, args.split, audio_length=all_results["audio_length_s"], transcription_time=all_results["transcription_time_s"], audio_filepaths=all_results["audio_filepath"], ) print("Results saved at:", os.path.abspath(manifest_path)) norm_refs = [data_utils.normalizer(r) for r in all_results["references"]] norm_preds = [data_utils.normalizer(p) for p in all_results["predictions"]] wer = round(100 * wer_metric.compute(references=norm_refs, predictions=norm_preds), 2) rtfx = round(sum(all_results["audio_length_s"]) / sum(all_results["transcription_time_s"]), 2) print(f"WER: {wer}%, RTFx: {rtfx}") if __name__ == "__main__": parser = argparse.ArgumentParser() parser.add_argument("--model_id", type=str, required=True) parser.add_argument("--dataset_path", type=str, default="hf-audio/open-asr-leaderboard") parser.add_argument("--dataset", type=str, required=True) parser.add_argument("--split", type=str, default="test") parser.add_argument("--device", type=int, default=-1) parser.add_argument("--batch_size", type=int, default=16) parser.add_argument("--max_eval_samples", type=int, default=None) parser.add_argument("--max_new_tokens", type=int, default=200) parser.add_argument("--num_beams", type=int, default=1) parser.add_argument("--confidence_threshold", type=float, default=0.01) parser.add_argument("--ctc_threshold", type=float, default=0.5) parser.add_argument("--streaming", action="store_true", help="Stream the dataset lazily over the network instead of downloading it in full before the evaluation.") args = parser.parse_args() main(args)