Transcribe live phone calls in real time with Twilio and OrbitalsAI

1
2
3
4
5
ORBITALSAI_API_KEY=your_orbitalsai_key_here
NGROK_AUTH_TOKEN=your_ngrok_token_here
TWILIO_ACCOUNT_SID=your_twilio_account_sid
TWILIO_API_KEY_SID=your_twilio_api_key_sid
TWILIO_API_SECRET=your_twilio_api_secret

1
pip install orbitalsai fastapi uvicorn python-dotenv

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
import os
import audioop
import threading
import time
from typing import Optional
 
from orbitalsai.streaming import StreamingClient, StreamingConfig, StreamingEventHandlers
from dotenv import load_dotenv
 
load_dotenv()
 
ORBITALSAI_API_KEY = os.getenv("ORBITALSAI_API_KEY")
 
# Twilio sends 8 kHz mu-law. OrbitalsAI expects 16 kHz PCM16.
TWILIO_SAMPLE_RATE = 8000
ORBITALS_SAMPLE_RATE = 16000
 
# 4000 samples at 16 kHz = 0.25 s per chunk sent to OrbitalsAI.
# Going smaller increases WebSocket overhead; going larger adds latency
# before you see the first partial transcript.
CHUNK_SAMPLES = 4000
 
MAX_CONNECT_RETRIES = 3
CONNECT_RETRY_DELAY = 1.0
 
 
class TwilioOrbitalsHandlers(StreamingEventHandlers):
    """Print partial and final transcripts to stdout."""
 
    def on_open(self, session_info: dict) -> None:
        print(f"OrbitalsAI session opened: {session_info.get('session_id')}")
        print(f"Language: {session_info.get('language')}")
 
    def on_transcript_partial(self, transcript: str) -> None:
        if transcript:
            print(f"\rPartial: {transcript}", end="", flush=True)
 
    def on_transcript_final(self, transcript: str, metadata: dict) -> None:
        if transcript:
            print(f"\nFinal: {transcript}")
            print(
                f"  Cost: \${metadata.get('cost', 0):.4f} | "
                f"Duration: {metadata.get('audio_seconds', 0):.1f}s"
            )
 
    def on_speech_start(self) -> None:
        print("\n[Speech detected]")
 
    def on_speech_end(self) -> None:
        print("\n[Silence detected]")
 
    def on_error(self, error: Exception) -> None:
        print(f"\nOrbitalsAI error: {error}")
 
    def on_close(self, code: Optional[int], reason: Optional[str]) -> None:
        print(f"\nOrbitalsAI session closed (code={code}, reason={reason})")
 
 
