Agent and Agent Runtime#

In this and the following section, we focus on the core concepts of AutoGen: agents, agent runtime, messages, and communication – the foundational building blocks for an 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 interface Agent. It has a unique identifier of the type AgentId, a metadata dictionary of the type AgentMetadata.

In most cases, you can subclass your agents from higher level class RoutedAgent which enables you to route messages to corresponding message handler specified with message_handler() decorator and proper type hint for the message variable. 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.

If you are already familiar with AgentChat, it is important to note that AgentChat’s agents such as AssistantAgent are created by application and thus not directly managed by the runtime. To use an AgentChat agent in Core, you need to create a wrapper Core agent that delegates messages to the AgentChat agent and let the runtime manage the wrapper agent.

Implementing an Agent#

To implement an agent, the developer must subclass the RoutedAgent class and implement a message handler method for each message type the agent is expected to handle using the message_handler() decorator. For example, the following agent handles a simple message type MyMessageType and prints the message it receives:

from dataclasses import dataclass

from autogen_core import AgentId, MessageContext, RoutedAgent, message_handler


@dataclass
class MyMessageType:
    content: str


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

    @message_handler
    async def handle_my_message_type(self, message: MyMessageType, ctx: MessageContext) -> None:
        print(f"{self.id.type} received message: {message.content}")

This agent only handles MyMessageType and messages will be delivered to handle_my_message_type method. Developers can have multiple message handlers for different message types by using message_handler() decorator and setting the type hint for the message variable in the handler function. You can also leverage python typing union for the message variable in one message handler function if it better suits agent’s logic. See the next section on message and communication.

Using an AgentChat Agent#

If you have an AgentChat agent and want to use it in the Core API, you can create a wrapper RoutedAgent that delegates messages to the AgentChat agent. The following example shows how to create a wrapper agent for the AssistantAgent in AgentChat.

from autogen_agentchat.agents import AssistantAgent
from autogen_agentchat.messages import TextMessage
from autogen_ext.models.openai import OpenAIChatCompletionClient


class MyAssistant(RoutedAgent):
    def __init__(self, name: str) -> None:
        super().__init__(name)
        model_client = OpenAIChatCompletionClient(model="gpt-4o")
        self._delegate = AssistantAgent(name, model_client=model_client)

    @message_handler
    async def handle_my_message_type(self, message: MyMessageType, ctx: MessageContext) -> None:
        print(f"{self.id.type} received message: {message.content}")
        response = await self._delegate.on_messages(
            [TextMessage(content=message.content, source="user")], ctx.cancellation_token
        )
        print(f"{self.id.type} responded: {response.chat_message.content}")

For how to use model client, see the Model Client section.

Since the Core API is unopinionated, you are not required to use the AgentChat API to use the Core API. You can implement your own agents or use another agent framework.

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") or AgentType("my_assistant") and the agent class is the Python class MyAgent or MyAssistantAgent. 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 our agent types with the SingleThreadedAgentRuntime, the following code can be used:

from autogen_core import SingleThreadedAgentRuntime

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

AgentType(type='my_assistant')

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.

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

my_agent received message: Hello, World!
my_assistant received message: Hello, World!
my_assistant responded: Hello! How can I assist you today?

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 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.

To close the runtime and release resources, use the close() method:

await runtime.close()

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

On this page
Edit on GitHub
Show Source