Реализация потокового синтеза речи в реальном времени (Streaming TTS)

Реализация потокового синтеза речи в реальном времени (Streaming TTS)

Streaming TTS начинает воспроизводить аудио до завершения генерации всего текста — time-to-first-audio падает с 1–3 сек до 100–400 мс. Критично для голосовых ботов, где пользователь слышит ответ немедленно.

Принцип streaming TTS

Текст разбивается на предложения → каждое предложение синтезируется отдельно → чанки отдаются клиенту по мере готовности. Пока клиент слушает первый чанк, сервер готовит второй.

Реализация с OpenAI TTS Streaming

from openai import AsyncOpenAI
import asyncio

client = AsyncOpenAI()

async def stream_tts(text: str):
    """Потоковая генерация TTS через OpenAI"""
    async with client.audio.speech.with_streaming_response.create(
        model="tts-1",
        voice="alloy",
        input=text,
        response_format="pcm",  # raw PCM для минимальной задержки
    ) as response:
        async for chunk in response.iter_bytes(chunk_size=4096):
            yield chunk

WebSocket сервер для real-time TTS

from fastapi import FastAPI, WebSocket
from TTS.api import TTS
import numpy as np
import asyncio

app = FastAPI()
tts = TTS("tts_models/multilingual/multi-dataset/xtts_v2").to("cuda")

def split_into_sentences(text: str) -> list[str]:
    import re
    sentences = re.split(r'(?<=[.!?])\s+', text)
    return [s.strip() for s in sentences if s.strip()]

@app.websocket("/tts-stream")
async def tts_websocket(websocket: WebSocket):
    await websocket.accept()
    try:
        while True:
            text = await websocket.receive_text()
            sentences = split_into_sentences(text)

            for sentence in sentences:
                wav = await asyncio.get_event_loop().run_in_executor(
                    None,
                    lambda s=sentence: tts.tts(
                        text=s,
                        language="ru",
                        speaker_wav="default.wav"
                    )
                )
                # Конвертируем в bytes и отправляем
                audio_bytes = (np.array(wav) * 32767).astype(np.int16).tobytes()
                await websocket.send_bytes(audio_bytes)

            # Сигнал конца синтеза
            await websocket.send_json({"type": "done"})
    except Exception:
        await websocket.close()

Оптимизация задержки

Chunk-based синтез: разбиваем текст на фразы по 10–20 слов, синтезируем и стримим параллельно.

Предгенерация для шаблонных фраз: «Подождите, пожалуйста», «Один момент» — кэшируются и отдаются без синтеза.

ElevenLabs Streaming API:

from elevenlabs.client import ElevenLabs

client = ElevenLabs()
audio_stream = client.text_to_speech.convert_as_stream(
    voice_id="voice_id",
    text="Текст для потокового синтеза",
    model_id="eleven_turbo_v2_5",  # 75 мс задержка
)

Time-to-first-audio по движкам

Сроки: интеграция облачного streaming TTS — 2–3 дня. Self-hosted WebSocket TTS сервер — 1 неделя.