class TwilioOrbitalsTranscriber:
    """
    Converts Twilio media frames to OrbitalsAI streaming audio.
 
    Per-frame pipeline:
      1. 8-bit mu-law @ 8 kHz  →  PCM16 @ 8 kHz   (audioop.ulaw2lin)
      2. PCM16 @ 8 kHz         →  PCM16 @ 16 kHz   (audioop.ratecv)
      3. Buffer until 0.25 s of audio, then send_audio()
    """
 
    def __init__(self) -> None:
        if not ORBITALSAI_API_KEY:
            raise RuntimeError("ORBITALSAI_API_KEY is not set in .env")
 
        self._config = StreamingConfig(
            sample_rate=ORBITALS_SAMPLE_RATE,
            chunk_size=CHUNK_SAMPLES,
            interim_results=True,
        )
        self._client = StreamingClient(api_key=ORBITALSAI_API_KEY, config=self._config)
        self._handlers = TwilioOrbitalsHandlers()
        self._buffer = bytearray()
        self._lock = threading.Lock()
        self._min_send_bytes = CHUNK_SAMPLES * 2  # 2 bytes per PCM16 sample
 
        # audioop.ratecv carries filter state between calls.
        # Never reset this to None mid-call. Doing so produces a
        # pop or dropped syllables at the reset point.
        self._ratecv_state = None
 
    def connect(self) -> None:
        """Connect to OrbitalsAI with automatic retry on transient errors."""
        last_error: Optional[Exception] = None
 
        for attempt in range(1, MAX_CONNECT_RETRIES + 1):
            try:
                print(f"Connecting to OrbitalsAI (attempt {attempt})...")
                self._client.connect(self._handlers)
                print("OrbitalsAI connected.")
                return
            except Exception as exc:
                last_error = exc
                print(f"Attempt {attempt} failed: {exc}")
                if attempt < MAX_CONNECT_RETRIES:
                    time.sleep(CONNECT_RETRY_DELAY)
                    self._client = StreamingClient(
                        api_key=ORBITALSAI_API_KEY,
                        config=self._config,
                    )
 
        raise RuntimeError(
            f"Could not connect after {MAX_CONNECT_RETRIES} attempts: {last_error}"
        )
 
    def stream(self, audio_chunk: bytes) -> None:
        """Convert one Twilio mu-law frame and forward it to OrbitalsAI."""
        try:
            pcm16_8k = audioop.ulaw2lin(audio_chunk, 2)
        except Exception as exc:
            print(f"mu-law decode error: {exc}")
            return
 
        try:
            pcm16_16k, self._ratecv_state = audioop.ratecv(
                pcm16_8k, 2, 1,
                TWILIO_SAMPLE_RATE, ORBITALS_SAMPLE_RATE,
                self._ratecv_state,
            )
        except Exception as exc:
            print(f"Resample error: {exc}")
            return
 
        with self._lock:
            self._buffer.extend(pcm16_16k)
            if len(self._buffer) < self._min_send_bytes:
                return
            to_send = bytes(self._buffer)
            self._buffer.clear()
 
        try:
            self._client.send_audio(to_send)
        except Exception as exc:
            print(f"send_audio error: {exc}")
 
    def close(self) -> None:
        """Flush any buffered audio and close the OrbitalsAI session."""
        with self._lock:
            remaining = bytes(self._buffer)
            self._buffer.clear()
 
        if remaining:
            try:
                self._client.send_audio(remaining)
            except Exception as exc:
                print(f"Error flushing remaining audio: {exc}")
 
        try:
            self._client.flush()
        except Exception as exc:
            print(f"Error calling flush(): {exc}")
 
        time.sleep(1.0)  # Give OrbitalsAI time to emit the final transcript
        self._client.disconnect()

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
import base64
import json
 
from fastapi import FastAPI, Request, WebSocket, WebSocketDisconnect
from fastapi.responses import PlainTextResponse, Response
 
from twilio_transcriber import TwilioOrbitalsTranscriber
 
PORT = 5000
INCOMING_CALL_ROUTE = "/"
WEBSOCKET_ROUTE = "/realtime"
 
app = FastAPI()
 
 
@app.api_route(INCOMING_CALL_ROUTE, methods=["GET", "POST"])
async def receive_call(request: Request):
    """
    GET  → health check
    POST → TwiML that tells Twilio to open an audio stream to /realtime
    """
    if request.method == "POST":
        host = request.headers.get("host", "localhost")
        twiml = f"""
<Response>
  <Say>Speak to see your speech transcribed in the console.</Say>
  <Connect>
    <Stream url="wss://{host}{WEBSOCKET_ROUTE}"/>
  </Connect>
</Response>
""".strip()
        return Response(content=twiml, media_type="text/xml")
 
    return PlainTextResponse("OrbitalsAI real-time phone-call transcription")
 
 
@app.websocket(WEBSOCKET_ROUTE)
async def transcription_websocket(ws: WebSocket):
    """
    Twilio sends four event types over this WebSocket:
      connected: handshake complete, no audio yet
      start: audio stream is about to begin
      media: one frame of base64-encoded mu-law audio
      stop: call ended or stream closed
    """
    await ws.accept()
    transcriber: TwilioOrbitalsTranscriber | None = None
 
    try:
        while True:
            message = await ws.receive_text()
            data = json.loads(message)
            event_type = data.get("event")
 
            if event_type == "connected":
                try:
                    transcriber = TwilioOrbitalsTranscriber()
                    transcriber.connect()
                    print("Twilio connected: OrbitalsAI streaming active")
                except Exception as exc:
                    print(f"Could not start transcriber: {exc}")
                    transcriber = None
 
            elif event_type == "start":
                print("Audio stream started")
 
            elif event_type == "media":
                if not transcriber:
                    continue
                raw = base64.b64decode(data["media"]["payload"])
                transcriber.stream(raw)
 
            elif event_type == "stop":
                print("Audio stream stopped")
                if transcriber:
                    transcriber.close()
                break
 
    except WebSocketDisconnect:
        print("Twilio WebSocket disconnected")
        if transcriber:
            transcriber.close()
 
 
if __name__ == "__main__":
    import uvicorn
    uvicorn.run("main:app", host="0.0.0.0", port=PORT, reload=False)

1
python main.py

1
ngrok http 5000