Примеры интеграции

Быстрый тест с wscat

Установите wscat и подключитесь:

bash

wscat -c "wss://api.genvoice.ru/v1/api/tts/stream"

После подключения отправьте:

json

{"event": "session.begin", "api_key": "sk_live_YOUR_API_KEY", "voice_id": "9efc2a63-911e-4e19-9d5e-01b0640fc4e6", "output_format": "pcm_24000"}

Дождитесь session.ready, затем:

json

{"event": "text.chunk", "text": "Привет, мир!", "flush": true}
{"event": "text.end"}

Вы получите audio.chunk сообщения с base64-аудио и финальный generation.complete.

Python-клиент (aiohttp)

Установка

bash

pip install aiohttp

Полный клиент

python

import asyncio
import base64
import json
import wave
from pathlib import Path

import aiohttp


class GenVoiceTTSError(Exception):
    """Ошибка от сервера GenVoice."""

    def __init__(self, code: str, message: str):
        self.code = code
        self.message = message
        super().__init__(f"[{code}] {message}")


class GenVoiceRealtimeClient:
    """WebSocket-клиент для Realtime TTS GenVoice."""

    WS_URL = "wss://api.genvoice.ru/v1/api/tts/stream"

    FORMAT_PARAMS = {
        "pcm_8000": {"sample_rate": 8000, "sample_width": 2},
        "pcm_16000": {"sample_rate": 16000, "sample_width": 2},
        "pcm_22050": {"sample_rate": 22050, "sample_width": 2},
        "pcm_24000": {"sample_rate": 24000, "sample_width": 2},
        "ulaw_8000": {"sample_rate": 8000, "sample_width": 1},
        "alaw_8000": {"sample_rate": 8000, "sample_width": 1},
    }

    def __init__(
        self,
        api_key: str,
        voice_id: str,
        output_format: str = "pcm_24000",
        speed: float = 1.0,
    ):
        self.api_key = api_key
        self.voice_id = voice_id
        self.output_format = output_format
        self.speed = speed
        self._session: aiohttp.ClientSession | None = None
        self._ws: aiohttp.ClientWebSocketResponse | None = None
        self._session_id: str | None = None

    async def __aenter__(self):
        self._session = aiohttp.ClientSession()
        await self._connect()
        return self

    async def __aexit__(self, *exc):
        await self.close()

    async def _connect(self) -> None:
        """Подключиться и аутентифицироваться."""
        self._ws = await self._session.ws_connect(self.WS_URL)
        await self._ws.send_json({
            "event": "session.begin",
            "api_key": self.api_key,
            "voice_id": self.voice_id,
            "speed": self.speed,
            "output_format": self.output_format,
            "inactivity_timeout": 60,
        })

        msg = await asyncio.wait_for(self._ws.receive(), timeout=15)
        data = json.loads(msg.data)

        if data.get("event") == "session.ready":
            self._session_id = data.get("session_id")
        elif data.get("event") == "error":
            raise GenVoiceTTSError(data["code"], data["message"])
        else:
            raise RuntimeError(f"Unexpected response: {data}")

    async def synthesize(self, text: str, flush: bool = True) -> bytes:
        """Синтезировать текст и вернуть raw audio."""
        await self._ws.send_json({
            "event": "text.chunk",
            "text": text,
            "flush": flush,
        })
        await self._ws.send_json({"event": "text.end"})
        return await self._collect_audio()

    async def synthesize_stream(self, text_chunks: list[str]) -> bytes:
        """Синтезировать текст, поступающий порциями (стриминг от LLM)."""
        for chunk in text_chunks:
            await self._ws.send_json({"event": "text.chunk", "text": chunk})
            await asyncio.sleep(0.05)

        await self._ws.send_json({"event": "text.end"})
        return await self._collect_audio()

    async def interrupt(self) -> int:
        """Прервать генерацию. Возвращает chunks_sent.

        Вычитывает все in-flight audio.chunk до generation.interrupted.
        """
        await self._ws.send_json({"event": "interrupt"})

        while True:
            msg = await asyncio.wait_for(self._ws.receive(), timeout=10)
            data = json.loads(msg.data)

            if data["event"] == "generation.interrupted":
                return data["chunks_sent"]
            elif data["event"] == "audio.chunk":
                continue  # drain — не воспроизводим
            elif data["event"] == "error":
                raise GenVoiceTTSError(data["code"], data["message"])

    async def _collect_audio(self, timeout: float = 60.0) -> bytes:
        """Собрать audio.chunk до generation.complete."""
        audio_parts: list[bytes] = []
        deadline = asyncio.get_event_loop().time() + timeout

        while True:
            remaining = deadline - asyncio.get_event_loop().time()
            if remaining <= 0:
                raise TimeoutError("Timeout waiting for generation.complete")

            msg = await asyncio.wait_for(self._ws.receive(), timeout=remaining)

            if msg.type in (aiohttp.WSMsgType.CLOSED, aiohttp.WSMsgType.ERROR):
                raise ConnectionError("WebSocket closed unexpectedly")

            data = json.loads(msg.data)

            if data["event"] == "audio.chunk":
                audio_parts.append(base64.b64decode(data["audio"]))
            elif data["event"] == "generation.complete":
                break
            elif data["event"] == "error":
                raise GenVoiceTTSError(data["code"], data["message"])

        return b"".join(audio_parts)

    def save_wav(self, audio_data: bytes, filepath: str | Path) -> None:
        """Сохранить raw audio в WAV-файл."""
        params = self.FORMAT_PARAMS[self.output_format]
        with wave.open(str(filepath), "wb") as wf:
            wf.setnchannels(1)
            wf.setsampwidth(params["sample_width"])
            wf.setframerate(params["sample_rate"])
            wf.writeframes(audio_data)

    async def close(self) -> None:
        """Завершить сессию."""
        if self._ws and not self._ws.closed:
            await self._ws.send_json({"event": "session.end"})
            await self._ws.close()
        if self._session:
            await self._session.close()

