GPT Live 1

# 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-live-1"
CHUNK_FRAMES = 4800  # 200 ms at 24 kHz


async def append(ws, chunk):
    await ws.send(json.dumps({
        "type": "session.input_audio.append",
        "audio": base64.b64encode(chunk).decode("ascii"),
    }))
    await asyncio.sleep(len(chunk) / 48000)


async def send_audio(ws, path):
    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(CHUNK_FRAMES):
            await append(ws, chunk)
    # Keep the stream continuous while the model answers.
    for _ in range(25):
        await append(ws, bytes(CHUNK_FRAMES * 2))
    await ws.send(json.dumps({"type": "session.close"}))


async def receive_events(ws):
    with open("reply.pcm", "wb") as output:
        async for raw in ws:
            event = json.loads(raw)
            if event["type"] == "session.output_audio.delta":
                output.write(base64.b64decode(event["delta"]))
            elif event["type"] == "session.output_transcript.delta":
                print(event["delta"], end="", flush=True)
            elif event["type"] == "error":
                raise RuntimeError(event["error"])
            elif event["type"] == "session.closed":
                print("\nReason:", event["reason"], "Seconds:", event["usage"]["seconds"])
                return
    raise RuntimeError("Connection ended before session.closed")


async def main(path):
    headers = {"Authorization": f"Bearer {os.environ['MODELLIX_API_KEY']}"}
    async with websockets.connect(URL, additional_headers=headers, max_size=16 * 1024 * 1024) as ws:
        # 1. The first frame must be session.start.
        await ws.send(json.dumps({
            "type": "session.start",
            "session": {
                "instructions": "Answer briefly.",
                "audio": {"output": {"voice": "marin"}},
            },
        }))
        started = json.loads(await ws.recv())
        if started["type"] != "session.started":
            raise RuntimeError(started)

        # 2. Stream audio and read events at the same time.
        await asyncio.gather(send_audio(ws, path), receive_events(ws))


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

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