talklabs 2.1.2

TalkLabs SDK - Ultra-low latency Text-to-Speech with intelligent streaming and persistent sessions (ElevenLabs compatible)

🐍 TalkLabs Python SDK

SDK oficial da TalkLabs para síntese de voz com streaming ultra-baixa latência

🚀 Quick Start • 📚 Documentação • 💡 Exemplos • 🆘 Suporte

---

🚀 v2.1.1: Streaming com latência de ~200-500ms + Sessões Persistentes!

Características

• ✅ Compatível com ElevenLabs: Drop-in replacement para APIs existentes
• ⚡ Ultra-Low Latency: ~200-500ms até primeiro áudio (vs 5-10s tradicional)
• 🧠 Processamento Inteligente: Segmentação natural avançada
• 📡 Streaming Otimizado: Sistema paralelo com 3 níveis de prioridade
• 🎧 Real-time Playback: Chunks de áudio prontos para reprodução imediata
• 🔄 Incremental Streaming: Envio palavra-por-palavra para máxima responsividade

Instalação

pip install talklabs

Uso Rápido

1. Geração Simples (Síncrona)

from talklabs import TalkLabsClient

client = TalkLabsClient(api_key="tlk_live_xxxxx")

audio = client.generate(
    text="Olá! Bem-vindo ao TalkLabs.",
    voice="yasmin_alves"
)

with open("output.wav", "wb") as f:
    f.write(audio)

2. 🚀 Streaming WebSocket (Ultra-Low Latency)

import asyncio
from talklabs import TalkLabsClient

async def stream_example():
    client = TalkLabsClient(api_key="tlk_live_xxxxx")

    # Streaming via WebSocket (latência ~200-500ms)
    async for audio_chunk in client.stream_text(
        text="Este é um teste de ultra baixa latência!",
        voice="yasmin_alves",
        language="pt"
    ):
        # Reproduzir audio_chunk imediatamente
        # Ex: play_audio(audio_chunk) ou salvar em arquivo
        print(f"Chunk recebido: {len(audio_chunk)} bytes")

asyncio.run(stream_example())

3. Sessão Persistente (RECOMENDADO para Produção)

async def persistent_session_example():
    client = TalkLabsClient(api_key="tlk_live_xxxxx")

    # Criar sessão persistente (mantém conexão aberta)
    session = await client.create_session(
        voice="yasmin_alves",
        language="pt",
        speed=1.0
    )

    # Múltiplas sínteses na mesma sessão (sem reconectar)
    for text in ["Primeira frase.", "Segunda frase.", "Terceira frase."]:
        async for audio_chunk in session.stream_text(text):
            print(f"Chunk: {len(audio_chunk)} bytes")

    # Fechar sessão quando terminar
    await session.close()

asyncio.run(persistent_session_example())

4. Streaming HTTP (Método Alternativo)

# Streaming tradicional via HTTP
for chunk in client.generate_stream(
    text="Streaming HTTP incremental",
    voice="yasmin_alves"
):
    # Processa chunks
    pass

API Reference

TalkLabsClient

Métodos Principais

generate(text, voice, **kwargs) → bytes

• Geração síncrona completa via HTTP
• Retorna áudio WAV completo
• Útil para textos curtos e testes simples
• Parâmetros:
  • text: Texto para sintetizar
  • voice: ID da voz (ex: "yasmin_alves", "adam_rocha")
  • language: Idioma ("pt", "en", "es", etc) - padrão: "pt"
  • speed: Velocidade (0.5-2.0) - padrão: 1.0
  • voice_settings: Configurações opcionais de voz

generate_stream(text, voice, **kwargs) → Iterator[bytes]

• Streaming HTTP tradicional
• Retorna chunks progressivamente via HTTP
• Alternativa quando WebSocket não está disponível
• Parâmetros: mesmos do generate()

stream_text(text, voice, **kwargs) → AsyncIterator[bytes] ⚡ RECOMENDADO

• Streaming via WebSocket com ultra-baixa latência (~200-500ms)
• Conexão one-shot (abre e fecha para cada síntese)
• Retorna chunks de áudio conforme são gerados
• Parâmetros: mesmos do generate()

create_session(voice, **kwargs) → StreamingSession 🎯 MELHOR PARA PRODUÇÃO

• Cria sessão persistente que mantém conexão WebSocket aberta
• Ideal para múltiplas sínteses sem overhead de reconexão
• Parâmetros:
  • voice: ID da voz para a sessão
  • language: Idioma padrão da sessão
  • speed: Velocidade padrão
  • voice_settings: Configurações de voz
  • ping_interval: Intervalo de keep-alive (padrão: 20s)
  • ping_timeout: Timeout do keep-alive (padrão: 20s)

get_voices() → list

• Lista todas as vozes disponíveis
• Retorna array com metadados de cada voz

Vozes Disponíveis

# Listar todas as vozes
voices = client.get_voices()
for voice in voices:
    print(f"{voice['voice_id']}: {voice['name']}")

Vozes Populares:

• yasmin_alves - Português (BR) - Feminina
• adam_rocha - Português (BR) - Masculina
• maria_silva - Português (PT) - Feminina
• joao_santos - Português (PT) - Masculina