Сценарий 1: простой синтез

python

async def main():
    async with GenVoiceRealtimeClient(
        api_key="sk_live_YOUR_API_KEY",
        voice_id="9efc2a63-911e-4e19-9d5e-01b0640fc4e6",
    ) as client:
        audio = await client.synthesize("Привет! Это тест синтеза речи.")
        client.save_wav(audio, "output.wav")
        print(f"Saved output.wav ({len(audio)} bytes)")

asyncio.run(main())

Сценарий 2: потоковый синтез (имитация LLM)

Текст поступает токен за токеном. Буфер сервера автоматически определяет оптимальные границы для синтеза.

python

async def llm_streaming_example():
    async with GenVoiceRealtimeClient(
        api_key="sk_live_YOUR_API_KEY",
        voice_id="9efc2a63-911e-4e19-9d5e-01b0640fc4e6",
    ) as client:
        # Имитация потокового получения от LLM
        tokens = [
            "Добрый ", "день! ", "Сегодня ", "мы ", "рассмотрим ",
            "основные ", "возможности ", "нашей ", "платформы. ",
            "Начнём ", "с ", "самого ", "простого ", "примера.",
        ]

        for token in tokens:
            await client._ws.send_json({
                "event": "text.chunk",
                "text": token,
            })
            await asyncio.sleep(0.05)  # Имитация задержки LLM

        await client._ws.send_json({"event": "text.end"})
        audio = await client._collect_audio()
        client.save_wav(audio, "llm_output.wav")

asyncio.run(llm_streaming_example())

Сценарий 3: прерывание генерации

Типичный сценарий для голосовых ботов: пользователь начал говорить — нужно прервать текущий ответ бота.

python

async def interrupt_example():
    async with GenVoiceRealtimeClient(
        api_key="sk_live_YOUR_API_KEY",
        voice_id="9efc2a63-911e-4e19-9d5e-01b0640fc4e6",
    ) as client:
        # Начинаем длинную генерацию
        long_text = (
            "Это очень длинный текст, который будет генерироваться "
            "несколько секунд. Мы прервём его на полпути, имитируя "
            "ситуацию, когда пользователь начинает говорить."
        )
        await client._ws.send_json({
            "event": "text.chunk",
            "text": long_text,
            "flush": True,
        })
        await client._ws.send_json({"event": "text.end"})

        # Собираем первые несколько чанков
        played_audio: list[bytes] = []
        for _ in range(3):
            msg = await asyncio.wait_for(client._ws.receive(), timeout=30)
            data = json.loads(msg.data)
            if data["event"] == "audio.chunk":
                played_audio.append(base64.b64decode(data["audio"]))
            elif data["event"] == "generation.complete":
                break

        # Пользователь заговорил — прерываем
        chunks_sent = await client.interrupt()
        print(f"Interrupted after {chunks_sent} chunks sent by server")
        print(f"We played {len(played_audio)} chunks")

        # Сессия в READY — можно начать новую генерацию
        audio = await client.synthesize("Понял, слушаю вас.")
        client.save_wav(audio, "after_interrupt.wav")

asyncio.run(interrupt_example())

Напоминание

