Realtime transcription

1
await socket.send(json.dumps({
2
    "type": "input_audio_buffer.append",
3
    "audio": base64.b64encode(pcm_chunk).decode("ascii"),
4
}))

1
"""WebSocket transcription example. Requires Python 3.11+ and websockets 14–15."""
2
import asyncio
3
import base64
4
import json
5
import os
6
7
from websockets.asyncio.client import connect
8
9
URL = "wss://api.narilabs.com/v1/realtime?intent=transcription"
10
11
12
async def pcm_file(path):
13
    with open(path, "rb") as audio:
14
        while chunk := audio.read(3200):  # 100 ms at 16 kHz PCM16 mono
15
            yield chunk
16
            await asyncio.sleep(len(chunk) / 32000)
17
18
19
async def transcribe(audio_source, *, turn_detection=None, on_event=print):
20
    """Source yields PCM bytes, or None to manually commit an utterance."""
21
    async with connect(
22
        URL,
23
        additional_headers={"Authorization": f"Bearer {os.environ['NARI_API_KEY']}"},
24
    ) as socket:
25
        async def receive():
26
            event = json.loads(await socket.recv())
27
            if event["type"] == "error":
28
                raise RuntimeError(event["error"])
29
            return event
30
31
        assert (await receive())["type"] == "session.created"
32
        await socket.send(json.dumps({
33
            "type": "session.configure",
34
            "session": {"model": "qwen3-asr:free", "turn_detection": turn_detection},
35
        }))
36
        assert (await receive())["type"] == "session.configured"
37
38
        async def send_audio():
39
            commit_number = 0
40
            async for chunk in audio_source:
41
                if chunk is None:
42
                    commit_number += 1
43
                    event = {"type": "input_audio_buffer.commit",
44
                             "event_id": f"commit_{commit_number}"}
45
                else:
46
                    if not chunk or len(chunk) % 2:
47
                        raise ValueError("Send nonempty chunks of whole PCM16 samples")
48
                    event = {"type": "input_audio_buffer.append",
49
                             "audio": base64.b64encode(chunk).decode("ascii")}
50
                await socket.send(json.dumps(event))
51
            # Flush any remaining audio, including when using VAD.
52
            await socket.send(json.dumps({
53
                "type": "input_audio_buffer.commit", "event_id": "end_of_input",
54
            }))
55
56
        async def read_results():
57
            pending = set()
58
            end_acknowledged = False
59
            while True:
60
                event = await receive()
61
                kind = event["type"]
62
                if kind == "input_audio_buffer.committed":
63
                    pending.add(event["item_id"])
64
                elif kind == "transcript.completed":
65
                    pending.discard(event["item_id"])
66
                if kind in ("input_audio_buffer.committed", "input_audio_buffer.commit_empty"):
67
                    if event.get("client_event_id") == "end_of_input":
68
                        end_acknowledged = True
69
                on_event(event)
70
                if end_acknowledged and not pending:
71
                    return
72
73
        async with asyncio.TaskGroup() as tasks:
74
            tasks.create_task(send_audio())
75
            tasks.create_task(read_results())
76
77
78
if __name__ == "__main__":
79
    asyncio.run(transcribe(pcm_file("hello.pcm")))