Skip to content

RealtimeObserver

autogen.agentchat.realtime.experimental.RealtimeObserver #

RealtimeObserver(*, logger=None)

Bases: ABC

Observer for the OpenAI Realtime API.

Observer for the OpenAI Realtime API.

PARAMETER DESCRIPTION
logger

The logger for the observer.

TYPE: Logger DEFAULT: None

Source code in autogen/agentchat/realtime/experimental/realtime_observer.py 
27
28
29
30
31
32
33
34
35
def __init__(self, *, logger: Optional[Logger] = None) -> None:
    """Observer for the OpenAI Realtime API.

    Args:
        logger (Logger): The logger for the observer.
    """
    self._ready_event = Event()
    self._agent: Optional[RealtimeAgent] = None
    self._logger = logger

logger property #

logger

agent property #

agent

realtime_client property #

realtime_client

run async #

run(agent)

Run the observer with the agent.

When implementing, be sure to call self._ready_event.set() when the observer is ready to process events.

PARAMETER DESCRIPTION
agent

The realtime agent attached to the observer.

TYPE: RealtimeAgent

Source code in autogen/agentchat/realtime/experimental/realtime_observer.py 
56
57
58
59
60
61
62
63
64
65
66
67
68
async def run(self, agent: "RealtimeAgent") -> None:
    """Run the observer with the agent.

    When implementing, be sure to call `self._ready_event.set()` when the observer is ready to process events.

    Args:
        agent (RealtimeAgent): The realtime agent attached to the observer.
    """
    self._agent = agent
    await self.initialize_session()
    self._ready_event.set()

    await self.run_loop()

run_loop abstractmethod async #

run_loop()

Run the loop if needed.

This method is called after the observer is ready to process events. Events will be processed by the on_event method, this is just a hook for additional processing. Use initialize_session to set up the session.

Source code in autogen/agentchat/realtime/experimental/realtime_observer.py 
70
71
72
73
74
75
76
77
78
@abstractmethod
async def run_loop(self) -> None:
    """Run the loop if needed.

    This method is called after the observer is ready to process events.
    Events will be processed by the on_event method, this is just a hook for additional processing.
    Use initialize_session to set up the session.
    """
    ...

initialize_session abstractmethod async #

initialize_session()

Initialize the session for the observer.

Source code in autogen/agentchat/realtime/experimental/realtime_observer.py 
80
81
82
83
@abstractmethod
async def initialize_session(self) -> None:
    """Initialize the session for the observer."""
    ...

wait_for_ready async #

wait_for_ready()

Get the event that is set when the observer is ready.

Source code in autogen/agentchat/realtime/experimental/realtime_observer.py 
85
86
87
async def wait_for_ready(self) -> None:
    """Get the event that is set when the observer is ready."""
    await self._ready_event.wait()

on_event abstractmethod async #

on_event(event)

Handle an event from the OpenAI Realtime API.

PARAMETER DESCRIPTION
event

The event from the OpenAI Realtime API.

TYPE: RealtimeServerEvent

Source code in autogen/agentchat/realtime/experimental/realtime_observer.py 
89
90
91
92
93
94
95
96
@abstractmethod
async def on_event(self, event: RealtimeEvent) -> None:
    """Handle an event from the OpenAI Realtime API.

    Args:
        event (RealtimeServerEvent): The event from the OpenAI Realtime API.
    """
    ...

on_close async #

on_close()

Handle close of RealtimeClient.

Source code in autogen/agentchat/realtime/experimental/realtime_observer.py 
 98
 99
100
async def on_close(self) -> None:
    """Handle close of RealtimeClient."""
    ...