AudioObserver - AG2
AudioObserver
``autogen.agentchat.realtime.experimental.AudioObserver #
AudioObserver(*, logger=None)
Bases: RealtimeObserver
Observer for user voice input
Source code in autogen/agentchat/realtime/experimental/audio_observer.py
<br>19<br>20<br>21<br> |
<br>def __init__(self, *, logger: Optional["Logger"] = None) -> None:<br> """Observer for user voice input"""<br> super().__init__(logger=logger)<br> |
``loggerproperty#
logger
``agentproperty#
agent
``realtime_clientproperty#
realtime_client
``runasync#
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
<br>56<br>57<br>58<br>59<br>60<br>61<br>62<br>63<br>64<br>65<br>66<br>67<br>68<br> |
<br>async def run(self, agent: "RealtimeAgent") -> None:<br> """Run the observer with the agent.<br> When implementing, be sure to call `self._ready_event.set()` when the observer is ready to process events.<br> Args:<br> agent (RealtimeAgent): The realtime agent attached to the observer.<br> """<br> self._agent = agent<br> await self.initialize_session()<br> self._ready_event.set()<br> await self.run_loop()<br> |
``wait_for_readyasync#
wait_for_ready()
Get the event that is set when the observer is ready.
Source code in autogen/agentchat/realtime/experimental/realtime_observer.py
<br>85<br>86<br>87<br> |
<br>async def wait_for_ready(self) -> None:<br> """Get the event that is set when the observer is ready."""<br> await self._ready_event.wait()<br> |
``on_closeasync#
on_close()
Handle close of RealtimeClient.
Source code in autogen/agentchat/realtime/experimental/realtime_observer.py
<br> 98<br> 99<br>100<br> |
<br>async def on_close(self) -> None:<br> """Handle close of RealtimeClient."""<br> ...<br> |
``on_eventasync#
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
<br>23<br>24<br>25<br>26<br>27<br>28<br>29<br>30<br> |
<br>async def on_event(self, event: RealtimeEvent) -> None:<br> """Observe voice input events from the Realtime.<br> Args:<br> event (dict[str, Any]): The event from the OpenAI Realtime API.<br> """<br> if isinstance(event, InputAudioBufferDelta):<br> self.logger.info("Received audio buffer delta")<br> |
``initialize_sessionasync#
initialize_session()
No need to initialize session from this observer
Source code in autogen/agentchat/realtime/experimental/audio_observer.py
<br>32<br>33<br>34<br> |
<br>async def initialize_session(self) -> None:<br> """No need to initialize session from this observer"""<br> pass<br> |
``run_loopasync#
run_loop()
Run the observer loop.
Source code in autogen/agentchat/realtime/experimental/audio_observer.py
<br>36<br>37<br>38<br> |
<br>async def run_loop(self) -> None:<br> """Run the observer loop."""<br> pass<br> |