Skip to content

AudioObserver

autogen.agentchat.realtime.experimental.AudioObserver #

AudioObserver(*, logger=None)

Bases: RealtimeObserver

Observer for user voice input

Observer for user voice input

Source code in autogen/agentchat/realtime/experimental/audio_observer.py 
19
20
21
def __init__(self, *, logger: Optional["Logger"] = None) -> None:
    """Observer for user voice input"""
    super().__init__(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()

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_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."""
    ...

on_event async #

on_event(event)

Observe voice input events from the Realtime.

PARAMETER DESCRIPTION
event

The event from the OpenAI Realtime API.

TYPE: dict[str, Any]

Source code in autogen/agentchat/realtime/experimental/audio_observer.py 
23
24
25
26
27
28
29
30
async def on_event(self, event: RealtimeEvent) -> None:
    """Observe voice input events from the Realtime.

    Args:
        event (dict[str, Any]): The event from the OpenAI Realtime API.
    """
    if isinstance(event, InputAudioBufferDelta):
        self.logger.info("Received audio buffer delta")

initialize_session async #

initialize_session()

No need to initialize session from this observer

Source code in autogen/agentchat/realtime/experimental/audio_observer.py 
32
33
34
async def initialize_session(self) -> None:
    """No need to initialize session from this observer"""
    pass

run_loop async #

run_loop()

Run the observer loop.

Source code in autogen/agentchat/realtime/experimental/audio_observer.py 
36
37
38
async def run_loop(self) -> None:
    """Run the observer loop."""
    pass