RealtimeAgent(*, name, audio_adapter=None, system_message='You are a helpful AI Assistant.', llm_config=None, logger=None, observers=None, **client_kwargs)
(Deprecated) Agent for real-time voice interactions.
.. deprecated:: RealtimeAgent is deprecated and will be removed in v0.14. It relies on deprecated realtime API endpoints.
(Experimental) Agent for interacting with the Realtime Clients.
| PARAMETER | DESCRIPTION |
name | TYPE: str |
audio_adapter | The audio adapter for the agent. TYPE: Optional[RealtimeObserver] = None DEFAULT: None |
system_message | The system message for the agent. TYPE: str DEFAULT: 'You are a helpful AI Assistant.' |
llm_config | The config for the agent. TYPE: (LLMConfig, dict[str, Any], bool) DEFAULT: None |
logger | The logger for the agent. TYPE: Optional[Logger] DEFAULT: None |
observers | The additional observers for the agent. TYPE: Optional[list[RealtimeObserver]] DEFAULT: None |
**client_kwargs | The keyword arguments for the client. TYPE: Any DEFAULT: {} |
Source code in autogen/agentchat/realtime/experimental/realtime_agent.py
| def __init__(
self,
*,
name: str,
audio_adapter: RealtimeObserver | None = None,
system_message: str = "You are a helpful AI Assistant.",
llm_config: LLMConfig | dict[str, Any] | None = None,
logger: Logger | None = None,
observers: list[RealtimeObserver] | None = None,
**client_kwargs: Any,
):
"""(Experimental) Agent for interacting with the Realtime Clients.
Args:
name (str): The name of the agent.
audio_adapter (Optional[RealtimeObserver] = None): The audio adapter for the agent.
system_message (str): The system message for the agent.
llm_config (LLMConfig, dict[str, Any], bool): The config for the agent.
logger (Optional[Logger]): The logger for the agent.
observers (Optional[list[RealtimeObserver]]): The additional observers for the agent.
**client_kwargs (Any): The keyword arguments for the client.
"""
warnings.warn(
"RealtimeAgent is deprecated and will be removed in v0.14. It relies on deprecated realtime API endpoints.",
DeprecationWarning,
stacklevel=2,
)
self._logger = logger
self._name = name
self._system_message = system_message
self._realtime_client: RealtimeClientProtocol = get_client(
llm_config=llm_config, logger=self.logger, **client_kwargs
)
self._registered_realtime_tools: dict[str, Tool] = {}
self._observers: list[RealtimeObserver] = observers if observers else []
self._observers.append(FunctionObserver(logger=logger))
if audio_adapter:
self._observers.append(audio_adapter)
self.callbacks = RealtimeAgentCallbacks()
|
instance-attribute
callbacks = RealtimeAgentCallbacks()
property
Get the system message for the agent.
property
Get the logger for the agent.
property
Get the OpenAI Realtime Client.
registered_realtime_tools
Get the registered realtime tools.
register_observer(observer)
Register an observer with the Realtime Agent.
Source code in autogen/agentchat/realtime/experimental/realtime_agent.py
| def register_observer(self, observer: RealtimeObserver) -> None:
"""Register an observer with the Realtime Agent.
Args:
observer (RealtimeObserver): The observer to register.
"""
self._observers.append(observer)
|
async
Source code in autogen/agentchat/realtime/experimental/realtime_agent.py
| async def start_observers(self) -> None:
for observer in self._observers:
self._tg.start_soon(observer.run, self)
# wait for the observers to be ready
for observer in self._observers:
await observer.wait_for_ready()
await self.callbacks.on_observers_ready()
|
async
Run the agent.
Source code in autogen/agentchat/realtime/experimental/realtime_agent.py
| async def run(self) -> None:
"""Run the agent."""
# everything is run in the same task group to enable easy cancellation using self._tg.cancel_scope.cancel()
async with create_task_group() as self._tg: # noqa: SIM117
# connect with the client first (establishes a connection and initializes a session)
async with self._realtime_client.connect():
# start the observers and wait for them to be ready
await self.realtime_client.session_update(session_options={"instructions": self.system_message})
await self.start_observers()
# iterate over the events
async for event in self.realtime_client.read_events():
for observer in self._observers:
await observer.on_event(event)
|
register_realtime_function(*, name=None, description=None)
Decorator for registering a function to be used by an agent.
| PARAMETER | DESCRIPTION |
name | The name of the function. TYPE: str DEFAULT: None |
description | The description of the function. TYPE: str DEFAULT: None |
| RETURNS | DESCRIPTION |
Callable[[F | Tool], Tool] | Callable[[Union[F, Tool]], Tool]: The decorator for registering a function. |
Source code in autogen/agentchat/realtime/experimental/realtime_agent.py
| def register_realtime_function(
self,
*,
name: str | None = None,
description: str | None = None,
) -> Callable[[F | Tool], Tool]:
"""Decorator for registering a function to be used by an agent.
Args:
name (str): The name of the function.
description (str): The description of the function.
Returns:
Callable[[Union[F, Tool]], Tool]: The decorator for registering a function.
"""
def _decorator(func_or_tool: F | Tool) -> Tool:
"""Decorator for registering a function to be used by an agent.
Args:
func_or_tool (Union[F, Tool]): The function or tool to register.
Returns:
Tool: The registered tool.
"""
tool = Tool(func_or_tool=func_or_tool, name=name, description=description)
self._registered_realtime_tools[tool.name] = tool
return tool
return _decorator
|