Реалізація потокового синтезу мови в реальному часі (Streaming TTS) Streaming TTS починає відтворювати аудіо до завершення генерації всього тексту - time-to-first-audio падає з 1-3 с до 100-400 мс. Критично для голосових роботів, де користувач чує відповідь негайно. ### Принцип streaming TTS Текст розбивається на пропозиції → кожна пропозиція синтезується окремо → чанки віддаються клієнту в міру готовності. Поки клієнт слухає перший чанк, сервер готує другий. ### Реалізація з OpenAI TTS Streaming```python
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 TTSpython
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**:python
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 мс задержка )