diff --git a/.env.example b/.env.example index 3ad03dbe..bae2de55 100644 --- a/.env.example +++ b/.env.example @@ -1 +1 @@ -OPENAI_API_KEY=your_api_key \ No newline at end of file +GEMINI_API_KEY=your_api_key diff --git a/Readme.md b/Readme.md index 36b869fd..dd791ae7 100644 --- a/Readme.md +++ b/Readme.md @@ -1,27 +1,24 @@ -# Speech Assistant with Twilio Voice and the OpenAI Realtime API (Python) +# Speech Assistant with Twilio Voice and the Gemini Live API (Python) -This application demonstrates how to use Python, [Twilio Voice](https://www.twilio.com/docs/voice) and [Media Streams](https://www.twilio.com/docs/voice/media-streams), and [OpenAI's Realtime API](https://platform.openai.com/docs/) to make a phone call to speak with an AI Assistant. +This application demonstrates how to use Python, [Twilio Voice](https://www.twilio.com/docs/voice) and [Media Streams](https://www.twilio.com/docs/voice/media-streams), and [Google's Gemini Live API](https://ai.google.dev/gemini-api/docs/live) to make a phone call to speak with an AI Assistant. -The application opens websockets with the OpenAI Realtime API and Twilio, and sends voice audio from one to the other to enable a two-way conversation. +The application opens websockets with the Gemini Live API and Twilio, and sends voice audio from one to the other to enable a two-way conversation. -See [here](https://www.twilio.com/en-us/blog/voice-ai-assistant-openai-realtime-api-python) for a tutorial overview of the code. - -This application uses the following Twilio products in conjunction with OpenAI's Realtime API: +This application uses the following Twilio products in conjunction with the Gemini Live API: - Voice (and TwiML, Media Streams) - Phone Numbers > [!NOTE] -> Outbound calling is beyond the scope of this app. However, we demoed [one way to do it here](https://www.twilio.com/en-us/blog/outbound-calls-python-openai-realtime-api-voice). +> Outbound calling is beyond the scope of this app. ## Prerequisites To use the app, you will need: -- **Python 3.9+** We used \`3.9.13\` for development; download from [here](https://www.python.org/downloads/). +- **Python 3.9+** (3.11 or earlier recommended; `audioop` is removed in 3.13+). Download from [here](https://www.python.org/downloads/). - **A Twilio account.** You can sign up for a free trial [here](https://www.twilio.com/try-twilio). - **A Twilio number with _Voice_ capabilities.** [Here are instructions](https://help.twilio.com/articles/223135247-How-to-Search-for-and-Buy-a-Twilio-Phone-Number-from-Console) to purchase a phone number. -- **An OpenAI account and an OpenAI API Key.** You can sign up [here](https://platform.openai.com/). - - **OpenAI Realtime API access.** +- **A Google AI Studio account and a Gemini API Key.** You can get one [here](https://aistudio.google.com/apikey). ## Local Setup @@ -43,7 +40,7 @@ Once the tunnel has been opened, copy the `Forwarding` URL. It will look somethi need this when configuring your Twilio number setup. Note that the `ngrok` command above forwards to a development server running on port `5050`, which is the default port configured in this application. If -you override the `PORT` defined in `index.js`, you will need to update the `ngrok` command accordingly. +you override the `PORT` defined in `main.py`, you will need to update the `ngrok` command accordingly. Keep in mind that each time you run the `ngrok http` command, a new URL will be created, and you'll need to update it everywhere it is referenced below. @@ -72,13 +69,13 @@ In your Phone Number configuration settings, update the first **A call comes in* ### Update the .env file -Create a `/env` file, or copy the `.env.example` file to `.env`: +Create a `.env` file, or copy the `.env.example` file to `.env`: ``` cp .env.example .env ``` -In the .env file, update the `OPENAI_API_KEY` to your OpenAI API key from the **Prerequisites**. +In the .env file, update the `GEMINI_API_KEY` to your Gemini API key from the **Prerequisites**. ## Run the app Once ngrok is running, dependencies are installed, Twilio is configured properly, and the `.env` is set up, run the dev server with the following command: @@ -88,12 +85,13 @@ python main.py ## Test the app With the development server running, call the phone number you purchased in the **Prerequisites**. After the introduction, you should be able to talk to the AI Assistant. Have fun! -## Special features +## Audio Pipeline + +Twilio sends PCMU (G.711 μ-law, 8kHz) audio. Gemini Live expects raw PCM (16-bit, little-endian, 16kHz). The server handles transcoding: -### Have the AI speak first -To have the AI voice assistant talk before the user, uncomment the line `# await send_initial_conversation_item(openai_ws)`. The initial greeting is controlled in `async def send_initial_conversation_item(openai_ws)`. +- **Twilio → Gemini**: PCMU 8kHz → `audioop.ulaw2lin` → PCM 8kHz → `audioop.ratecv` (upsample 8→16kHz) → Gemini +- **Gemini → Twilio**: PCM 24kHz → struct averaging (downsample 24→8kHz) → `audioop.lin2ulaw` → PCMU → Twilio -### Interrupt handling/AI preemption -When the user speaks and OpenAI sends `input_audio_buffer.speech_started`, the code will clear the Twilio Media Streams buffer and send OpenAI `conversation.item.truncate`. +## Interrupt handling / Barge-in -Depending on your application's needs, you may want to use the [`input_audio_buffer.speech_stopped`](https://platform.openai.com/docs/api-reference/realtime-server-events/input-audio-buffer-speech-stopped) event, instead, or a combination of the two. +Gemini Live handles barge-in natively with server-side VAD. When the user speaks during an AI response, Gemini sends `serverContent.interrupted: true`. The server then clears Twilio's audio buffer so the caller stops hearing the old response immediately. No manual truncation messages needed. diff --git a/main.py b/main.py index d09eb664..5da44794 100644 --- a/main.py +++ b/main.py @@ -1,238 +1,244 @@ import os import json import base64 +import struct import asyncio +import audioop import websockets from fastapi import FastAPI, WebSocket, Request from fastapi.responses import HTMLResponse, JSONResponse from fastapi.websockets import WebSocketDisconnect -from twilio.twiml.voice_response import VoiceResponse, Connect, Say, Stream +from twilio.twiml.voice_response import VoiceResponse, Connect from dotenv import load_dotenv -load_dotenv() +load_dotenv(override=True) # Configuration -OPENAI_API_KEY = os.getenv('OPENAI_API_KEY') -PORT = int(os.getenv('PORT', 5050)) -TEMPERATURE = float(os.getenv('TEMPERATURE', 0.8)) +GEMINI_API_KEY = os.getenv("GEMINI_API_KEY") +PORT = int(os.getenv("PORT", 5050)) SYSTEM_MESSAGE = ( "You are a helpful and bubbly AI assistant who loves to chat about " "anything the user is interested in and is prepared to offer them facts. " "You have a penchant for dad jokes, owl jokes, and rickrolling – subtly. " "Always stay positive, but work in a joke when appropriate." ) -VOICE = 'alloy' -LOG_EVENT_TYPES = [ - 'error', 'response.content.done', 'rate_limits.updated', - 'response.done', 'input_audio_buffer.committed', - 'input_audio_buffer.speech_stopped', 'input_audio_buffer.speech_started', - 'session.created', 'session.updated' -] -SHOW_TIMING_MATH = False +MODEL = "gemini-3.1-flash-live-preview" +VOICE = "Aoede" app = FastAPI() -if not OPENAI_API_KEY: - raise ValueError('Missing the OpenAI API key. Please set it in the .env file.') +if not GEMINI_API_KEY: + raise ValueError("Missing the Gemini API key. Please set it in the .env file.") + + +def ulaw_to_pcm16k_base64(ulaw_b64: str) -> str: + """PCMU base64 (8kHz) → PCM16 base64 (16kHz). Upsampled for Gemini.""" + pcm8 = audioop.ulaw2lin(base64.b64decode(ulaw_b64), 2) + pcm16, _ = audioop.ratecv(pcm8, 2, 1, 8000, 16000, None) + return base64.b64encode(pcm16).decode("utf-8") + + +class OutputAudioConverter: + """Stateful 24kHz PCM → 8kHz PCMU converter with buffered alignment.""" + + def __init__(self): + self._buffer = b"" + + def pcm24k_to_ulaw_base64(self, pcm_b64: str) -> str | None: + raw = base64.b64decode(pcm_b64) + self._buffer += raw + + # Need groups of 3 samples (6 bytes) for 3:1 downsampling + n_groups = len(self._buffer) // 6 + if n_groups == 0: + return None + + usable = n_groups * 6 + samples = struct.unpack_from(f"<{n_groups * 3}h", self._buffer) + self._buffer = self._buffer[usable:] + + # Average every 3 samples for 24kHz → 8kHz + averages = [ + (samples[i * 3] + samples[i * 3 + 1] + samples[i * 3 + 2]) // 3 + for i in range(n_groups) + ] + downsampled = struct.pack(f"<{n_groups}h", *averages) + ulaw = audioop.lin2ulaw(downsampled, 2) + return base64.b64encode(ulaw).decode("utf-8") + @app.get("/", response_class=JSONResponse) async def index_page(): return {"message": "Twilio Media Stream Server is running!"} + @app.api_route("/incoming-call", methods=["GET", "POST"]) async def handle_incoming_call(request: Request): """Handle incoming call and return TwiML response to connect to Media Stream.""" response = VoiceResponse() # punctuation to improve text-to-speech flow - response.say( - "Please wait while we connect your call to the A. I. voice assistant, powered by Twilio and the Open A I Realtime API", - voice="Google.en-US-Chirp3-HD-Aoede" - ) - response.pause(length=1) - response.say( - "O.K. you can start talking!", - voice="Google.en-US-Chirp3-HD-Aoede" - ) + response.say("Connecting you now.", voice="Google.en-US-Chirp3-HD-Aoede") host = request.url.hostname connect = Connect() - connect.stream(url=f'wss://{host}/media-stream') + connect.stream(url=f"wss://{host}/media-stream") response.append(connect) return HTMLResponse(content=str(response), media_type="application/xml") + @app.websocket("/media-stream") async def handle_media_stream(websocket: WebSocket): - """Handle WebSocket connections between Twilio and OpenAI.""" + """Handle WebSocket connections between Twilio and Gemini Live.""" print("Client connected") await websocket.accept() async with websockets.connect( - f"wss://api.openai.com/v1/realtime?model=gpt-realtime&temperature={TEMPERATURE}", - additional_headers={ - "Authorization": f"Bearer {OPENAI_API_KEY}" - } - ) as openai_ws: - await initialize_session(openai_ws) + f"wss://generativelanguage.googleapis.com/ws/google.ai.generativelanguage.v1beta.GenerativeService.BidiGenerateContent?key={GEMINI_API_KEY}" + ) as gemini_ws: + await initialize_session(gemini_ws) + + # Have Gemini speak first + await gemini_ws.send( + json.dumps( + { + "realtimeInput": { + "text": "Say hi and ask how you can help. Keep it under 15 words." + } + } + ) + ) # Connection specific state stream_sid = None latest_media_timestamp = 0 - last_assistant_item = None mark_queue = [] - response_start_timestamp_twilio = None - + gemini_transcript = "" + output_converter = OutputAudioConverter() + async def receive_from_twilio(): - """Receive audio data from Twilio and send it to the OpenAI Realtime API.""" + """Receive audio data from Twilio and send it to Gemini Live.""" nonlocal stream_sid, latest_media_timestamp try: async for message in websocket.iter_text(): data = json.loads(message) - if data['event'] == 'media' and openai_ws.state.name == 'OPEN': - latest_media_timestamp = int(data['media']['timestamp']) - audio_append = { - "type": "input_audio_buffer.append", - "audio": data['media']['payload'] - } - await openai_ws.send(json.dumps(audio_append)) - elif data['event'] == 'start': - stream_sid = data['start']['streamSid'] + if data["event"] == "media" and gemini_ws.state.name == "OPEN": + latest_media_timestamp = int(data["media"]["timestamp"]) + await gemini_ws.send( + json.dumps( + { + "realtimeInput": { + "audio": { + "data": ulaw_to_pcm16k_base64( + data["media"]["payload"] + ), + "mimeType": "audio/pcm;rate=16000", + } + } + } + ) + ) + elif data["event"] == "start": + stream_sid = data["start"]["streamSid"] print(f"Incoming stream has started {stream_sid}") - response_start_timestamp_twilio = None latest_media_timestamp = 0 - last_assistant_item = None - elif data['event'] == 'mark': + elif data["event"] == "mark": if mark_queue: mark_queue.pop(0) except WebSocketDisconnect: print("Client disconnected.") - if openai_ws.state.name == 'OPEN': - await openai_ws.close() + if gemini_ws.state.name == "OPEN": + await gemini_ws.close() async def send_to_twilio(): - """Receive events from the OpenAI Realtime API, send audio back to Twilio.""" - nonlocal stream_sid, last_assistant_item, response_start_timestamp_twilio + """Receive events from Gemini Live, send audio back to Twilio.""" + nonlocal stream_sid, gemini_transcript try: - async for openai_message in openai_ws: - response = json.loads(openai_message) - if response['type'] in LOG_EVENT_TYPES: - print(f"Received event: {response['type']}", response) - - if response.get('type') == 'response.output_audio.delta' and 'delta' in response: - audio_payload = base64.b64encode(base64.b64decode(response['delta'])).decode('utf-8') - audio_delta = { - "event": "media", - "streamSid": stream_sid, - "media": { - "payload": audio_payload - } - } - await websocket.send_json(audio_delta) - - - if response.get("item_id") and response["item_id"] != last_assistant_item: - response_start_timestamp_twilio = latest_media_timestamp - last_assistant_item = response["item_id"] - if SHOW_TIMING_MATH: - print(f"Setting start timestamp for new response: {response_start_timestamp_twilio}ms") - - await send_mark(websocket, stream_sid) - - # Trigger an interruption. Your use case might work better using `input_audio_buffer.speech_stopped`, or combining the two. - if response.get('type') == 'input_audio_buffer.speech_started': - print("Speech started detected.") - if last_assistant_item: - print(f"Interrupting response with id: {last_assistant_item}") - await handle_speech_started_event() + async for gemini_message in gemini_ws: + response = json.loads(gemini_message) + + if "serverContent" in response: + server_content = response["serverContent"] + + # Audio output from Gemini + if "modelTurn" in server_content: + for part in server_content["modelTurn"].get("parts", []): + if "inlineData" in part: + audio_b64_ulaw = ( + output_converter.pcm24k_to_ulaw_base64( + part["inlineData"]["data"] + ) + ) + if audio_b64_ulaw is None: + continue + audio_delta = { + "event": "media", + "streamSid": stream_sid, + "media": {"payload": audio_b64_ulaw}, + } + await websocket.send_json(audio_delta) + await send_mark(websocket, stream_sid) + + # Handle interruption (barge-in) + if server_content.get("interrupted"): + print("Gemini response interrupted by user speech.") + gemini_transcript = "" + await websocket.send_json( + {"event": "clear", "streamSid": stream_sid} + ) + mark_queue.clear() + + # Log transcription + if "outputTranscription" in server_content: + gemini_transcript += server_content["outputTranscription"][ + "text" + ] + + # User transcription + if "inputTranscription" in server_content: + user_text = server_content["inputTranscription"]["text"] + if user_text: + print(f"You said: {user_text}") + + # Print full Gemini transcript on turn complete + if server_content.get("turnComplete") and gemini_transcript: + print(f"Gemini said: {gemini_transcript}") + gemini_transcript = "" + except Exception as e: print(f"Error in send_to_twilio: {e}") - async def handle_speech_started_event(): - """Handle interruption when the caller's speech starts.""" - nonlocal response_start_timestamp_twilio, last_assistant_item - print("Handling speech started event.") - if mark_queue and response_start_timestamp_twilio is not None: - elapsed_time = latest_media_timestamp - response_start_timestamp_twilio - if SHOW_TIMING_MATH: - print(f"Calculating elapsed time for truncation: {latest_media_timestamp} - {response_start_timestamp_twilio} = {elapsed_time}ms") - - if last_assistant_item: - if SHOW_TIMING_MATH: - print(f"Truncating item with ID: {last_assistant_item}, Truncated at: {elapsed_time}ms") - - truncate_event = { - "type": "conversation.item.truncate", - "item_id": last_assistant_item, - "content_index": 0, - "audio_end_ms": elapsed_time - } - await openai_ws.send(json.dumps(truncate_event)) - - await websocket.send_json({ - "event": "clear", - "streamSid": stream_sid - }) - - mark_queue.clear() - last_assistant_item = None - response_start_timestamp_twilio = None - async def send_mark(connection, stream_sid): if stream_sid: mark_event = { "event": "mark", "streamSid": stream_sid, - "mark": {"name": "responsePart"} + "mark": {"name": "responsePart"}, } await connection.send_json(mark_event) - mark_queue.append('responsePart') + mark_queue.append("responsePart") await asyncio.gather(receive_from_twilio(), send_to_twilio()) -async def send_initial_conversation_item(openai_ws): - """Send initial conversation item if AI talks first.""" - initial_conversation_item = { - "type": "conversation.item.create", - "item": { - "type": "message", - "role": "user", - "content": [ - { - "type": "input_text", - "text": "Greet the user with 'Hello there! I am an AI voice assistant powered by Twilio and the OpenAI Realtime API. You can ask me for facts, jokes, or anything you can imagine. How can I help you?'" - } - ] - } - } - await openai_ws.send(json.dumps(initial_conversation_item)) - await openai_ws.send(json.dumps({"type": "response.create"})) - - -async def initialize_session(openai_ws): - """Control initial session with OpenAI.""" - session_update = { - "type": "session.update", - "session": { - "type": "realtime", - "model": "gpt-realtime", - "output_modalities": ["audio"], - "audio": { - "input": { - "format": {"type": "audio/pcmu"}, - "turn_detection": {"type": "server_vad"} + +async def initialize_session(gemini_ws): + """Send initial configuration to Gemini Live API.""" + setup_message = { + "setup": { + "model": f"models/{MODEL}", + "generationConfig": { + "responseModalities": ["AUDIO"], + "speechConfig": { + "voiceConfig": {"prebuiltVoiceConfig": {"voiceName": VOICE}} }, - "output": { - "format": {"type": "audio/pcmu"}, - "voice": VOICE - } }, - "instructions": SYSTEM_MESSAGE, + "systemInstruction": {"parts": [{"text": SYSTEM_MESSAGE}]}, + "inputAudioTranscription": {}, } } - print('Sending session update:', json.dumps(session_update)) - await openai_ws.send(json.dumps(session_update)) + print("Sending setup:", json.dumps(setup_message)) + await gemini_ws.send(json.dumps(setup_message)) - # Uncomment the next line to have the AI speak first - # await send_initial_conversation_item(openai_ws) if __name__ == "__main__": import uvicorn + uvicorn.run(app, host="0.0.0.0", port=PORT)