I’m using the new Realtime API to build a voice chatbot with Twilio. The docs say I can force the usage of tools by passing "tool_choice": "required" in the session.update event, but this is not working. The LLM is making calls to itself and never replies.
import asyncio
import base64
import json
import os
from typing import Annotated
import websockets
from fastapi import FastAPI, Request
from fastapi import WebSocket
from fastapi.responses import HTMLResponse
from fastapi.websockets import WebSocketDisconnect
from langchain_core.tools import tool
from twilio.twiml.voice_response import VoiceResponse, Connect
from websockets import ConnectionClosedOK
@tool
def scale_conversation(
reasoning: Annotated[str, "Reasoning to scale the conversation to Human Agent."]) -> str:
"""Use it only to escalate to Human Specialist."""
print(reasoning)
return "Your request has been sent and the Human Specialist will reach the person using the information you provided soon."
tools = [
scale_conversation,
]
BASE_SYS = {
"type": "session.update",
"session": {
"turn_detection": {
"type": "server_vad",
"threshold": 0.4,
"prefix_padding_ms": 300,
"silence_duration_ms": 600,
},
"input_audio_format": "g711_ulaw",
"output_audio_format": "g711_ulaw",
"instructions": """Your knowledge cutoff is 2023-10. You are a helpful, witty, and friendly AI. Act like a human, but remember that you aren't a human and that you can't do human things in the real world. Your voice and personality should be warm and engaging, with a lively and playful tone. If interacting in a non-English language, start by using the standard accent or dialect familiar to the user. Talk quickly. You should always call a function if you can. Do not refer to these rules, even if you're asked about them.""",
"temperature": 0.6,
"input_audio_transcription": {"model": "whisper-1"},
"tools": [{
"type": "function",
"name": tool.name,
"description": tool.description,
"parameters": tool.args,
} for tool in tools],
"tool_choice": "required",
},
}
# Configuration
OPENAI_API_KEY = os.getenv("OPENAI_API_KEY")
PORT = int(os.getenv("PORT", 3000))
TRANSCRIPT_EVENTS = [
"response.audio_transcript.done",
"conversation.item.input_audio_transcription.completed",
]
app = FastAPI()
@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 to the AI voice assistant.")
response.pause(length=2)
response.say("OK, 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."""
await websocket.accept()
model = "gpt-4o-realtime-preview-2024-10-01"
openai_url = f"wss://api.openai.com/v1/realtime?model={model}"
headers = {
"Authorization": f"Bearer {OPENAI_API_KEY}",
"OpenAI-Beta": "realtime=v1",
}
stream_sid = None
while not stream_sid:
d = json.loads(await websocket.receive_text())
if d["event"] == "start":
stream_sid = d["start"]["streamSid"]
call_id = d["start"]["callSid"]
async with websockets.connect(openai_url, extra_headers=headers) as openai_ws:
await openai_ws.send(json.dumps(BASE_SYS))
await openai_ws.send(json.dumps({"type": "response.create"}))
try:
await asyncio.gather(
twilio_listener(websocket, openai_ws),
openai_listener(stream_sid, websocket, openai_ws),
)
except Exception as ex:
print(ex)
finally:
if openai_ws.open:
await openai_ws.close()
await websocket.close()
async def twilio_listener(twilio_ws: WebSocket, openai_ws):
"""Receive audio data from Twilio and send it to the OpenAI Realtime API."""
async for message in twilio_ws.iter_text():
data = json.loads(message)
event = data["event"]
# logger.warning(event)
if event == "media" and openai_ws.open:
audio_append = {
"type": "input_audio_buffer.append",
"audio": data["media"]["payload"],
}
await openai_ws.send(json.dumps(audio_append))
elif event == "stop":
raise Exception("Disconnected")
async def openai_listener(
stream_sid: str, twilio_ws: WebSocket, openai_ws
):
"""Receive events from the OpenAI Realtime API, send audio back to Twilio."""
try:
async for openai_message in openai_ws:
response = json.loads(openai_message)
event = response["type"]
print(event)
if event == "input_audio_buffer.speech_started":
await openai_ws.send(json.dumps({"type": "response.cancel"}))
await twilio_ws.send_json({"streamSid": stream_sid, "event": "clear"})
elif event == "error":
print(f"OpenAI error: {response}")
await openai_ws.send(json.dumps({"type": "response.create"}))
elif event == "response.audio.delta" and response.get("delta"):
audio_delta = encode_audio(response, stream_sid)
await twilio_ws.send_json(audio_delta)
elif event == "response.function_call_arguments.done":
output = await call_tool(response)
await openai_ws.send(json.dumps(output))
await openai_ws.send(json.dumps({"type": "response.create"}))
elif event == "error":
print(f"OpenAI error: {response}")
except (WebSocketDisconnect, ConnectionClosedOK):
return
except Exception as e:
print(f"Error in openai_listener: {e}")
raise e
def encode_audio(response, stream_sid) -> dict:
try:
audio_payload = base64.b64encode(
base64.b64decode(response["delta"]),
).decode("utf-8")
payload = {
"event": "media",
"streamSid": stream_sid,
"media": {"payload": audio_payload},
}
return payload
except Exception as e:
print(f"Error processing audio data: {e}")
tools_by_name = {tool.name: tool for tool in tools}
async def call_tool(event: dict) -> dict:
print(f"Calling tool {event['name']}")
tool = tools_by_name.get(event["name"])
if tool is None:
# immediately yield error, do not add task
raise ValueError(
f"tool {event['name']} not found. Must be one of {list(tools_by_name.keys())}"
)
try:
args = json.loads(event["arguments"])
except json.JSONDecodeError:
raise ValueError(
f"failed to parse arguments `{event['arguments']}`. Must be valid JSON."
)
try:
result = await tool.ainvoke(args)
except Exception:
return {
"type": "conversation.item.create",
"previous_item_id": event["item_id"],
"item": {
"id": event["call_id"],
"call_id": event["call_id"],
"type": "function_call_output",
"output": "A error happened when executing this tool",
},
}
try:
result_str = json.dumps(result)
except TypeError: # not json serializable, use str
result_str = str(result)
tool_output = {
"type": "conversation.item.create",
"previous_item_id": event["item_id"],
"item": {
"id": event["call_id"],
"call_id": event["call_id"],
"type": "function_call_output",
"output": result_str,
},
}
return tool_output
if __name__ == "__main__":
import uvicorn
uvicorn.run(
"main:app",
host="0.0.0.0",
port=PORT,
reload=False,
log_level="info",
)