GPT Live Transcribe

# pip install websockets
import asyncio
import base64
import json
import os
import websockets

# Realtime transcription is a WebSocket session: model must be in the handshake URL.
URL = "wss://aihubmix.com/v1/realtime?intent=transcription&model=gpt-live-transcribe"

async def main():
    # websockets >= 13 uses additional_headers; older versions use extra_headers
    async with websockets.connect(
        URL, additional_headers={"Authorization": "Bearer " + os.environ["AIHUBMIX_API_KEY"]}
    ) as ws:
        # 1) Configure the transcription session (first frame; server VAD auto-segments speech as you stream)
        await ws.send(json.dumps({
          "type": "session.update",
          "session": {
            "type": "transcription",
            "audio": {
              "input": {
                "format": {
                  "type": "audio/pcm",
                  "rate": 24000
                },
                "transcription": {
                  "model": "gpt-live-transcribe"
                },
                "turn_detection": {
                  "type": "server_vad"
                }
              }
            }
          }
        }))

        flushed = asyncio.Event()  # set once the whole stream is sent + flushed

        # 2) Stream raw PCM16 / 24kHz / mono audio in ~100ms chunks
        async def send_audio():
            with open("audio_pcm16_24k.raw", "rb") as f:
                pcm = f.read()
            chunk = 24000 * 2 * 100 // 1000  # 100ms of 16-bit mono samples
            for i in range(0, len(pcm), chunk):
                await ws.send(json.dumps({
                    "type": "input_audio_buffer.append",
                    "audio": base64.b64encode(pcm[i:i + chunk]).decode(),
                }))
                await asyncio.sleep(0.1)  # simulate realtime pacing
            # Server VAD segments on pauses; flush any trailing audio at end of stream
            await ws.send(json.dumps({"type": "input_audio_buffer.commit"}))
            flushed.set()

        asyncio.create_task(send_audio())

        # 3) Receive events: deltas stream live; each pause finalizes a segment (server VAD)
        async for msg in ws:
            evt = json.loads(msg)
            etype = evt.get("type", "")
            if etype.endswith("transcription.delta"):
                print(evt.get("delta", ""), end="", flush=True)
            elif etype.endswith("transcription.completed"):
                print("\n[segment]", evt.get("transcript", ""))
                if flushed.is_set():  # last segment after the final flush → done
                    break
            elif etype == "error":
                print("\n[error]", evt.get("error"))
                break

asyncio.run(main())