Inverse Text Normalization (ITN)

1
import asyncio
2
import websockets
3
import json
4
from urllib.parse import urlencode
5
6
BASE_WS_URL = "wss://api.smallest.ai/waves/v1/stt/live?model=pulse"
7
params = {
8
    "language": "en",
9
    "encoding": "linear16",
10
    "sample_rate": "16000",
11
    "word_timestamps": "true",
12
    "itn_normalize": "true",
13
}
14
WS_URL = f"{BASE_WS_URL}&{urlencode(params)}"
15
16
API_KEY = "YOUR_API_KEY"
17
18
async def transcribe_file(audio_file: str):
19
    """Single-shot: stream a fixed audio buffer, then close the session.
20
    For multi-turn voice agents, use the Multi-turn voice agent example
21
    further down — it keeps the WebSocket open across user turns.
22
    """
23
    headers = {"Authorization": f"Bearer {API_KEY}"}
24
25
    async with websockets.connect(WS_URL, additional_headers=headers) as ws:
26
        # Stream the file in 4096-byte chunks
27
        with open(audio_file, "rb") as f:
28
            while chunk := f.read(4096):
29
                await ws.send(chunk)
30
31
        # All audio sent. For one-off transcription with no more audio coming,
32
        # send close_stream — it flushes, runs ITN over the full buffer, emits
33
        # is_final + is_last, and closes the WebSocket.
34
        await ws.send(json.dumps({"type": "close_stream"}))
35
36
        async for message in ws:
37
            data = json.loads(message)
38
            if data.get("is_final"):
39
                print(f"Final: {data['transcript']}")
40
                # With ITN: "the total is $25."
41
                # Without:  "the total is twenty five dollars."
42
            else:
43
                print(f"Interim: {data['transcript']}")
44
45
            if data.get("is_last"):
46
                break
47
48
asyncio.run(transcribe_file("audio.wav"))

1
params = {
2
    "language": "en",
3
    "encoding": "linear16",
4
    "sample_rate": "16000",
5
    "itn_normalize": "true",
6
    "finalize_on_words": "false",   # Disable internal word-count finalization
7
    "eou_timeout_ms": "600",        # Match your VAD silence threshold
8
    "word_timestamps": "true",
9
}
10
WS_URL = f"{BASE_WS_URL}&{urlencode(params)}"
11
12
async def run_voice_agent():
13
    headers = {"Authorization": f"Bearer {API_KEY}"}
14
15
    async with websockets.connect(WS_URL, additional_headers=headers) as ws:
16
        # Audio producer — stream mic frames as the user speaks across many turns
17
        async def stream_audio():
18
            while session_active:
19
                frame = await mic_queue.get()
20
                await ws.send(frame)
21
22
        # Per-turn finalizer — your VAD calls this when the user pauses
23
        async def on_user_end_of_turn():
24
            await ws.send(json.dumps({"type": "finalize"}))
25
            # Server emits is_final: true with the full ITN-normalized transcript;
26
            # the WS stays open so the user can speak again.
27
28
        # Consumer — receive transcripts (each turn yields one is_final: true frame)
29
        async def consume():
30
            async for message in ws:
31
                data = json.loads(message)
32
                if data.get("is_final"):
33
                    print(f"Turn final: {data['transcript']}")
34
                    # → hand to your LLM, generate the agent reply, speak via TTS,
35
                    #   and loop back to listening for the next user turn
36
                if data.get("is_last"):
37
                    break  # only fires after close_stream below
38
39
        # ... wire stream_audio / consume / VAD-driven on_user_end_of_turn together ...
40
41
        # End of call (user hung up, agent finished, etc.) — close the session
42
        await ws.send(json.dumps({"type": "close_stream"}))

1
async def transcribe_once(audio_file: str):
2
    headers = {"Authorization": f"Bearer {API_KEY}"}
3
4
    async with websockets.connect(WS_URL, additional_headers=headers) as ws:
5
        with open(audio_file, "rb") as f:
6
            while chunk := f.read(4096):
7
                await ws.send(chunk)
8
9
        # All audio sent → close_stream is the only message that emits is_last=true
10
        await ws.send(json.dumps({"type": "close_stream"}))
11
12
        async for message in ws:
13
            data = json.loads(message)
14
            if data.get("is_final"):
15
                print(f"Final: {data['transcript']}")
16
            if data.get("is_last"):
17
                break

1
params = {
2
    "language": "en",
3
    "encoding": "linear16",
4
    "sample_rate": "16000",
5
    "itn_normalize": "true",
6
    "redact_pii": "true",           # Redact names, SSN, emails, phone numbers
7
    "diarize": "true",              # Speaker diarization
8
    "word_timestamps": "true",
9
}