Skip to content

RealtimeAgent

autogen.agentchat.realtime.experimental.RealtimeAgent #

RealtimeAgent(*, name, audio_adapter=None, system_message='You are a helpful AI Assistant.', llm_config, 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: (dict[str, Any], bool)

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 
33
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
def __init__(
    self,
    *,
    name: str,
    audio_adapter: Optional[RealtimeObserver] = None,
    system_message: str = "You are a helpful AI Assistant.",
    llm_config: dict[str, Any],
    logger: Optional[Logger] = None,
    observers: Optional[list[RealtimeObserver]] = 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 (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

    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 #

callbacks = RealtimeAgentCallbacks()

system_message property #

system_message

Get the system message for the agent.

logger property #

logger

Get the logger for the agent.

realtime_client property #

realtime_client

Get the OpenAI Realtime Client.

registered_realtime_tools property #

registered_realtime_tools

Get the registered realtime tools.

register_observer #

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 
91
92
93
94
95
96
97
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 #

start_observers()
Source code in autogen/agentchat/realtime/experimental/realtime_agent.py 
 99
100
101
102
103
104
105
106
107
async def start_observers(self) -> None:
    for observer in self._observers:
        self._tg.soonify(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 #

run()

Run the agent.

Source code in autogen/agentchat/realtime/experimental/realtime_agent.py 
109
110
111
112
113
114
115
116
117
118
119
120
121
122
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 #

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[[Union[F, Tool]], Tool]

Callable[[Union[F, Tool]], Tool]: The decorator for registering a function.
Source code in autogen/agentchat/realtime/experimental/realtime_agent.py 
124
125
126
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
def register_realtime_function(
    self,
    *,
    name: Optional[str] = None,
    description: Optional[str] = None,
) -> Callable[[Union[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: Union[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