StreamingSession

Classe para sessões persistentes. Métodos disponíveis:

stream_text(text) → AsyncIterator[bytes]

• Sintetiza texto usando a sessão existente
• Não reconecta, usa WebSocket já aberto
• Mesma assinatura do método principal

close()

• Fecha a conexão WebSocket
• Sempre chame ao terminar de usar a sessão

Exemplo com context manager:

async with await client.create_session(voice="yasmin_alves") as session:
    async for chunk in session.stream_text("Olá!"):
        process(chunk)
    # close() é chamado automaticamente

Arquitetura do Streaming Otimizado

Como Funciona

1. Segmentação Inteligente: Texto é dividido em sentenças naturais
2. Sistema de Filas: Chunks são processados com prioridades:
   • P1 (Alta): Primeira sentença - processada imediatamente
   • P2 (Média): Sentenças intermediárias
   • P3 (Baixa): Última sentença
3. Processamento Paralelo: TTS processa chunks simultaneamente
4. Streaming Real-time: Áudio retorna conforme é gerado

Benefícios

• ⚡ Latência 95% menor: ~200-500ms vs 5-10s
• 🎯 Primeira Palavra Rápida: Usuário ouve resposta quase instantânea
• 📊 Escalável: Suporta múltiplas sessões simultâneas
• 🧠 Inteligente: Quebras naturais de sentença garantidas

💡 Exemplos Práticos

Exemplos Disponíveis

Confira nossos exemplos completos na pasta examples/:

• quick_start.py - Teste rápido para começar
• generate_simple.py - Geração síncrona simples
• generate_stream.py - Streaming HTTP
• stream_websocket.py - Streaming WebSocket (ultra-baixa latência)
• persistent_session.py - Sessões persistentes (recomendado para produção)
• get_voices.py - Listar vozes disponíveis

Nota: Os exemplos estão incluídos no pacote. Após instalar, você pode copiá-los ou consultá-los diretamente no repositório GitHub.

Exemplo: Salvar Chunks Progressivamente

async def save_streaming():
    client = TalkLabsClient(api_key="tlk_live_xxxxx")

    with open("output_streaming.wav", "wb") as f:
        async for chunk in client.stream_text(
            text="Este áudio será salvo em tempo real.",
            voice="yasmin_alves"
        ):
            f.write(chunk)

    print("Áudio salvo!")

asyncio.run(save_streaming())

Reprodução em Tempo Real com pyaudio

import pyaudio
import asyncio
from talklabs import TalkLabsClient

async def play_realtime():
    client = TalkLabsClient(api_key="tlk_live_xxxxx")

    # Inicializar pyaudio
    p = pyaudio.PyAudio()
    stream = p.open(format=pyaudio.paInt16, channels=1, rate=24000, output=True)

    try:
        async for chunk in client.stream_text(
            text="Olá! Este áudio está sendo reproduzido em tempo real.",
            voice="yasmin_alves"
        ):
            # Reproduzir imediatamente
            stream.write(chunk)
    finally:
        stream.stop_stream()
        stream.close()
        p.terminate()

asyncio.run(play_realtime())

Configurações Avançadas de Voz

from talklabs import TalkLabsClient, VoiceSettings

client = TalkLabsClient(api_key="tlk_live_xxxxx")

settings = VoiceSettings(
    stability=0.85,           # Estabilidade da voz (0-1)
    similarity_boost=0.75,    # Similaridade com voz original
    style=0.0,                # Estilo expressivo (0-1)
    use_speaker_boost=True    # Boost de clareza
)

audio = client.generate(
    text="Teste com configurações customizadas",
    voice="yasmin_alves",
    voice_settings=settings,
    speed=1.2  # 20% mais rápido
)

Compatibilidade com ElevenLabs

Este SDK é 100% compatível com o SDK da ElevenLabs. Basta trocar:

# ElevenLabs
from elevenlabs import ElevenLabs
client = ElevenLabs(api_key="...")

# TalkLabs (drop-in replacement)
from talklabs import TalkLabsClient
client = TalkLabsClient(api_key="tlk_live_...")

Configuração

Base URL

• Produção: https://api.talklabs.com.br
• Local: http://localhost:5000 (desenvolvimento)

Endpoints

• HTTP: /v1/text-to-speech/{voice_id}
• WebSocket: /v1/text-to-speech/{voice_id}/stream

Troubleshooting

Erro: "Connection refused"

Verifique se a API está rodando:

curl https://api.talklabs.com.br/health

Latência alta no streaming

1. Use stream_text() ou sessão persistente ao invés de generate_stream()
2. Verifique sua conexão com a API
3. Certifique-se que está usando a região mais próxima

Chunks de áudio corrompidos

• Certifique-se de salvar/reproduzir como WAV 24kHz mono
• Use io.BytesIO para buffer temporário se necessário

---

🆘 Suporte

• 📧 Email: [e-mail]
• 🌐 Website: https://talklabs.com.br
• 💡 Exemplos: Ver pasta examples/ incluída no pacote
• 📚 Documentação: Ver seção API Reference acima

---

📄 Licença

MIT License - veja LICENSE para detalhes.

---

Desenvolvido com ❤️ pela equipe TalkLabs