A Palms-On Coding Tutorial for Microsoft VibeVoice Masking Speaker-Conscious ASR, Actual-Time TTS, and Speech-to-Speech Pipelines

!pip uninstall -y transformers -q
!pip set up -q git+
!pip set up -q torch torchaudio speed up soundfile librosa scipy numpy
!pip set up -q huggingface_hub ipywidgets gradio einops
!pip set up -q flash-attn --no-build-isolation 2>/dev/null || echo "flash-attn optional"
!git clone -q --depth 1  /content material/VibeVoice 2>/dev/null || echo "Already cloned"
!pip set up -q -e /content material/VibeVoice


print("="*70)
print("IMPORTANT: If this is your first run, restart the runtime now!")
print("Go to: Runtime -> Restart runtime, then run from CELL 2.")
print("="*70)


import torch
import numpy as np
import soundfile as sf
import warnings
import sys
from IPython.show import Audio, show


warnings.filterwarnings('ignore')
sys.path.insert(0, '/content material/VibeVoice')


import transformers
print(f"Transformers version: {transformers.__version__}")


strive:
   from transformers import VibeVoiceAsrForConditionalGeneration
   print("VibeVoice ASR: Available")
besides ImportError:
   print("ERROR: VibeVoice not available. Please restart runtime and run Cell 1 again.")
   elevate


SAMPLE_PODCAST = "
SAMPLE_GERMAN = "


print("Setup complete!")

from transformers import AutoProcessor, VibeVoiceAsrForConditionalGeneration


print("Loading VibeVoice ASR model (7B parameters)...")
print("First run downloads ~14GB - please wait...")


asr_processor = AutoProcessor.from_pretrained("microsoft/VibeVoice-ASR-HF")
asr_model = VibeVoiceAsrForConditionalGeneration.from_pretrained(
   "microsoft/VibeVoice-ASR-HF",
   device_map="auto",
   torch_dtype=torch.float16,
)


print(f"ASR Model loaded on {asr_model.device}")


def transcribe(audio_path, context=None, output_format="parsed"):
   inputs = asr_processor.apply_transcription_request(
       audio=audio_path,
       immediate=context,
   ).to(asr_model.system, asr_model.dtype)
  
   output_ids = asr_model.generate(**inputs)
   generated_ids = output_ids[:, inputs["input_ids"].form[1]:]
   outcome = asr_processor.decode(generated_ids, return_format=output_format)[0]
  
   return outcome


print("="*70)
print("ASR DEMO: Podcast Transcription with Speaker Diarization")
print("="*70)


print("nPlaying sample audio:")
show(Audio(SAMPLE_PODCAST))


print("nTranscribing with speaker identification...")
outcome = transcribe(SAMPLE_PODCAST, output_format="parsed")


print("nTRANSCRIPTION RESULTS:")
print("-"*70)
for phase in outcome:
   speaker = phase['Speaker']
   begin = phase['Start']
   finish = phase['End']
   content material = phase['Content']
   print(f"n[Speaker {speaker}] {start:.2f}s - {end:.2f}s")
   print(f"  {content}")


print("n" + "="*70)
print("ASR DEMO: Context-Aware Transcription")
print("="*70)


print("nComparing transcription WITH and WITHOUT context hotwords:")
print("-"*70)


result_no_ctx = transcribe(SAMPLE_GERMAN, context=None, output_format="transcription_only")
print(f"nWITHOUT context: {result_no_ctx}")


result_with_ctx = transcribe(SAMPLE_GERMAN, context="About VibeVoice", output_format="transcription_only")
print(f"WITH context:    {result_with_ctx}")


print("nNotice how 'VibeVoice' is recognized correctly when context is provided!")

print("n" + "="*70)
print("ASR DEMO: Batch Processing")
print("="*70)


audio_batch = [SAMPLE_GERMAN, SAMPLE_PODCAST]
prompts_batch = ["About VibeVoice", None]


inputs = asr_processor.apply_transcription_request(
   audio=audio_batch,
   immediate=prompts_batch
).to(asr_model.system, asr_model.dtype)


output_ids = asr_model.generate(**inputs)
generated_ids = output_ids[:, inputs["input_ids"].form[1]:]
transcriptions = asr_processor.decode(generated_ids, return_format="transcription_only")


print("nBatch transcription results:")
print("-"*70)
for i, trans in enumerate(transcriptions):
   preview = trans[:150] + "..." if len(trans) > 150 else trans
   print(f"nAudio {i+1}: {preview}")


