transcribing and translating in realtime with ai

# 1. Get a bearer token for Azure OpenAI.
credential = DefaultAzureCredential()
token = await credential.get_token("https://cognitiveservices.azure.com/.default")

# 2. Open the realtime websocket.
ws = await connect(
    URL,
    additional_headers={"Authorization": f"Bearer {token.token}"},
    max_size=None,
)

# 3. Configure the transcription session. gpt-realtime-whisper does not support turn_detection, so we find utterance boundaries ourselves and commit the buffer manually.
await ws.send(
    json.dumps(
        {
            "type": "session.update",
            "session": {
                "type": "transcription",
                "audio": {
                    "input": {
                        "format": {"type": "audio/pcm", "rate": SAMPLE_RATE},
                        "transcription": {"model": DEPLOYMENT},
                    },
                },
            },
        }
    )
)

# 4. Mic capture. The sounddevice callback runs on its own thread, so it hands raw PCM16 bytes to asyncio through this queue.
loop = asyncio.get_running_loop()
mic_queue = asyncio.Queue()

def on_audio(indata, frames, time, status):
    loop.call_soon_threadsafe(mic_queue.put_nowait, bytes(indata))

stream = sd.RawInputStream(
    samplerate=SAMPLE_RATE,
    blocksize=CHUNK_SAMPLES,
    channels=1,
    dtype="int16",
    callback=on_audio,
)
stream.start()
print("Listening... press Ctrl+C to stop.\n")

# 5. Pump mic audio up to Azure, and commit the buffer once the talking stops.
async def send_audio():
    buffered_chunks = 0
    silent_chunks = 0
    heard_speech = False

    while True:
        chunk = await mic_queue.get()
        await ws.send(
            json.dumps(
                {
                    "type": "input_audio_buffer.append",
                    "audio": base64.b64encode(chunk).decode("ascii"),
                }
            )
        )
        buffered_chunks += 1

        # Loudness of this chunk, 0.0 (silence) to 1.0 (clipping).
        samples = struct.unpack(f"<{len(chunk) // 2}h", chunk)
        rms = math.sqrt(sum(s * s for s in samples) / len(samples)) / 32768

        if rms >= SPEECH_RMS:
            heard_speech = True
            silent_chunks = 0
        else:
            silent_chunks += 1

        if (
            heard_speech
            and silent_chunks >= SILENT_CHUNKS_TO_COMMIT
            and buffered_chunks >= MIN_CHUNKS_TO_COMMIT
        ):
            await ws.send(json.dumps({"type": "input_audio_buffer.commit"}))
            buffered_chunks = 0
            silent_chunks = 0
            heard_speech = False

# 6. Print transcripts as they come back.
async def receive_transcripts():
    async for message in ws:
        event = json.loads(message)
        kind = event["type"]

        if kind == "conversation.item.input_audio_transcription.delta":
            print(event["delta"], end="", flush=True)
        elif kind == "conversation.item.input_audio_transcription.completed":
            print(f"\n--- {event['transcript']}\n")
        elif kind == "error":
            # filter out "buffer too small". just means we committed when Azure had nothing pending.
            message = event["error"]["message"]
            if "buffer too small" not in message.lower():
                print(f"\n[error] {message}\n")

sender = asyncio.create_task(send_audio())
receiver = asyncio.create_task(receive_transcripts())

try:
    await asyncio.gather(sender, receiver)
finally:
    stream.stop()
    stream.close()
    await ws.close()
    await credential.close()

uv run gpt_realtime_whisper.py

# 1. Get a bearer token for Azure OpenAI.
credential = DefaultAzureCredential()
token = await credential.get_token("https://cognitiveservices.azure.com/.default")

# 2. Open the realtime translations websocket. The openai-alpha header opts into the translation API.
ws = await connect(
    URL,
    additional_headers={
        "Authorization": f"Bearer {token.token}",
        "openai-alpha": "translation=v1",
    },
    max_size=None,
)

# 3. Configure the translation session. The translation endpoint finds utterance boundaries itself, so there is no turn detection or audio commit to configure.
await ws.send(
    json.dumps(
        {
            "type": "session.update",
            "session": {
                "audio": {
                    "input": {
                        "transcription": {"model": DEPLOYMENT}, # remove the entire "input" thing if you don't want input transcriptions (note that if you remove this, you will need to modify multiple pieces of code from this sample to not handle input transcriptions logic)
                    },
                    "output": {
                        "language": TARGET_LANGUAGE,
                    },
                },
            },
        }
    )
)

# 4. Mic capture. The sounddevice callback runs on its own thread, so it hands raw PCM16 bytes to asyncio through this queue.
loop = asyncio.get_running_loop()
mic_queue = asyncio.Queue()

def on_audio(indata, frames, time, status):
    loop.call_soon_threadsafe(mic_queue.put_nowait, bytes(indata))

stream = sd.RawInputStream(
    samplerate=SAMPLE_RATE,
    blocksize=CHUNK_SAMPLES,
    channels=1,
    dtype="int16",
    callback=on_audio,
)
stream.start()
print("Listening... press Ctrl+C to stop.\n")

# 5. Pump mic audio up to Azure.
async def send_audio():
    while True:
        chunk = await mic_queue.get()
        await ws.send(
            json.dumps(
                {
                    "type": "session.input_audio_buffer.append",
                    "audio": base64.b64encode(chunk).decode("ascii"),
                }
            )
        )

# 6. Print the translation live. Keep the source transcript for the summary at exit.
transcript = []
pending = ""  # this is used to track input transcript until the end

async def receive_translations():
    nonlocal pending

    async for message in ws:
        event = json.loads(message)
        kind = event["type"]

        if kind == "session.output_transcript.delta":
            print(event["delta"], end="", flush=True)
        elif kind in (
            "session.output_transcript.completed",
            "session.output_transcript.done",
        ):
            print()
        elif kind == "session.input_transcript.delta":
            pending += event["delta"]
        elif kind in (
            "session.input_transcript.completed",
            "session.input_transcript.done",
        ):
            transcript.append(event.get("transcript") or pending)
            pending = ""
        elif kind == "error":
            print(f"\n[error] {event['error']['message']}\n")

sender = asyncio.create_task(send_audio())
receiver = asyncio.create_task(receive_translations())

try:
    await asyncio.gather(sender, receiver)
finally:
    stream.stop()
    stream.close()
    await ws.close()
    await credential.close()

    if pending:
        transcript.append(pending)

    print("\n\n--- what you said ---")
    for line in transcript:
        print(line)

uv run gpt_realtime_translate.py