{
"cells": [
{
"cell_type": "markdown",
"metadata": {},
"source": [
"# Agent and Agent Runtime\n",
"\n",
"In this and the following section, we focus on the core concepts of AutoGen:\n",
"agents, agent runtime, messages, and communication -- \n",
"the foundational building blocks for an multi-agent applications.\n",
"\n",
"```{note}\n",
"The Core API is designed to be unopinionated and flexible. So at times, you\n",
"may find it challenging. Continue if you are building\n",
"an interactive, scalable and distributed multi-agent system and want full control\n",
"of all workflows.\n",
"If you just want to get something running\n",
"quickly, you may take a look at the [AgentChat API](../../agentchat-user-guide/index.md).\n",
"```"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"An agent in AutoGen is an entity defined by the base interface {py:class}`~autogen_core.Agent`.\n",
"It has a unique identifier of the type {py:class}`~autogen_core.AgentId`,\n",
"a metadata dictionary of the type {py:class}`~autogen_core.AgentMetadata`.\n",
"\n",
"In most cases, you can subclass your agents from higher level class {py:class}`~autogen_core.RoutedAgent` which enables you to route messages to corresponding message handler specified with {py:meth}`~autogen_core.message_handler` decorator and proper type hint for the `message` variable.\n",
"An agent runtime is the execution environment for agents in AutoGen.\n",
"\n",
"Similar to the runtime environment of a programming language,\n",
"an agent runtime provides the necessary infrastructure to facilitate communication\n",
"between agents, manage agent lifecycles, enforce security boundaries, and support monitoring and\n",
"debugging.\n",
"\n",
"For local development, developers can use {py:class}`~autogen_core.SingleThreadedAgentRuntime`,\n",
"which can be embedded in a Python application.\n",
"\n",
"```{note}\n",
"Agents are not directly instantiated and managed by application code.\n",
"Instead, they are created by the runtime when needed and managed by the runtime.\n",
"\n",
"If you are already familiar with [AgentChat](../../agentchat-user-guide/index.md),\n",
"it is important to note that AgentChat's agents such as\n",
"{py:class}`~autogen_agentchat.agents.AssistantAgent` are created by application \n",
"and thus not directly managed by the runtime. To use an AgentChat agent in Core,\n",
"you need to create a wrapper Core agent that delegates messages to the AgentChat agent\n",
"and let the runtime manage the wrapper agent.\n",
"```"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"## Implementing an Agent\n",
"\n",
"To implement an agent, the developer must subclass the {py:class}`~autogen_core.RoutedAgent` class\n",
"and implement a message handler method for each message type the agent is expected to handle using\n",
"the {py:meth}`~autogen_core.message_handler` decorator.\n",
"For example,\n",
"the following agent handles a simple message type `MyMessageType` and prints the message it receives:"
]
},
{
"cell_type": "code",
"execution_count": 11,
"metadata": {},
"outputs": [],
"source": [
"from dataclasses import dataclass\n",
"\n",
"from autogen_core import AgentId, MessageContext, RoutedAgent, message_handler\n",
"\n",
"\n",
"@dataclass\n",
"class MyMessageType:\n",
" content: str\n",
"\n",
"\n",
"class MyAgent(RoutedAgent):\n",
" def __init__(self) -> None:\n",
" super().__init__(\"MyAgent\")\n",
"\n",
" @message_handler\n",
" async def handle_my_message_type(self, message: MyMessageType, ctx: MessageContext) -> None:\n",
" print(f\"{self.id.type} received message: {message.content}\")"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"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 {py:meth}`~autogen_core.message_handler` decorator and setting the type hint for the `message` variable in the handler function. You can also leverage [python typing union](https://docs.python.org/3/library/typing.html#typing.Union) for the `message` variable in one message handler function if it better suits agent's logic.\n",
"See the next section on [message and communication](./message-and-communication.ipynb)."
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"## Using an AgentChat Agent\n",
"\n",
"If you have an [AgentChat](../../agentchat-user-guide/index.md) agent and want to use it in the Core API, you can create\n",
"a wrapper {py:class}`~autogen_core.RoutedAgent` that delegates messages to the AgentChat agent.\n",
"The following example shows how to create a wrapper agent for the {py:class}`~autogen_agentchat.agents.AssistantAgent`\n",
"in AgentChat."
]
},