from transformers import AutoModelForCausalLM
from vibevoice.modular.modular_vibevoice_text_tokenizer import VibeVoiceTextTokenizerFast


print("n" + "="*70)
print("Loading VibeVoice Realtime TTS model (0.5B parameters)...")
print("="*70)


tts_model = AutoModelForCausalLM.from_pretrained(
   "microsoft/VibeVoice-Realtime-0.5B",
   trust_remote_code=True,
   torch_dtype=torch.float16,
).to("cuda" if torch.cuda.is_available() else "cpu")


tts_tokenizer = VibeVoiceTextTokenizerFast.from_pretrained("microsoft/VibeVoice-Realtime-0.5B")
tts_model.set_ddpm_inference_steps(20)


print(f"TTS Model loaded on {next(tts_model.parameters()).device}")


VOICES = ["Carter", "Grace", "Emma", "Davis"]


def synthesize(textual content, voice="Grace", cfg_scale=3.0, steps=20, save_path=None):
   tts_model.set_ddpm_inference_steps(steps)
   input_ids = tts_tokenizer(textual content, return_tensors="pt").input_ids.to(tts_model.system)
  
   output = tts_model.generate(
       inputs=input_ids,
       tokenizer=tts_tokenizer,
       cfg_scale=cfg_scale,
       return_speech=True,
       show_progress_bar=True,
       speaker_name=voice,
   )
  
   audio = output.audio.squeeze().cpu().numpy()
   sample_rate = 24000
  
   if save_path:
       sf.write(save_path, audio, sample_rate)
       print(f"Saved to: {save_path}")
  
   return audio, sample_rate

print("n" + "="*70)
print("TTS DEMO: Basic Speech Synthesis")
print("="*70)


demo_texts = [
   ("Hello! Welcome to VibeVoice, Microsoft's open-source voice AI.", "Grace"),
   ("This model generates natural, expressive speech in real-time.", "Carter"),
   ("You can choose from multiple voice presets for different styles.", "Emma"),
]


for textual content, voice in demo_texts:
   print(f"nText: {text}")
   print(f"Voice: {voice}")
   audio, sr = synthesize(textual content, voice=voice)
   print(f"Duration: {len(audio)/sr:.2f} seconds")
   show(Audio(audio, fee=sr))


print("n" + "="*70)
print("TTS DEMO: Compare All Voice Presets")
print("="*70)


comparison_text = "VibeVoice produces remarkably natural and expressive speech synthesis."
print(f"nSame text with different voices: "{comparison_text}"n")


for voice in VOICES:
   print(f"Voice: {voice}")
   audio, sr = synthesize(comparison_text, voice=voice, steps=15)
   show(Audio(audio, fee=sr))
   print()


print("n" + "="*70)
print("TTS DEMO: Long-form Speech Generation")
print("="*70)


long_text = """
Welcome to as we speak's expertise podcast! I am excited to share the most recent developments in synthetic intelligence and speech synthesis.


Microsoft's VibeVoice represents a breakthrough in voice AI. In contrast to conventional text-to-speech techniques, which battle with long-form content material, VibeVoice can generate coherent speech for prolonged durations.


The important thing innovation is the ultra-low frame-rate tokenizers working at 7.5 hertz. This preserves audio high quality whereas dramatically enhancing computational effectivity.


The system makes use of a next-token diffusion framework that mixes a big language mannequin for context understanding with a diffusion head for high-fidelity audio technology. This allows pure prosody, acceptable pauses, and expressive speech patterns.


Whether or not you are constructing voice assistants, creating podcasts, or creating accessibility instruments, VibeVoice gives a strong basis to your tasks.


Thanks for listening!
"""


print("Generating long-form speech (this takes a moment)...")
audio, sr = synthesize(long_text.strip(), voice="Carter", cfg_scale=3.5, steps=25)
print(f"nGenerated {len(audio)/sr:.2f} seconds of speech")
show(Audio(audio, fee=sr))


sf.write("/content/longform_output.wav", audio, sr)
print("Saved to: /content/longform_output.wav")


print("n" + "="*70)
print("ADVANCED: Speech-to-Speech Pipeline")
print("="*70)


print("nStep 1: Transcribing input audio...")
transcription = transcribe(SAMPLE_GERMAN, context="About VibeVoice", output_format="transcription_only")
print(f"Transcription: {transcription}")


response_text = f"I understood you said: {transcription} That's a fascinating topic about AI technology!"