Метод interrupt() вычитывает все in-flight audio.chunk до получения generation.interrupted. Не пытайтесь обрабатывать interrupt самостоятельно, не дочитав канал — это приведёт к рассинхронизации.

Сценарий 4: несколько генераций в одной сессии

После generation.complete сессия возвращается в READY. Переподключение не нужно.

python

async def multi_generation_example():
    async with GenVoiceRealtimeClient(
        api_key="sk_live_YOUR_API_KEY",
        voice_id="9efc2a63-911e-4e19-9d5e-01b0640fc4e6",
    ) as client:
        phrases = [
            "Привет! Чем могу помочь?",
            "Конечно, расскажу подробнее.",
            "Есть ещё вопросы?",
        ]

        for i, phrase in enumerate(phrases):
            audio = await client.synthesize(phrase)
            client.save_wav(audio, f"phrase_{i + 1}.wav")
            print(f"phrase_{i + 1}.wav — {len(audio)} bytes")

asyncio.run(multi_generation_example())

Сценарий 5: keepalive (фоновый ping)

Для длительных соединений (ожидание ввода пользователя) отправляйте ping каждые 15 секунд:

python

async def keepalive_loop(ws, interval: float = 15.0):
    """Фоновая задача keepalive."""
    try:
        while not ws.closed:
            await ws.send_json({"event": "ping"})
            await asyncio.sleep(interval)
    except (ConnectionError, asyncio.CancelledError):
        pass


async def bot_with_keepalive():
    async with GenVoiceRealtimeClient(
        api_key="sk_live_YOUR_API_KEY",
        voice_id="9efc2a63-911e-4e19-9d5e-01b0640fc4e6",
    ) as client:
        # Запускаем keepalive в фоне
        ping_task = asyncio.create_task(keepalive_loop(client._ws))

        try:
            # Бот ждёт событий и синтезирует ответы
            audio = await client.synthesize("Привет! Жду ваш вопрос.")
            # ... ожидание ввода пользователя ...
            audio = await client.synthesize("Спасибо за вопрос!")
        finally:
            ping_task.cancel()

asyncio.run(bot_with_keepalive())

Получение voice_id

Для получения списка доступных голосов используйте REST API:

python

import aiohttp

async def list_voices(api_key: str) -> list[dict]:
    headers = {"Authorization": f"Bearer {api_key}"}
    async with aiohttp.ClientSession() as session:
        async with session.get(
            "https://api.genvoice.ru/v1/api/voices/public",
            headers=headers,
        ) as resp:
            resp.raise_for_status()
            return await resp.json()

Используйте поле id из ответа как voice_id при подключении к Realtime API.

Для быстрого старта

В примерах на этой странице используется голос Кира (9efc2a63-911e-4e19-9d5e-01b0640fc4e6) — публичный женский голос, доступный на всех тарифах.

Подробнее: API Reference — GET /api/voices/public

Сохранение аудио в WAV

Аудио приходит в raw-формате (без заголовков). Для сохранения в WAV:

Декодируйте base64 из каждого audio.chunk
Соедините все чанки по порядку sequence
Запишите WAV с правильными параметрами

Формат	Channels	Sample width	Frame rate
`pcm_24000`	1 (моно)	2 байта	24 000
`pcm_16000`	1 (моно)	2 байта	16 000
`pcm_8000`	1 (моно)	2 байта	8 000
`ulaw_8000`	1 (моно)	1 байт	8 000
`alaw_8000`	1 (моно)	1 байт	8 000

Для воспроизведения в реальном времени

Если вы воспроизводите аудио на лету (не сохраняете в файл), передавайте декодированные bytes напрямую в аудио-буфер плеера. WAV-заголовок не нужен.

Примеры интеграции ​

Быстрый тест с wscat ​

Python-клиент (aiohttp) ​

Установка ​

Полный клиент ​

Сценарий 1: простой синтез ​

Сценарий 2: потоковый синтез (имитация LLM) ​

Сценарий 3: прерывание генерации ​

Сценарий 4: несколько генераций в одной сессии ​

Сценарий 5: keepalive (фоновый ping) ​

Получение voice_id ​

Сохранение аудио в WAV ​

Примеры интеграции

Быстрый тест с wscat

Python-клиент (aiohttp)

Установка

Полный клиент

Сценарий 1: простой синтез

Сценарий 2: потоковый синтез (имитация LLM)

Сценарий 3: прерывание генерации

Сценарий 4: несколько генераций в одной сессии

Сценарий 5: keepalive (фоновый ping)

Получение voice_id

Сохранение аудио в WAV