# RealtimeAgent ## ``autogen.agentchat.realtime.experimental.RealtimeAgent [\#](https://docs.ag2.ai/0.10.3/docs/api-reference/autogen/agentchat/realtime/experimental/RealtimeAgent/#autogen.agentchat.realtime.experimental.RealtimeAgent "Permanent link") ``` RealtimeAgent(*, name, audio_adapter=None, system_message='You are a helpful AI Assistant.', llm_config=None, logger=None, observers=None, **client_kwargs) ``` (Experimental) Agent for interacting with the Realtime Clients. | PARAMETER | DESCRIPTION | | --- | --- | | `name` | The name of the agent.
**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` | | | | --- | --- | | ```
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
``` | ```
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.
"""
self._logger = logger
self._name = name
self._system_message = system_message
llm_config = LLMConfig.get_current_llm_config(llm_config)
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()
``` | ### ``callbacks`instance-attribute`[\#](https://docs.ag2.ai/0.10.3/docs/api-reference/autogen/agentchat/realtime/experimental/RealtimeAgent/#autogen.agentchat.realtime.experimental.RealtimeAgent.callbacks "Permanent link") ``` callbacks = RealtimeAgentCallbacks() ``` ### ``system_message`property`[\#](https://docs.ag2.ai/0.10.3/docs/api-reference/autogen/agentchat/realtime/experimental/RealtimeAgent/#autogen.agentchat.realtime.experimental.RealtimeAgent.system_message "Permanent link") ``` system_message ``` Get the system message for the agent. ### ``logger`property`[\#](https://docs.ag2.ai/0.10.3/docs/api-reference/autogen/agentchat/realtime/experimental/RealtimeAgent/#autogen.agentchat.realtime.experimental.RealtimeAgent.logger "Permanent link") ``` logger ``` Get the logger for the agent. ### ``realtime_client`property`[\#](https://docs.ag2.ai/0.10.3/docs/api-reference/autogen/agentchat/realtime/experimental/RealtimeAgent/#autogen.agentchat.realtime.experimental.RealtimeAgent.realtime_client "Permanent link") ``` realtime_client ``` Get the OpenAI Realtime Client. ### ``registered_realtime_tools`property`[\#](https://docs.ag2.ai/0.10.3/docs/api-reference/autogen/agentchat/realtime/experimental/RealtimeAgent/#autogen.agentchat.realtime.experimental.RealtimeAgent.registered_realtime_tools "Permanent link") ``` registered_realtime_tools ``` Get the registered realtime tools. ### ``register_observer [\#](https://docs.ag2.ai/0.10.3/docs/api-reference/autogen/agentchat/realtime/experimental/RealtimeAgent/#autogen.agentchat.realtime.experimental.RealtimeAgent.register_observer "Permanent link") ``` register_observer(observer) ``` Register an observer with the Realtime Agent. | PARAMETER | DESCRIPTION | | --- | --- | | `observer` | The observer to register.
**TYPE:**`RealtimeObserver` | Source code in `autogen/agentchat/realtime/experimental/realtime_agent.py` | | | | --- | --- | | ```
94
95
96
97
98
99
100
``` | ```
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)
``` | ### ``start_observers`async`[\#](https://docs.ag2.ai/0.10.3/docs/api-reference/autogen/agentchat/realtime/experimental/RealtimeAgent/#autogen.agentchat.realtime.experimental.RealtimeAgent.start_observers "Permanent link") ``` start_observers() ``` Source code in `autogen/agentchat/realtime/experimental/realtime_agent.py` | | | | --- | --- | | ```
102
103
104
105
106
107
108
109
110
``` | ```
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()
``` | ### ``run`async`[\#](https://docs.ag2.ai/0.10.3/docs/api-reference/autogen/agentchat/realtime/experimental/RealtimeAgent/#autogen.agentchat.realtime.experimental.RealtimeAgent.run "Permanent link") ``` run() ``` Run the agent. Source code in `autogen/agentchat/realtime/experimental/realtime_agent.py` | | | | --- | --- | | ```
112
113
114
115
116
117
118
119
120
121
122
123
124
125
``` | ```
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 [\#](https://docs.ag2.ai/0.10.3/docs/api-reference/autogen/agentchat/realtime/experimental/RealtimeAgent/#autogen.agentchat.realtime.experimental.RealtimeAgent.register_realtime_function "Permanent link") ``` 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` | | | | --- | --- | | ```
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
``` | ```
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
``` |