print(f"nStep 2: Generating speech response...")
print(f"Response: {response_text}")


audio, sr = synthesize(response_text, voice="Grace", cfg_scale=3.0, steps=20)


print(f"nStep 3: Playing generated response ({len(audio)/sr:.2f}s)")
show(Audio(audio, fee=sr))

import gradio as gr


def tts_gradio(textual content, voice, cfg, steps):
   if not textual content.strip():
       return None
   audio, sr = synthesize(textual content, voice=voice, cfg_scale=cfg, steps=int(steps))
   return (sr, audio)


demo = gr.Interface(
   fn=tts_gradio,
   inputs=[
       gr.Textbox(label="Text to Synthesize", lines=5,
                  value="Hello! This is VibeVoice real-time text-to-speech."),
       gr.Dropdown(choices=VOICES, value="Grace", label="Voice"),
       gr.Slider(1.0, 5.0, value=3.0, step=0.5, label="CFG Scale"),
       gr.Slider(5, 50, value=20, step=5, label="Inference Steps"),
   ],
   outputs=gr.Audio(label="Generated Speech"),
   title="VibeVoice Realtime TTS",
   description="Generate natural speech from text using Microsoft's VibeVoice model.",
)


print("nLaunching interactive TTS interface...")
demo.launch(share=True, quiet=True)


from google.colab import information
import os


print("n" + "="*70)
print("UPLOAD YOUR OWN AUDIO")
print("="*70)


print("nUpload an audio file (wav, mp3, flac, etc.):")
uploaded = information.add()


if uploaded:
   for filename, knowledge in uploaded.objects():
       filepath = f"/content/{filename}"
       with open(filepath, 'wb') as f:
           f.write(knowledge)
      
       print(f"nProcessing: {filename}")
       show(Audio(filepath))
      
       outcome = transcribe(filepath, output_format="parsed")
      
       print("nTranscription:")
       print("-"*50)
       if isinstance(outcome, checklist):
           for seg in outcome:
               print(f"[{seg.get('Start',0):.2f}s-{seg.get('End',0):.2f}s] Speaker {seg.get('Speaker',0)}: {seg.get('Content','')}")
       else:
           print(outcome)
else:
   print("No file uploaded - skipping this step")


print("n" + "="*70)
print("MEMORY OPTIMIZATION TIPS")
print("="*70)


print("""
1. REDUCE ASR CHUNK SIZE (if out of reminiscence with lengthy audio):
  output_ids = asr_model.generate(**inputs, acoustic_tokenizer_chunk_size=64000)


2. USE BFLOAT16 DTYPE:
  mannequin = VibeVoiceAsrForConditionalGeneration.from_pretrained(
      model_id, torch_dtype=torch.bfloat16, device_map="auto")


3. REDUCE TTS INFERENCE STEPS (sooner however decrease high quality):
  tts_model.set_ddpm_inference_steps(10)


4. CLEAR GPU CACHE:
  import gc
  torch.cuda.empty_cache()
  gc.gather()


5. GRADIENT CHECKPOINTING FOR TRAINING:
  mannequin.gradient_checkpointing_enable()
""")


print("n" + "="*70)
print("DOWNLOAD GENERATED FILES")
print("="*70)


output_files = ["/content/longform_output.wav"]


for filepath in output_files:
   if os.path.exists(filepath):
       print(f"Downloading: {os.path.basename(filepath)}")
       information.obtain(filepath)
   else:
       print(f"File not found: {filepath}")


print("n" + "="*70)
print("TUTORIAL COMPLETE!")
print("="*70)


print("""
WHAT YOU LEARNED:


VIBEVOICE ASR (Speech-to-Textual content):
 - 60-minute single-pass transcription
 - Speaker diarization (who stated what, when)
 - Context-aware hotword recognition
 - 50+ language help
 - Batch processing


VIBEVOICE REALTIME TTS (Textual content-to-Speech):
 - Actual-time streaming (~300ms latency)
 - A number of voice presets
 - Lengthy-form technology (~10 minutes)
 - Configurable high quality/pace


RESOURCES:
 GitHub:     
 ASR Mannequin:  
 TTS Mannequin:  
 ASR Paper:  
 TTS Paper:  


RESPONSIBLE USE:
 - That is for analysis/growth solely
 - All the time disclose AI-generated content material
 - Don't use for impersonation or fraud
 - Comply with relevant legal guidelines and laws
""")