Создаём real-time голосового агента на WebSocket: потоковое распознавание речи (STT), инференс GPT-4o, синтез речи (TTS) и function calling — всё с задержкой менее 500 мс.
OpenAI Realtime API — это WebSocket-интерфейс, обеспечивающий двустороннюю потоковую передачу аудио между клиентом и моделью GPT-4o. В отличие от классического подхода (STT → текст → LLM → текст → TTS), Realtime API передаёт аудио напрямую модели, которая понимает речь, рассуждает и синтезирует ответ в одном сквозном проходе (end-to-end). Это сокращает задержку до 200–400 мс и сохраняет интонации, эмоции и невербальные сигналы.
# Сравнение подходов # Традиционный (latency ~2-5 сек): # Микрофон → Whisper STT → Текст → GPT-4 → Текст → ElevenLabs TTS → Динамик # Realtime API (latency ~200-400 мс): # Микрофон → PCM16 Audio → WebSocket → GPT-4o Realtime → PCM16 Audio → Динамик # ↕ # Function Calling # Ключевые возможности: # - Аудиоформат: PCM16, 24 кГц, моно, 16-bit # - VAD (Voice Activity Detection) на стороне сервера # - Прерывание ответа (interruption) — пользователь перебивает модель # - Function calling прямо из аудиосессии # - 6 голосов: alloy, ash, ballad, coral, echo, sage, shimmer, verse
Подключение к Realtime API выполняется через WebSocket с эфемерным ключом. Мы запрашиваем временный токен через REST API (модель создаёт сессию), затем открываем WebSocket-соединение с этим токеном. Важно: токен действителен 60 секунд только для установки соединения.
import asyncio import json import websockets from openai import AsyncOpenAI client = AsyncOpenAI(api_key="sk-...") async def get_ephemeral_token(model="gpt-4o-realtime-preview-2024-12-17"): """Получаем эфемерный токен для WebSocket-сессии""" # Важно: используем /v1/realtime/sessions REST-эндпоинт # В production берите токен через бэкенд, не expose API-ключ клиенту! session = await client.post("/v1/realtime/sessions", body={ "model": model, "voice": "alloy", "input_audio_format": "pcm16", "output_audio_format": "pcm16", "instructions": "Ты — голосовой ассистент интернет-магазина. Отвечай кратко.", "modalities": ["text", "audio"], "temperature": 0.8, "turn_detection": { "type": "server_vad", "threshold": 0.5, "prefix_padding_ms": 300, "silence_duration_ms": 500 } }) return session["client_secret"]["value"] async def connect_realtime(): token = await get_ephemeral_token() ws_url = f"wss://api.openai.com/v1/realtime?model=gpt-4o-realtime-preview-2024-12-17" headers = {"Authorization": f"Bearer {token}", "OpenAI-Beta": "realtime=v1"} ws = await websockets.connect(ws_url, extra_headers=headers) print("✅ WebSocket connected to Realtime API") return ws
Основной цикл агента: захватываем аудио с микрофона, кодируем в PCM16 24 кГц, кодируем в base64 и отправляем через WebSocket событием input_audio_buffer.append. Когда VAD на сервере детектирует конец фразы, сервер автоматически начинает инференс. Мы слушаем события response.audio.delta и накапливаем аудио для воспроизведения.
import base64 import numpy as np import pyaudio # Константы аудио SAMPLE_RATE = 24000 CHUNK_SIZE = 2400 # 100 мс при 24 кГц CHANNELS = 1 async def send_audio_loop(ws, mic_stream): """Непрерывно читает микрофон и отправляет аудио в Realtime API""" while True: data = mic_stream.read(CHUNK_SIZE, exception_on_overflow=False) audio_b64 = base64.b64encode(data).decode("utf-8") event = { "type": "input_audio_buffer.append", "audio": audio_b64 } await ws.send(json.dumps(event)) await asyncio.sleep(0.08) # ~12 fps async def receive_loop(ws, audio_player): """Принимает ответы от Realtime API и воспроизводит аудио""" audio_buffer = bytearray() async for message in ws: event = json.loads(message) etype = event.get("type") if etype == "response.audio.delta": chunk = base64.b64decode(event["delta"]) audio_buffer.extend(chunk) audio_player.write(chunk) # Потоковое воспроизведение elif etype == "response.audio.done": print(f"🎵 Аудио получено: {len(audio_buffer)} байт") audio_buffer = bytearray() # Сброс elif etype == "response.text.delta": print(event["delta"], end="", flush=True) # Транскрипт elif etype == "input_audio_buffer.speech_started": print("\n🎤 Говорит пользователь...") elif etype == "input_audio_buffer.speech_stopped": print("\n⏳ Обработка...")
Realtime API поддерживает вызов функций прямо во время аудиосессии. Мы регистрируем инструменты через session.update, и когда модель решает, что нужен вызов функции, она отправляет событие response.function_call_arguments.done. Мы выполняем функцию и отправляем результат обратно через conversation.item.create.
# Регистрируем инструменты при старте сессии async def register_tools(ws): """Отправляем доступные функции модели через session.update""" tools_payload = { "type": "session.update", "session": { "tools": [ { "type": "function", "name": "search_products", "description": "Поиск товаров в каталоге магазина", "parameters": { "type": "object", "properties": { "query": {"type": "string", "description": "Поисковый запрос"}, "max_price": {"type": "number", "description": "Максимальная цена в рублях"} }, "required": ["query"] } }, { "type": "function", "name": "check_order_status", "description": "Проверить статус заказа по номеру", "parameters": { "type": "object", "properties": { "order_id": {"type": "string", "description": "Номер заказа, например ORD-12345"} }, "required": ["order_id"] } } ], "tool_choice": "auto" } } await ws.send(json.dumps(tools_payload)) print("🔧 Инструменты зарегистрированы") # Обработчик function call в receive_loop elif etype == "response.function_call_arguments.done": call_id = event["call_id"] name = event["name"] args = json.loads(event["arguments"]) print(f"\n🔨 Вызов: {name}({args})") # Выполняем функцию result = await execute_function(name, args) # Отправляем результат обратно в сессию await ws.send(json.dumps({ "type": "conversation.item.create", "item": { "type": "function_call_output", "call_id": call_id, "output": json.dumps(result) } }))
Для браузерного клиента используем MediaStream API (getUserMedia) для захвата микрофона и AudioContext для воспроизведения ответа. WebSocket-соединение проксируется через бэкенд (нельзя подключаться к OpenAI напрямую из браузера — нужно скрывать ключ).
// Клиентский JavaScript для Realtime API const SAMPLE_RATE = 24000; // 1. Захват микрофона с ресемплингом в 24 кГц async function startMicrophone() { const stream = await navigator.mediaDevices.getUserMedia({ audio: { sampleRate: SAMPLE_RATE, channelCount: 1, echoCancellation: true } }); const ctx = new AudioContext({ sampleRate: SAMPLE_RATE }); const source = ctx.createMediaStreamSource(stream); const processor = ctx.createScriptProcessor(4096, 1, 1); processor.onaudioprocess = (e) => { const input = e.inputBuffer.getChannelData(0); const pcm16 = floatTo16BitPCM(input); const base64 = arrayBufferToBase64(pcm16.buffer); ws.send(JSON.stringify({ type: "input_audio_buffer.append", audio: base64 })); }; source.connect(processor); processor.connect(ctx.destination); } // 2. Конвертация Float32 → PCM16 function floatTo16BitPCM(float32) { const buf = new Int16Array(float32.length); for (let i = 0; i < float32.length; i++) { const s = Math.max(-1, Math.min(1, float32[i])); buf[i] = s < 0 ? s * 0x8000 : s * 0x7FFF; } return buf; } // 3. Воспроизведение ответа — накопление чанков const audioCtx = new AudioContext({ sampleRate: SAMPLE_RATE }); let nextPlayTime = 0; ws.onmessage = (event) => { const msg = JSON.parse(event.data); if (msg.type === "response.audio.delta") { const pcm16 = base64ToArrayBuffer(msg.delta); const float32 = pcm16ToFloat32(new Int16Array(pcm16)); const buffer = audioCtx.createBuffer(1, float32.length, SAMPLE_RATE); buffer.getChannelData(0).set(float32); const source = audioCtx.createBufferSource(); source.buffer = buffer; source.connect(audioCtx.destination); const now = audioCtx.currentTime; nextPlayTime = Math.max(now, nextPlayTime); source.start(nextPlayTime); nextPlayTime += buffer.duration; } };
Когда пользователь перебивает агента, сервер отправляет событие response.cancel. Мы должны немедленно остановить воспроизведение текущего аудио и очистить буфер. Это критически важно для естественного диалога — без обработки прерываний агент звучит как автоответчик.
# Серверная часть — обработка прерываний class AudioPlaybackManager: """Управляет воспроизведением аудио с поддержкой прерываний""" def __init__(self): self.current_response_id = None self.audio_queue = asyncio.Queue() self.is_playing = False self.should_stop = False async def handle_interruption(self): """Немедленно останавливает воспроизведение при прерывании""" self.should_stop = True self.audio_queue = asyncio.Queue() # Очищаем очередь print("⏹️ Воспроизведение прервано пользователем") # В receive_loop добавляем обработку: elif etype == "input_audio_buffer.speech_started": if playback.is_playing: await playback.handle_interruption() # Уведомляем сервер об отмене текущего ответа await ws.send(json.dumps({"type": "response.cancel"})) elif etype == "response.audio_transcript.done": transcript = event["transcript"] print(f"\n🤖 Агент: {transcript}")
В production мы не отдаём API-ключ клиенту. Бэкенд на FastAPI принимает WebSocket от браузера, проксирует аудио в OpenAI Realtime API и возвращает ответ. Добавляем мониторинг: логируем длительность сессий, количество вызовов функций и ошибки.
from fastapi import FastAPI, WebSocket, WebSocketDisconnect import time app = FastAPI() @app.websocket("/ws/realtime") async def realtime_proxy(client_ws: WebSocket): await client_ws.accept() session_start = time.time() tool_calls_count = 0 try: # 1. Получаем токен и подключаемся к OpenAI token = await get_ephemeral_token() async with websockets.connect( f"wss://api.openai.com/v1/realtime?model=gpt-4o-realtime-preview-2024-12-17", extra_headers={"Authorization": f"Bearer {token}", "OpenAI-Beta": "realtime=v1"} ) as openai_ws: # 2. Два направления прокси: client→OpenAI и OpenAI→client async def client_to_openai(): async for msg in client_ws.iter_text(): await openai_ws.send(msg) async def openai_to_client(): nonlocal tool_calls_count async for msg in openai_ws: event = json.loads(msg) if event.get("type") == "response.function_call_arguments.done": tool_calls_count += 1 # Здесь можно выполнить функцию и вернуть результат await client_ws.send_text(msg) await asyncio.gather(client_to_openai(), openai_to_client()) except WebSocketDisconnect: duration = time.time() - session_start print(f"📊 Сессия: {duration:.1f}с, вызовов функций: {tool_calls_count}") except Exception as e: print(f"❌ Ошибка сессии: {e}") await client_ws.send_text(json.dumps({"type": "error", "message": str(e)})) if __name__ == "__main__": import uvicorn uvicorn.run(app, host="0.0.0.0", port=8080)
Мы построили полноценного голосового агента на базе OpenAI Realtime API. Решение включает: установку WebSocket-соединения с эфемерным токеном, непрерывную отправку аудио с микрофона (PCM16 24 кГц), приём и потоковое воспроизведение ответов, регистрацию инструментов через session.update, обработку function call прямо из голосовой сессии, корректную обработку прерываний (interruption) и безопасный деплой через FastAPI WebSocket-прокси. Задержка «от конца фразы до начала ответа» составляет 200–400 мс — это уровень живого разговора. Следующий шаг — интеграция WebRTC для peer-to-peer аудио и добавление нескольких голосовых профилей для разных сценариев (поддержка, продажи, консультация).