AG2 supports RealtimeAgent, a powerful agent type that connects seamlessly to OpenAI’s Realtime API. In this example we will start a local RealtimeAgent and register a mock get_weather function that the agent will be able to call.

Note: This notebook cannot be run in Google Colab because it depends on local JavaScript files and HTML templates. To execute the notebook successfully, run it locally within the cloned project so that the notebooks/agentchat_realtime_websocket/static and notebooks/agentchat_realtime_websocket/templates folders are available in the correct relative paths.

git clone https://github.com/ag2ai/ag2.git
cd ag2

​

Install AG2 and dependencies

To use the realtime agent we will connect it to a local websocket trough the browser.

We have prepared a WebSocketAudioAdapter to enable you to connect your realtime agent to a websocket service.

To be able to run this notebook, you will need to install ag2, fastapi and uvicorn.

pip install "ag2", "fastapi>=0.115.0,<1", "uvicorn>=0.30.6,<1" "flaml[automl]"

!pip install "ag2" "fastapi>=0.115.0,<1" "uvicorn>=0.30.6,<1"

​

Import the dependencies

After installing the necessary requirements, we can import the necessary dependencies for the example

import os
from logging import getLogger
from pathlib import Path
from typing import Annotated

import uvicorn
from fastapi import FastAPI, Request, WebSocket
from fastapi.responses import HTMLResponse, JSONResponse
from fastapi.staticfiles import StaticFiles
from fastapi.templating import Jinja2Templates

import autogen
from autogen.agentchat.realtime_agent import RealtimeAgent, WebSocketAudioAdapter

​

Prepare your llm_config and realtime_llm_config

The config_list_from_json function loads a list of configurations from an environment variable or a json file.

realtime_config_list = autogen.config_list_from_json(
    "OAI_CONFIG_LIST",
    filter_dict={
        "tags": ["gpt-4o-mini-realtime"],
    },
)

realtime_llm_config = {
    "timeout": 600,
    "config_list": realtime_config_list,
    "temperature": 0.8,
}

assert realtime_config_list, (
    "No LLM found for the given model, please add the following lines to the OAI_CONFIG_LIST file:"
    """
    {
        "model": "gpt-4o-mini-realtime-preview",
        "api_key": "sk-***********************...*",
        "tags": ["gpt-4o-mini-realtime", "realtime"]
    }"""
)

​

Before you start the server

To run uvicorn server inside the notebook, you will need to use nest_asyncio. This is because Jupyter uses the asyncio event loop, and uvicorn uses its own event loop. nest_asyncio will allow uvicorn to run in Jupyter.

Please install nest_asyncio by running the following cell.

!pip install nest_asyncio

import nest_asyncio

nest_asyncio.apply()

​

Implementing and Running a Basic App

Let us set up and execute a FastAPI application that integrates real-time agent interactions.

​

Define basic FastAPI app

1. Define Port: Sets the PORT variable to 5050, which will be used for the server.
2. Initialize FastAPI App: Creates a FastAPI instance named app, which serves as the main application.
3. Define Root Endpoint: Adds a GET endpoint at the root URL (/). When accessed, it returns a JSON response with the message "Websocket Audio Stream Server is running!".

This sets up a basic FastAPI server and provides a simple health-check endpoint to confirm that the server is operational.

PORT = 5050

app = FastAPI()


@app.get("/", response_class=JSONResponse)
async def index_page():
    return {"message": "Websocket Audio Stream Server is running!"}

​

Prepare start-chat endpoint

1. Set the Working Directory: Define notebook_path as the current working directory using os.getcwd().
2. Mount Static Files: Mount the static directory (inside agentchat_realtime_websocket) to serve JavaScript, CSS, and other static assets under the /static path.
3. Set Up Templates: Configure Jinja2 to render HTML templates from the templates directory within agentchat_realtime_websocket.
4. Create the /start-chat/ Endpoint: Define a GET route that serves the chat.html template. Pass the client’s request and the port variable to the template for rendering a dynamic page for the audio chat interface.

This code sets up static file handling, template rendering, and a dedicated endpoint to deliver the chat interface.

notebook_path = os.getcwd()

app.mount(
    "/static", StaticFiles(directory=Path(notebook_path) / "agentchat_realtime_websocket" / "static"), name="static"
)

# Templates for HTML responses

templates = Jinja2Templates(directory=Path(notebook_path) / "agentchat_realtime_websocket" / "templates")


@app.get("/start-chat/", response_class=HTMLResponse)
async def start_chat(request: Request):
    """Endpoint to return the HTML page for audio chat."""
    port = PORT  # Extract the client's port
    return templates.TemplateResponse("chat.html", {"request": request, "port": port})

​

Prepare endpint for converstion audio stream

1. Set Up the WebSocket Endpoint: Define the /media-stream WebSocket route to handle audio streaming.
2. Accept WebSocket Connections: Accept incoming WebSocket connections from clients.
3. Initialize Logger: Retrieve a logger instance for logging purposes.
4. Configure Audio Adapter: Instantiate a WebSocketAudioAdapter, connecting the WebSocket to handle audio streaming with logging.
5. Set Up Realtime Agent: Create a RealtimeAgent with the following:
   • Name: Weather Bot.
   • System Message: Introduces the AI assistant and its capabilities.
   • LLM Configuration: Uses realtime_llm_config for language model settings.
   • Audio Adapter: Leverages the previously created audio_adapter.
   • Logger: Logs activities for debugging and monitoring.
6. Register a Realtime Function: Add a function get_weather to the agent, allowing it to respond with basic weather information based on the provided location.
7. Run the Agent: Start the realtime_agent to handle interactions in real time.

@app.websocket("/media-stream")
async def handle_media_stream(websocket: WebSocket):
    """Handle WebSocket connections providing audio stream and OpenAI."""
    await websocket.accept()

    logger = getLogger("uvicorn.error")

    audio_adapter = WebSocketAudioAdapter(websocket, logger=logger)
    realtime_agent = RealtimeAgent(
        name="Weather Bot",
        system_message="Hello there! I am an AI voice assistant powered by Autogen and the OpenAI Realtime API. You can ask me about weather, jokes, or anything you can imagine. Start by saying 'How can I help you'?",
        llm_config=realtime_llm_config,
        audio_adapter=audio_adapter,
        logger=logger,
    )

    @realtime_agent.register_realtime_function(name="get_weather", description="Get the current weather")
    def get_weather(location: Annotated[str, "city"]) -> str:
        return "The weather is cloudy." if location == "Seattle" else "The weather is sunny."

    await realtime_agent.run()

​

Run the app using uvicorn

uvicorn.run(app, host="0.0.0.0", port=PORT)