""" Teste de Time-To-First-Token (TTFT) para Orpheus TTS Mede a latência até o primeiro chunk de áudio ser gerado """ import os os.environ["VLLM_ATTENTION_BACKEND"] = "FLASH_ATTN" import torch import time import asyncio from transformers import AutoTokenizer from vllm import AsyncLLMEngine, AsyncEngineArgs, SamplingParams START_TOKEN = 128259 END_TOKENS = [128009, 128260, 128261, 128257] STOP_TOKEN = 128258 AUDIO_TOKEN_BASE = 128266 async def main(): print("=" * 70) print("TESTE DE TTFT (Time-To-First-Token) - ORPHEUS TTS") print("=" * 70) print("\n[1] Carregando modelo...") tokenizer = AutoTokenizer.from_pretrained("canopylabs/orpheus-3b-0.1-ft") engine_args = AsyncEngineArgs( model="canopylabs/orpheus-3b-0.1-ft", dtype="bfloat16", max_model_len=4096, gpu_memory_utilization=0.9, max_num_seqs=16, enable_chunked_prefill=True, enable_prefix_caching=True, enforce_eager=False, ) engine = AsyncLLMEngine.from_engine_args(engine_args) sampling_params = SamplingParams( temperature=0.2, top_p=0.9, max_tokens=4096, stop_token_ids=[STOP_TOKEN], repetition_penalty=1.1, ) def format_prompt(text, voice="tara"): adapted_prompt = f"{voice}: {text}" prompt_tokens = tokenizer(adapted_prompt, return_tensors="pt") start_token = torch.tensor([[START_TOKEN]], dtype=torch.int64) end_tokens = torch.tensor([END_TOKENS], dtype=torch.int64) all_input_ids = torch.cat([start_token, prompt_tokens.input_ids, end_tokens], dim=1) return tokenizer.decode(all_input_ids[0]) async def measure_ttft(text, request_id): """Mede o tempo até o primeiro token de áudio""" prompt_string = format_prompt(text) start = time.time() ttft = None ttfa = None # Time to first audio token first_audio_frame = None # Time to first complete 7-token frame audio_token_count = 0 total_tokens = 0 async for output in engine.generate(prompt_string, sampling_params, request_id): total_tokens = len(output.outputs[0].token_ids) # Medir TTFT (primeiro token qualquer) if ttft is None and total_tokens > 0: ttft = time.time() - start # Medir TTFA (primeiro token de áudio) if ttfa is None: for t in output.outputs[0].token_ids: if t >= AUDIO_TOKEN_BASE: ttfa = time.time() - start break # Contar tokens de áudio audio_token_count = sum(1 for t in output.outputs[0].token_ids if t >= AUDIO_TOKEN_BASE) # Medir tempo até primeiro frame completo (7 tokens de áudio) if first_audio_frame is None and audio_token_count >= 7: first_audio_frame = time.time() - start total_time = time.time() - start return { 'text': text[:40], 'ttft': ttft, # Time to first token 'ttfa': ttfa, # Time to first audio token 'ttff': first_audio_frame, # Time to first audio frame (7 tokens = ~23ms de áudio) 'total_time': total_time, 'total_tokens': total_tokens, 'audio_tokens': audio_token_count, } test_texts = [ "Hello, how are you doing today?", "The weather is beautiful outside.", "I love programming with Python.", "Machine learning is fascinating.", "Can you help me with this task?", "Let me explain how this works.", "This is a test of the system.", "Technology is amazing these days.", "Have a wonderful day ahead.", "Thank you for your patience.", "Let's work together on this.", "The future looks very bright.", "I appreciate your help today.", "This demonstration is working.", "Audio generation is fast now.", "Real-time speech synthesis.", ] print("\n[2] Teste SEQUENCIAL (baseline TTFT)...") print("-" * 70) # Warmup print(" Warmup...") await measure_ttft("Warmup test.", "warmup") seq_results = [] for i, text in enumerate(test_texts[:4]): result = await measure_ttft(text, f"seq_{i}") seq_results.append(result) print(f" [{i+1}] TTFT: {result['ttft']*1000:.0f}ms | TTFA: {result['ttfa']*1000:.0f}ms | TTFF: {result['ttff']*1000:.0f}ms | Total: {result['total_time']:.2f}s") avg_ttft = sum(r['ttft'] for r in seq_results) / len(seq_results) avg_ttfa = sum(r['ttfa'] for r in seq_results) / len(seq_results) avg_ttff = sum(r['ttff'] for r in seq_results) / len(seq_results) print(f"\n MÉDIA SEQUENCIAL:") print(f" - TTFT (primeiro token): {avg_ttft*1000:.0f}ms") print(f" - TTFA (primeiro áudio): {avg_ttfa*1000:.0f}ms") print(f" - TTFF (primeiro frame): {avg_ttff*1000:.0f}ms (~23ms de áudio)") # Teste com requisições simultâneas print("\n" + "=" * 70) print("[3] Teste TTFT com requisições SIMULTÂNEAS") print("=" * 70) for num_concurrent in [4, 8, 12, 16]: print(f"\n>>> {num_concurrent} REQUISIÇÕES SIMULTÂNEAS <<<") print("-" * 50) texts = test_texts[:num_concurrent] tasks = [measure_ttft(text, f"par_{num_concurrent}_{i}") for i, text in enumerate(texts)] results = await asyncio.gather(*tasks) ttfts = [r['ttft'] for r in results] ttfas = [r['ttfa'] for r in results] ttffs = [r['ttff'] for r in results if r['ttff']] print(f" TTFT - Min: {min(ttfts)*1000:.0f}ms | Max: {max(ttfts)*1000:.0f}ms | Avg: {sum(ttfts)/len(ttfts)*1000:.0f}ms") print(f" TTFA - Min: {min(ttfas)*1000:.0f}ms | Max: {max(ttfas)*1000:.0f}ms | Avg: {sum(ttfas)/len(ttfas)*1000:.0f}ms") if ttffs: print(f" TTFF - Min: {min(ttffs)*1000:.0f}ms | Max: {max(ttffs)*1000:.0f}ms | Avg: {sum(ttffs)/len(ttffs)*1000:.0f}ms") # Mostrar cada resultado print(f"\n Detalhes por requisição:") for i, r in enumerate(results): print(f" [{i+1}] TTFT: {r['ttft']*1000:.0f}ms | TTFA: {r['ttfa']*1000:.0f}ms | TTFF: {r['ttff']*1000:.0f}ms") print("\n" + "=" * 70) print("RESUMO - LATÊNCIA PARA PRIMEIRO ÁUDIO") print("=" * 70) print(""" TTFT = Time to First Token (qualquer token) TTFA = Time to First Audio token TTFF = Time to First Frame (7 tokens = ~23ms de áudio playable) Para streaming de áudio, o TTFF é o mais importante: é quando você tem o primeiro chunk de áudio para tocar. """) print("=" * 70) if __name__ == "__main__": asyncio.run(main())