GPT Realtime Whisper

# pip install "websockets>=14"
import asyncio
import base64
import json
import os
import sys
import wave

import websockets

URL = "wss://api.modellix.ai/api/v1/live/openai/gpt-realtime-whisper"


async def main(path):
    headers = {"Authorization": f"Bearer {os.environ['MODELLIX_API_KEY']}"}
    async with websockets.connect(URL, additional_headers=headers) as ws:
        # 1. Wait for session.created, then configure transcription.
        created = json.loads(await ws.recv())
        if created["type"] != "session.created":
            raise RuntimeError(created)
        await ws.send(json.dumps({
            "type": "session.update",
            "session": {
                "type": "transcription",
                "audio": {
                    "input": {
                        "format": {"type": "audio/pcm", "rate": 24000},
                        "transcription": {"model": "gpt-realtime-whisper"},
                        "turn_detection": None,
                    }
                },
            },
        }))

        # 2. Stream audio in real time, then commit the turn.
        with wave.open(path, "rb") as audio:
            if (audio.getnchannels(), audio.getsampwidth(), audio.getframerate()) != (1, 2, 24000):
                raise ValueError("Use mono, 16-bit PCM WAV audio at 24 kHz")
            while chunk := audio.readframes(4800):  # 200 ms at 24 kHz
                await ws.send(json.dumps({
                    "type": "input_audio_buffer.append",
                    "audio": base64.b64encode(chunk).decode("ascii"),
                }))
                await asyncio.sleep(len(chunk) / 48000)
        await ws.send(json.dumps({"type": "input_audio_buffer.commit"}))

        # 3. Match events to the committed item by item_id.
        item_id = None
        async for raw in ws:
            event = json.loads(raw)
            kind = event["type"]
            if kind == "input_audio_buffer.committed":
                item_id = event["item_id"]
            elif kind == "conversation.item.input_audio_transcription.delta":
                print("Partial:", event["delta"])
            elif kind == "conversation.item.input_audio_transcription.completed":
                if event["item_id"] == item_id:
                    print("Final:", event["transcript"])
                    print("Usage:", event["usage"])
                    return
            elif kind in ("conversation.item.input_audio_transcription.failed", "error"):
                raise RuntimeError(event)
    raise RuntimeError("Connection ended before the transcript completed")


if len(sys.argv) != 2:
    raise SystemExit("Usage: python whisper.py speech.wav")

asyncio.run(main(sys.argv[1]))