
{
"cell_type": "code",
"execution_count": 12,
"metadata": {},
"outputs": [],
"source": [
"from autogen_agentchat.agents import AssistantAgent\n",
"from autogen_agentchat.messages import TextMessage\n",
"from autogen_ext.models.openai import OpenAIChatCompletionClient\n",
"\n",
"\n",
"class MyAssistant(RoutedAgent):\n",
" def __init__(self, name: str) -> None:\n",
" super().__init__(name)\n",
" model_client = OpenAIChatCompletionClient(model=\"gpt-4o\")\n",
" self._delegate = AssistantAgent(name, model_client=model_client)\n",
"\n",
" @message_handler\n",
" async def handle_my_message_type(self, message: MyMessageType, ctx: MessageContext) -> None:\n",
" print(f\"{self.id.type} received message: {message.content}\")\n",
" response = await self._delegate.on_messages(\n",
" [TextMessage(content=message.content, source=\"user\")], ctx.cancellation_token\n",
" )\n",
" print(f\"{self.id.type} responded: {response.chat_message.content}\")"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"For how to use model client, see the [Model Client](./model-clients.ipynb) section.\n",
"\n",
"Since the Core API is unopinionated,\n",
"you are not required to use the AgentChat API to use the Core API.\n",
"You can implement your own agents or use another agent framework."
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"## Registering Agent Type\n",
"\n",
"To make agents available to the runtime, developers can use the\n",
"{py:meth}`~autogen_core.BaseAgent.register` class method of the\n",
"{py:class}`~autogen_core.BaseAgent` class.\n",
"The process of registration associates an agent type, which is uniquely identified by a string, \n",
"and a factory function\n",
"that creates an instance of the agent type of the given class.\n",
"The factory function is used to allow automatic creation of agent instances \n",
"when they are needed.\n",
"\n",
"Agent type ({py:class}`~autogen_core.AgentType`) is not the same as the agent class. In this example,\n",
"the agent type is `AgentType(\"my_agent\")` or `AgentType(\"my_assistant\")` and the agent class is the Python class `MyAgent` or `MyAssistantAgent`.\n",
"The factory function is expected to return an instance of the agent class \n",
"on which the {py:meth}`~autogen_core.BaseAgent.register` class method is invoked.\n",
"Read [Agent Identity and Lifecycles](../core-concepts/agent-identity-and-lifecycle.md)\n",
"to learn more about agent type and identity.\n",
"\n",
"```{note}\n",
"Different agent types can be registered with factory functions that return \n",
"the same agent class. For example, in the factory functions, \n",
"variations of the constructor parameters\n",
"can be used to create different instances of the same agent class.\n",
"```\n",
"\n",
"To register our agent types with the \n",
"{py:class}`~autogen_core.SingleThreadedAgentRuntime`,\n",
"the following code can be used:"
]
},
{
"cell_type": "code",
"execution_count": 13,
"metadata": {},
"outputs": [
{
"data": {
"text/plain": [
"AgentType(type='my_assistant')"
]
},
"execution_count": 13,
"metadata": {},
"output_type": "execute_result"
}
],
"source": [
"from autogen_core import SingleThreadedAgentRuntime\n",
"\n",
"runtime = SingleThreadedAgentRuntime()\n",
"await MyAgent.register(runtime, \"my_agent\", lambda: MyAgent())\n",
"await MyAssistant.register(runtime, \"my_assistant\", lambda: MyAssistant(\"my_assistant\"))"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"Once an agent type is registered, we can send a direct message to an agent instance\n",
"using an {py:class}`~autogen_core.AgentId`.\n",
"The runtime will create the instance the first time it delivers a\n",
"message to this instance."
]
},
{
"cell_type": "code",
"execution_count": 14,
"metadata": {},
"outputs": [
{
"name": "stdout",
"output_type": "stream",
"text": [
"my_agent received message: Hello, World!\n",
"my_assistant received message: Hello, World!\n",
"my_assistant responded: Hello! How can I assist you today?\n"
]
}
],
"source": [
"runtime.start() # Start processing messages in the background.\n",
"await runtime.send_message(MyMessageType(\"Hello, World!\"), AgentId(\"my_agent\", \"default\"))\n",
"await runtime.send_message(MyMessageType(\"Hello, World!\"), AgentId(\"my_assistant\", \"default\"))\n",
"await runtime.stop() # Stop processing messages in the background."
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"```{note}\n",
"Because the runtime manages the lifecycle of agents, an {py:class}`~autogen_core.AgentId`\n",
"is only used to communicate with the agent or retrieve its metadata (e.g., description).\n",
"```"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"## Running the Single-Threaded Agent Runtime\n",
"\n",
"The above code snippet uses {py:meth}`~autogen_core.SingleThreadedAgentRuntime.start` to start a background task\n",
"to process and deliver messages to recepients' message handlers.\n",
"This is a feature of the\n",
"local embedded runtime {py:class}`~autogen_core.SingleThreadedAgentRuntime`.\n",
"\n",
"To stop the background task immediately, use the {py:meth}`~autogen_core.SingleThreadedAgentRuntime.stop` method:"
]
},
{
"cell_type": "code",
"execution_count": 15,
"metadata": {},
"outputs": [],
"source": [
"runtime.start()\n",
"# ... Send messages, publish messages, etc.\n",
"await runtime.stop() # This will return immediately but will not cancel\n",
"# any in-progress message handling."
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"You can resume the background task by calling {py:meth}`~autogen_core.SingleThreadedAgentRuntime.start` again.\n",
"\n",
"For batch scenarios such as running benchmarks for evaluating agents,\n",
"you may want to wait for the background task to stop automatically when\n",
"there are no unprocessed messages and no agent is handling messages --\n",
"the batch may considered complete.\n",
"You can achieve this by using the {py:meth}`~autogen_core.SingleThreadedAgentRuntime.stop_when_idle` method:"
]
},
{
"cell_type": "code",
"execution_count": 16,
"metadata": {},
"outputs": [],
"source": [
"runtime.start()\n",
"# ... Send messages, publish messages, etc.\n",
"await runtime.stop_when_idle() # This will block until the runtime is idle."
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"To close the runtime and release resources, use the {py:meth}`~autogen_core.SingleThreadedAgentRuntime.close` method:"
]
},
{
"cell_type": "code",
"execution_count": 17,
"metadata": {},
"outputs": [],
"source": [
"await runtime.close()"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"Other runtime implementations will have their own ways of running the runtime."
]
}
],
"metadata": {
"kernelspec": {
"display_name": ".venv",
"language": "python",
"name": "python3"
},
"language_info": {
"codemirror_mode": {
"name": "ipython",
"version": 3
},
"file_extension": ".py",
"mimetype": "text/x-python",
"name": "python",
"nbconvert_exporter": "python",
"pygments_lexer": "ipython3",
"version": "3.11.5"
}
},
"nbformat": 4,
"nbformat_minor": 2
}