OpenAI-Realtime API support.

[drestr30]

Hello, I would like to ask if there is a future plan for integrating the new openAI realtime API (which works with sockets connections instead of HTTP request) to the langchain and langgraph frameworks.
Right now I'm building some voice agents with langgraph using the multi-agent architecture and using TTS and SST models to do voice processing, this approach instroduces a lot of lag into the voice interaction which is dramatically improved with the openAI realtime endpoint however I have not found documentation or a way to integrate this new model (sockets connection) to the langraph agent graph.
Please let me know if you already have an approach to solve this.
Thanks so much.

[tushar-madheshia]

Eagerly waiting for a reply on this.. @keenborder786

[surendrasalke]

Awaiting your response to this thread.

[ngDuyAnh]

Still waiting for this thread?

[edwinjosechittilappilly]

Looking forward to this support!

[acomito]

yes!

[uagarwal310]

Looking forward to the support on this, meanwhile found https://github.com/AK391/ai-gradio

Has anyone tried?

[dmitryrPlanner5D]

@drestr30
I found this langchain-ai/react-voice-agent repo which seems to be related to your request. It looks like a demo server integrating langchain tools with socket-based gpt-4o-realtime LLM.

[mahimairaja]

@dmitryrPlanner5D yes, but does lang graph supports for Web Socket to access the real time api?

[dmitryrPlanner5D]

@mahimairaja
As far as I know langgraph does not directly support web sockets,

but you can make an independent module with openai realtime llm (not in LangGraph) based on the repo above.
And this realtime module will call your main LangGraph logic as a tool.

Also I found this LlamaIndex realtime voice client implementation rather clean: https://github.com/run-llama/openai_realtime_client
And it looks like it is not hard to replace their LlamaIndex tools with LangChain ones or wrap LangGraph logic with LlamaIndex tool.

[citric-gabriel]

Any updates on this?

[austinmw]

Any updates?

[aniketjha1304]

Building Real-Time Voice Agents with VoiceAgentBuilder and Twilio

The agent-builder package provides a streamlined way to create real-time voice agents powered by OpenAI's Realtime API. This allows developers to integrate voice AI capabilities into their applications, handling voice interactions effectively.

Refer: https://github.com/Recall-Space/agent-builder

Example: Setting Up a Voice Agent with Twilio

Below is an example implementation of a voice agent that:

• Uses Twilio's Media Streams to handle voice calls.
• Streams real-time audio to OpenAI's Realtime API.
• Processes the response and sends it back to the caller.

FastAPI WebSocket Endpoints for Twilio Integration

import json
import logging
from typing import AsyncIterator

from fastapi import FastAPI, WebSocket, Request
from fastapi.responses import HTMLResponse
from starlette.websockets import WebSocketDisconnect
from twilio.twiml.voice_response import Connect, VoiceResponse

from agent_builder.builders.voice_agent_builder import VoiceAgentBuilder

logger = logging.getLogger(__name__)
app = FastAPI()

@app.api_route("/incoming-call", methods=["GET", "POST"])
async def handle_incoming_call(request: Request):
    """
    Handles incoming Twilio calls and returns TwiML to connect to the media stream.
    """
    response = VoiceResponse()
    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):
    """
    WebSocket endpoint to process audio streams from Twilio and send responses.
    """
    await websocket.accept()
    stream_sid = None

    async def input_audio_stream() -> AsyncIterator[str]:
        """
        Reads incoming audio from Twilio.
        """
        nonlocal stream_sid
        yield json.dumps({
            "type": "response.create",
            "response": {
                "modalities": ["text", "audio"],
                "voice": "alloy",
                "output_audio_format": "g711_ulaw"
            }
        })

        async for message in websocket.iter_text():
            data = json.loads(message)
            event = data.get("event")

            if event == "start":
                stream_sid = data["start"]["streamSid"]
                logger.info(f"Stream started: {stream_sid}")
            elif event == "media":
                yield json.dumps({
                    "type": "input_audio_buffer.append",
                    "audio": data["media"]["payload"]
                })
            elif event == "stop":
                logger.info("Stream stopped")
                break

    async def handle_output_event(event_str: str):
        """
        Processes responses from OpenAI and sends them to Twilio.
        """
        event = json.loads(event_str)
        if event.get("type") == "response.audio.delta" and "delta" in event:
            if stream_sid:
                await websocket.send_text(json.dumps({
                    "event": "media",
                    "streamSid": stream_sid,
                    "media": {"payload": event["delta"]}
                }))
    
    voice_agent = VoiceAgentBuilder().set_voice("alloy").set_input_audio_format("g711_ulaw").build()
    await voice_agent.ainvoke(input_audio_stream(), handle_output_event)

Example: Creating a Weather Tool for Use in Voice Agents

We can extend the capabilities of our voice agent by adding tools. Below is an example of a tool that fetches the current weather.

from pydantic import BaseModel, Field
from agent_builder.builders.tool_builder import ToolBuilder
import requests

class WeatherRequest(BaseModel):
    city: str = Field(description="City name to get the weather for.")

async def get_current_weather(city: str) -> str:
    """
    Fetches the current weather for a given city.
    """
    try:
        api_url = f"https://api.weatherapi.com/v1/current.json?key=YOUR_API_KEY&q={city}"
        response = requests.get(api_url).json()
        return f"Current weather in {city}: {response['current']['temp_c']}°C, {response['current']['condition']['text']}."
    except Exception as e:
        return f"Failed to fetch weather: {str(e)}"


def create_weather_tool():
    """
    Builds and returns a tool for fetching weather information.
    """
    tool_builder = ToolBuilder()
    tool_builder.set_name("GetWeather")
    tool_builder.set_function(get_current_weather)
    tool_builder.set_coroutine(get_current_weather)
    tool_builder.set_description("Fetches the current weather for a given city.")
    tool_builder.set_schema(WeatherRequest)
    return tool_builder.build()

Adding the Tool to the Voice Agent

weather_tool = create_weather_tool()
voice_agent = VoiceAgentBuilder().set_voice("alloy").set_input_audio_format("g711_ulaw").set_tools([weather_tool]).build()

Conclusion

This example showcases how you can:

• Use VoiceAgentBuilder to create real-time voice agents.
• Connect with Twilio for live calls.
• Extend capabilities with tools like a weather fetching tool.

Would love to hear how you're using this! 🚀

[himanshu-meragi]

Hey @aniketjha1304, thanks for this package, just quick note for people using this, set_output_audio_format("g711_ulaw") too while setting set_input_audio_format("g711_ulaw") otherwise you might get screeching sound. Thanks for the help @aniketjha1304

[tushar-madheshia]

As a workaround, I am keeping two layers of agents in my system. First layer has an openai agent (powered by realtime voice api), this agent is surfaced to the user. Since this is an OpenAI agent, I used the function call to actually call my Langgraph agents sitting in the second layer. So second layer has multiple agents built using langgraph.
Please checkout the code and youtube video for more details.

Git repo Link

Demo Link - Youtube