Hi,
Attached below is my code reference for the Twilio Realtime API (99% identical to the official Twilio example published last month).
Unfortunately, any temperature lower than 0.6 results in a “white noise” effect that makes the usability impossible. The white noise “replaces” the bot responses.
Has anyone experienced this or knows how to work around it?
import os
import json
import base64
import asyncio
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 dotenv import load_dotenv
load_dotenv()
# Configuration
OPENAI_API_KEY = os.getenv('OPENAI_API_KEY')
# PORT = int(os.getenv('PORT', 5050))
PORT = 8082
SYSTEM_MESSAGE = """
Example
"""
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'
]
SHOW_TIMING_MATH = False
app = FastAPI()
if not OPENAI_API_KEY:
raise ValueError('Missing the OpenAI API key. Please set it in the .env file.')
@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()
# <Say> punctuation to improve text-to-speech flow
# response.say("Please wait while we connect your call")
response.pause(length=1)
# response.say("O.K. you can start talking!")
host = request.url.hostname
connect = Connect()
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."""
print("Client connected")
await websocket.accept()
async with websockets.connect(
'wss://api.openai.com/v1/realtime?model=gpt-4o-realtime-preview-2024-10-01',
extra_headers={
"Authorization": f"Bearer {OPENAI_API_KEY}",
"OpenAI-Beta": "realtime=v1"
}
) as openai_ws:
await initialize_session(openai_ws)
# Connection specific state
stream_sid = None
latest_media_timestamp = 0
last_assistant_item = None
mark_queue = []
response_start_timestamp_twilio = None
async def receive_from_twilio():
"""Receive audio data from Twilio and send it to the OpenAI Realtime API."""
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.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']
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':
if mark_queue:
mark_queue.pop(0)
except WebSocketDisconnect:
print("Client disconnected.")
if openai_ws.open:
await openai_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
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.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_start_timestamp_twilio is None:
response_start_timestamp_twilio = latest_media_timestamp
if SHOW_TIMING_MATH:
print(f"Setting start timestamp for new response: {response_start_timestamp_twilio}ms")
# Update last_assistant_item safely
if response.get('item_id'):
last_assistant_item = response['item_id']
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()
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"}
}
await connection.send_json(mark_event)
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": SYSTEM_MESSAGE
}
]
}
}
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": {
"turn_detection": {"type": "server_vad"},
"input_audio_format": "g711_ulaw",
"output_audio_format": "g711_ulaw",
"voice": VOICE,
"instructions": SYSTEM_MESSAGE,
"modalities": ["text", "audio"],
"temperature": 0.6,
"input_audio_transcription": {
"model": "whisper-1"
},
}
}
print('Sending session update:', json.dumps(session_update))
await openai_ws.send(json.dumps(session_update))
# 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="8082")