{ "cells": [ { "cell_type": "markdown", "metadata": {}, "source": [ "# Agents\n", "\n", "```{include} ../warning.md\n", "```\n", "\n", "AutoGen AgentChat provides a set of preset Agents, each with variations in how an agent might respond to messages.\n", "All agents share the following attributes and methods:\n", "\n", "- {py:attr}`~autogen_agentchat.agents.BaseChatAgent.name`: The unique name of the agent.\n", "- {py:attr}`~autogen_agentchat.agents.BaseChatAgent.description`: The description of the agent in text.\n", "- {py:meth}`~autogen_agentchat.agents.BaseChatAgent.on_messages`: Send the agent a sequence of {py:class}`~autogen_agentchat.messages.ChatMessage` get a {py:class}`~autogen_agentchat.base.Response`.\n", "- {py:meth}`~autogen_agentchat.agents.BaseChatAgent.on_messages_stream`: Same as {py:meth}`~autogen_agentchat.agents.BaseChatAgent.on_messages` but returns an iterator of {py:class}`~autogen_agentchat.messages.AgentMessage` followed by a {py:class}`~autogen_agentchat.base.Response` as the last item.\n", "- {py:meth}`~autogen_agentchat.agents.BaseChatAgent.on_reset`: Reset the agent to its initial state.\n", "\n", "See {py:mod}`autogen_agentchat.messages` for more information on AgentChat message types.\n", "\n", "\n", "## Assistant Agent\n", "\n", "{py:class}`~autogen_agentchat.agents.AssistantAgent` is a built-in agent that\n", "uses a language model with ability to use tools." ] }, { "cell_type": "code", "execution_count": 11, "metadata": {}, "outputs": [], "source": [ "from autogen_agentchat.agents import AssistantAgent\n", "from autogen_agentchat.messages import TextMessage\n", "from autogen_core.base import CancellationToken\n", "from autogen_ext.models import OpenAIChatCompletionClient\n", "\n", "\n", "# Define a tool that searches the web for information.\n", "async def web_search(query: str) -> str:\n", " \"\"\"Find information on the web\"\"\"\n", " return \"AutoGen is a programming framework for building multi-agent applications.\"\n", "\n", "\n", "# Create an agent that uses the OpenAI GPT-4o model.\n", "model_client = OpenAIChatCompletionClient(\n", " model=\"gpt-4o\",\n", " # api_key=\"YOUR_API_KEY\",\n", ")\n", "agent = AssistantAgent(\n", " name=\"assistant\",\n", " model_client=model_client,\n", " tools=[web_search],\n", " system_message=\"Use tools to solve tasks.\",\n", ")" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "We can call the {py:meth}`~autogen_agentchat.agents.AssistantAgent.on_messages` \n", "method to get the agent to respond to a message." ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "[ToolCallMessage(source='assistant', models_usage=RequestUsage(prompt_tokens=61, completion_tokens=15), content=[FunctionCall(id='call_hqVC7UJUPhKaiJwgVKkg66ak', arguments='{\"query\":\"AutoGen\"}', name='web_search')]), ToolCallResultMessage(source='assistant', models_usage=None, content=[FunctionExecutionResult(content='AutoGen is a programming framework for building multi-agent applications.', call_id='call_hqVC7UJUPhKaiJwgVKkg66ak')])]\n", "source='assistant' models_usage=RequestUsage(prompt_tokens=92, completion_tokens=14) content='AutoGen is a programming framework designed for building multi-agent applications.'\n" ] } ], "source": [ "async def assistant_run() -> None:\n", " response = await agent.on_messages(\n", " [TextMessage(content=\"Find information on AutoGen\", source=\"user\")],\n", " cancellation_token=CancellationToken(),\n", " )\n", " print(response.inner_messages)\n", " print(response.chat_message)\n", "\n", "\n", "# Use asyncio.run(assistant_run()) when running in a script.\n", "await assistant_run()" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "The call to the {py:meth}`~autogen_agentchat.agents.AssistantAgent.on_messages` method\n", "returns a {py:class}`~autogen_agentchat.base.Response`\n", "that contains the agent's final response in the {py:attr}`~autogen_agentchat.base.Response.chat_message` attribute,\n", "as well as a list of inner messages in the {py:attr}`~autogen_agentchat.base.Response.inner_messages` attribute,\n", "which stores the agent's \"thought process\" that led to the final response.\n", "\n", "### Stream Messages\n", "\n", "We can also stream each message as it is generated by the agent by using the\n", "{py:meth}`~autogen_agentchat.agents.AssistantAgent.on_messages_stream` method." ] }, { "cell_type": "code", "execution_count": 12, "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "source='assistant' models_usage=RequestUsage(prompt_tokens=61, completion_tokens=15) content=[FunctionCall(id='call_fXhM4PeZsodhhUOlNiFkoBXF', arguments='{\"query\":\"AutoGen\"}', name='web_search')]\n", "source='assistant' models_usage=None content=[FunctionExecutionResult(content='AutoGen is a programming framework for building multi-agent applications.', call_id='call_fXhM4PeZsodhhUOlNiFkoBXF')]\n", "Response(chat_message=TextMessage(source='assistant', models_usage=RequestUsage(prompt_tokens=92, completion_tokens=31), content='AutoGen is a programming framework designed for building multi-agent applications. If you need more specific information about its features or usage, feel free to ask!'), inner_messages=[ToolCallMessage(source='assistant', models_usage=RequestUsage(prompt_tokens=61, completion_tokens=15), content=[FunctionCall(id='call_fXhM4PeZsodhhUOlNiFkoBXF', arguments='{\"query\":\"AutoGen\"}', name='web_search')]), ToolCallResultMessage(source='assistant', models_usage=None, content=[FunctionExecutionResult(content='AutoGen is a programming framework for building multi-agent applications.', call_id='call_fXhM4PeZsodhhUOlNiFkoBXF')])])\n" ] } ], "source": [ "async def assistant_run_stream() -> None:\n", " async for message in agent.on_messages_stream(\n", " [TextMessage(content=\"Find information on AutoGen\", source=\"user\")],\n", " cancellation_token=CancellationToken(),\n", " ):\n", " print(message)\n", "\n", "\n", "# Use asyncio.run(assistant_run_stream()) when running in a script.\n", "await assistant_run_stream()" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "The {py:meth}`~autogen_agentchat.agents.AssistantAgent.on_messages_stream` method\n", "returns an asynchronous generator that yields each inner message generated by the agent,\n", "and the last item is the final response message in the {py:attr}`~autogen_agentchat.base.Response.chat_message` attribute.\n", "\n", "From the messages, you can see the assistant agent used the `web_search` tool to\n", "search for information and responded using the search results.\n", "\n", "### Understanding Tool Calling\n", "\n", "Large Language Models (LLMs) are typically limited to generating text or code responses. However, many complex tasks benefit from the ability to use external tools that perform specific actions, such as fetching data from APIs or databases.\n", "\n", "To address this limitation, modern LLMs can now accept a list of available tool schemas (descriptions of tools and their arguments) and generate a tool call message. This capability is known as **Tool Calling** or **Function Calling** and is becoming a popular pattern in building intelligent agent-based applications.\n", "\n", "For more information on tool calling, refer to the documentation from [OpenAI](https://platform.openai.com/docs/guides/function-calling) and [Anthropic](https://docs.anthropic.com/en/docs/build-with-claude/tool-use)." ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## CodeExecutorAgent\n", "\n", "The {py:class}`~autogen_agentchat.agents.CodeExecutorAgent`\n", "preset extracts and executes code snippets found in received messages and returns the output. It is typically used within a team with another agent that generates code snippets to be executed.\n", "\n", "```{note}\n", "It is recommended that the {py:class}`~autogen_agentchat.agents.CodeExecutorAgent` agent\n", "uses a Docker container to execute code. This ensures that model-generated code is executed in an isolated environment. To use Docker, your environment must have Docker installed and running. \n", "Follow the installation instructions for [Docker](https://docs.docker.com/get-docker/).\n", "```\n", "\n", "In this example, we show how to set up a {py:class}`~autogen_agentchat.agents.CodeExecutorAgent` agent that uses the \n", "{py:class}`~autogen_ext.code_executors.DockerCommandLineCodeExecutor` \n", "to execute code snippets in a Docker container. The `work_dir` parameter indicates where all executed files are first saved locally before being executed in the Docker container." ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "source='code_executor' models_usage=None content='Hello world\\n'\n" ] } ], "source": [ "from autogen_agentchat.agents import CodeExecutorAgent\n", "from autogen_ext.code_executors import DockerCommandLineCodeExecutor\n", "\n", "\n", "async def run_code_executor_agent() -> None:\n", " # Create a code executor agent that uses a Docker container to execute code.\n", " code_executor = DockerCommandLineCodeExecutor(work_dir=\"coding\")\n", " await code_executor.start()\n", " code_executor_agent = CodeExecutorAgent(\"code_executor\", code_executor=code_executor)\n", "\n", " # Run the agent with a given code snippet.\n", " task = TextMessage(\n", " content=\"\"\"Here is some code\n", "```python\n", "print('Hello world')\n", "```\n", "\"\"\",\n", " source=\"user\",\n", " )\n", " response = await code_executor_agent.on_messages([task], CancellationToken())\n", " print(response.chat_message)\n", "\n", " # Stop the code executor.\n", " await code_executor.stop()\n", "\n", "\n", "# Use asyncio.run(run_code_executor_agent()) when running in a script.\n", "await run_code_executor_agent()" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "This example shows the agent executing a code snippet that prints \"Hello world\".\n", "The agent then returns the output of the code execution." ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Build Your Own Agents\n", "\n", "You may have agents with behaviors that do not fall into a preset. \n", "In such cases, you can build custom agents.\n", "\n", "All agents in AgentChat inherit from {py:class}`~autogen_agentchat.agents.BaseChatAgent` \n", "class and implement the following abstract methods and attributes:\n", "\n", "- {py:meth}`~autogen_agentchat.agents.BaseChatAgent.on_messages`: The abstract method that defines the behavior of the agent in response to messages. This method is called when the agent is asked to provide a response in {py:meth}`~autogen_agentchat.agents.BaseChatAgent.run`. It returns a {py:class}`~autogen_agentchat.base.Response` object.\n", "- {py:meth}`~autogen_agentchat.agents.BaseChatAgent.on_reset`: The abstract method that resets the agent to its initial state. This method is called when the agent is asked to reset itself.\n", "- {py:attr}`~autogen_agentchat.agents.BaseChatAgent.produced_message_types`: The list of possible {py:class}`~autogen_agentchat.messages.ChatMessage` message types the agent can produce in its response.\n", "\n", "Optionally, you can implement the the {py:meth}`~autogen_agentchat.agents.BaseChatAgent.on_messages_stream` method to stream messages as they are generated by the agent. If this method is not implemented, the agent\n", "uses the default implementation of {py:meth}`~autogen_agentchat.agents.BaseChatAgent.on_messages_stream`\n", "that calls the {py:meth}`~autogen_agentchat.agents.BaseChatAgent.on_messages` method and\n", "yields all messages in the response." ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "### CounterDownAgent\n", "\n", "In this example, we create a simple agent that counts down from a given number to zero,\n", "and produces a stream of messages with the current count." ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "3...\n", "2...\n", "1...\n", "Done!\n" ] } ], "source": [ "from typing import AsyncGenerator, List, Sequence\n", "\n", "from autogen_agentchat.agents import BaseChatAgent\n", "from autogen_agentchat.base import Response\n", "from autogen_agentchat.messages import AgentMessage, ChatMessage\n", "\n", "\n", "class CountDownAgent(BaseChatAgent):\n", " def __init__(self, name: str, count: int = 3):\n", " super().__init__(name, \"A simple agent that counts down.\")\n", " self._count = count\n", "\n", " @property\n", " def produced_message_types(self) -> List[type[ChatMessage]]:\n", " return [TextMessage]\n", "\n", " async def on_messages(self, messages: Sequence[ChatMessage], cancellation_token: CancellationToken) -> Response:\n", " # Calls the on_messages_stream.\n", " response: Response | None = None\n", " async for message in self.on_messages_stream(messages, cancellation_token):\n", " if isinstance(message, Response):\n", " response = message\n", " assert response is not None\n", " return response\n", "\n", " async def on_messages_stream(\n", " self, messages: Sequence[ChatMessage], cancellation_token: CancellationToken\n", " ) -> AsyncGenerator[AgentMessage | Response, None]:\n", " inner_messages: List[AgentMessage] = []\n", " for i in range(self._count, 0, -1):\n", " msg = TextMessage(content=f\"{i}...\", source=self.name)\n", " inner_messages.append(msg)\n", " yield msg\n", " # The response is returned at the end of the stream.\n", " # It contains the final message and all the inner messages.\n", " yield Response(chat_message=TextMessage(content=\"Done!\", source=self.name), inner_messages=inner_messages)\n", "\n", " async def on_reset(self, cancellation_token: CancellationToken) -> None:\n", " pass\n", "\n", "\n", "async def run_countdown_agent() -> None:\n", " # Create a countdown agent.\n", " countdown_agent = CountDownAgent(\"countdown\")\n", "\n", " # Run the agent with a given task and stream the response.\n", " async for message in countdown_agent.on_messages_stream([], CancellationToken()):\n", " if isinstance(message, Response):\n", " print(message.chat_message.content)\n", " else:\n", " print(message.content)\n", "\n", "\n", "# Use asyncio.run(run_countdown_agent()) when running in a script.\n", "await run_countdown_agent()" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "### UserProxyAgent \n", "\n", "A common use case for building a custom agent is to create an agent that acts as a proxy for the user.\n", "\n", "In the example below we show how to implement a `UserProxyAgent` - an agent that asks the user to enter\n", "some text through console and then returns that message as a response." ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "I am glad to be here.\n" ] } ], "source": [ "import asyncio\n", "\n", "\n", "class UserProxyAgent(BaseChatAgent):\n", " def __init__(self, name: str) -> None:\n", " super().__init__(name, \"A human user.\")\n", "\n", " @property\n", " def produced_message_types(self) -> List[type[ChatMessage]]:\n", " return [TextMessage]\n", "\n", " async def on_messages(self, messages: Sequence[ChatMessage], cancellation_token: CancellationToken) -> Response:\n", " user_input = await asyncio.get_event_loop().run_in_executor(None, input, \"Enter your response: \")\n", " return Response(chat_message=TextMessage(content=user_input, source=self.name))\n", "\n", " async def on_reset(self, cancellation_token: CancellationToken) -> None:\n", " pass\n", "\n", "\n", "async def run_user_proxy_agent() -> None:\n", " user_proxy_agent = UserProxyAgent(name=\"user_proxy_agent\")\n", " response = await user_proxy_agent.on_messages([], CancellationToken())\n", " print(response.chat_message.content)\n", "\n", "\n", "# Use asyncio.run(run_user_proxy_agent()) when running in a script.\n", "await run_user_proxy_agent()" ] } ], "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 }