Интеграция OpenAI Realtime API для голосового AI

Интеграция OpenAI Realtime API для голосового AI

OpenAI Realtime API (октябрь 2024) — WebSocket API для создания голосовых ассистентов с задержкой 200–500 мс. Убирает необходимость строить отдельный pipeline STT+LLM+TTS — всё в одном соединении. Поддерживает прерывание ответа пользователем.

Ключевые возможности

• Нативный voice-to-voice без промежуточных транскрипций
• Server VAD: автоматическое определение начала/конца речи
• Interruption handling: пользователь может перебить ответ
• Function calling в голосовом режиме
• Поддержка text+audio одновременно

WebSocket интеграция

import asyncio
import json
import websockets
import base64

async def voice_assistant():
    url = "wss://api.openai.com/v1/realtime?model=gpt-4o-realtime-preview-2024-12-17"
    headers = {
        "Authorization": f"Bearer {OPENAI_API_KEY}",
        "OpenAI-Beta": "realtime=v1"
    }

    async with websockets.connect(url, extra_headers=headers) as ws:
        # Инициализируем сессию
        await ws.send(json.dumps({
            "type": "session.update",
            "session": {
                "modalities": ["text", "audio"],
                "instructions": "Ты полезный голосовой ассистент. Отвечай по-русски, кратко.",
                "voice": "alloy",  # alloy | echo | fable | onyx | nova | shimmer
                "input_audio_format": "pcm16",
                "output_audio_format": "pcm16",
                "input_audio_transcription": {"model": "whisper-1"},
                "turn_detection": {
                    "type": "server_vad",
                    "threshold": 0.5,
                    "prefix_padding_ms": 300,
                    "silence_duration_ms": 700
                }
            }
        }))

        async def send_audio(audio_stream):
            async for chunk in audio_stream:
                encoded = base64.b64encode(chunk).decode()
                await ws.send(json.dumps({
                    "type": "input_audio_buffer.append",
                    "audio": encoded
                }))

        async def receive_responses():
            audio_buffer = bytearray()
            async for message in ws:
                event = json.loads(message)

                if event["type"] == "response.audio.delta":
                    audio_data = base64.b64decode(event["delta"])
                    audio_buffer.extend(audio_data)
                    # Воспроизводим чанки по мере поступления

                elif event["type"] == "response.audio.done":
                    # Вся аудиодорожка получена
                    pass

                elif event["type"] == "conversation.item.input_audio_transcription.completed":
                    print(f"User: {event['transcript']}")

        await asyncio.gather(send_audio(get_microphone_stream()),
                             receive_responses())

Function calling в голосовом режиме

tools = [{
    "type": "function",
    "name": "get_order_status",
    "description": "Получить статус заказа по номеру",
    "parameters": {
        "type": "object",
        "properties": {
            "order_id": {"type": "string", "description": "Номер заказа"}
        },
        "required": ["order_id"]
    }
}]

await ws.send(json.dumps({
    "type": "session.update",
    "session": {"tools": tools, "tool_choice": "auto"}
}))

Стоимость

• Input audio: $0.10/мин
• Output audio: $0.20/мин
• Input text: $5/1M tokens
• Output text: $20/1M tokens

Сроки: базовая интеграция — 2–3 дня. Production с function calling и бизнес-логикой — 1–2 недели.