Agent and Agent Runtime

In this and the following section, we focus on the core concepts of AutoGen: agents, agent runtime, messages, and communication. You will not find any AI models or tools here, just the foundational building blocks for building multi-agent applications.

Note

The Core API is designed to be unopinionated and flexible. So at times, you may find it challenging. Continue if you are building an interactive, scalable and distributed multi-agent system and want full control of all workflows. If you just want to get something running quickly, you may take a look at the AgentChat API.

An agent in AutoGen is an entity defined by the base class autogen_core.Agent. It has a unique identifier of the type autogen_core.AgentId, a metadata dictionary of the type autogen_core.AgentMetadata, and method for handling messages autogen_core.BaseAgent.on_message_impl().

An agent runtime is the execution environment for agents in AutoGen. Similar to the runtime environment of a programming language, an agent runtime provides the necessary infrastructure to facilitate communication between agents, manage agent lifecycles, enforce security boundaries, and support monitoring and debugging. For local development, developers can use SingleThreadedAgentRuntime, which can be embedded in a Python application.

Note

Agents are not directly instantiated and managed by application code. Instead, they are created by the runtime when needed and managed by the runtime.

Implementing an Agent

To implement an agent, the developer must subclass the BaseAgent class and implement the on_message_impl() method. This method is invoked when the agent receives a message. For example, the following agent handles a simple message type and prints the message it receives:

from dataclasses import dataclass

from autogen_core import AgentId, BaseAgent, MessageContext


@dataclass
class MyMessageType:
    content: str


class MyAgent(BaseAgent):
    def __init__(self) -> None:
        super().__init__("MyAgent")

    async def on_message_impl(self, message: MyMessageType, ctx: MessageContext) -> None:
        print(f"Received message: {message.content}")  # type: ignore

This agent only handles MyMessageType messages. To handle multiple message types, developers can subclass the RoutedAgent class which provides an easy-to use API to implement different message handlers for different message types. See the next section on message and communication.

Registering Agent Type

To make agents available to the runtime, developers can use the register() class method of the BaseAgent class. The process of registration associates an agent type, which is uniquely identified by a string, and a factory function that creates an instance of the agent type of the given class. The factory function is used to allow automatic creation of agent instances when they are needed.

Agent type (AgentType) is not the same as the agent class. In this example, the agent type is AgentType("my_agent") and the agent class is the Python class MyAgent. The factory function is expected to return an instance of the agent class on which the register() class method is invoked. Read Agent Identity and Lifecycles to learn more about agent type and identity.

Note

Different agent types can be registered with factory functions that return the same agent class. For example, in the factory functions, variations of the constructor parameters can be used to create different instances of the same agent class.

To register an agent type with the SingleThreadedAgentRuntime, the following code can be used:

from autogen_core import SingleThreadedAgentRuntime

runtime = SingleThreadedAgentRuntime()
await MyAgent.register(runtime, "my_agent", lambda: MyAgent())

AgentType(type='my_agent')

Once an agent type is registered, we can send a direct message to an agent instance using an AgentId. The runtime will create the instance the first time it delivers a message to this instance.

agent_id = AgentId("my_agent", "default")
runtime.start()  # Start processing messages in the background.
await runtime.send_message(MyMessageType("Hello, World!"), agent_id)
await runtime.stop()  # Stop processing messages in the background.

Received message: Hello, World!

Note

Because the runtime manages the lifecycle of agents, an AgentId is only used to communicate with the agent or retrieve its metadata (e.g., description).

Running the Single-Threaded Agent Runtime

The above code snippet uses runtime.start() to start a background task to process and deliver messages to recepients’ message handlers. This is a feature of the local embedded runtime SingleThreadedAgentRuntime.

To stop the background task immediately, use the stop() method:

runtime.start()
# ... Send messages, publish messages, etc.
await runtime.stop()  # This will return immediately but will not cancel
# any in-progress message handling.

You can resume the background task by calling start() again.

For batch scenarios such as running benchmarks for evaluating agents, you may want to wait for the background task to stop automatically when there are no unprocessed messages and no agent is handling messages – the batch may considered complete. You can achieve this by using the stop_when_idle() method:

runtime.start()
# ... Send messages, publish messages, etc.
await runtime.stop_when_idle()  # This will block until the runtime is idle.

You can also directly process messages one-by-one without a background task using:

await runtime.process_next()

Other runtime implementations will have their own ways of running the runtime.