Agents#

Warning

AgentChat is Work in Progress. APIs may change in future releases.

AutoGen AgentChat provides a set of preset Agents, each with variations in how an agent might respond to messages. All agents share the following attributes and methods:

See autogen_agentchat.messages for more information on AgentChat message types.

Assistant Agent#

AssistantAgent is a built-in agent that uses a language model with ability to use tools.

from autogen_agentchat.agents import AssistantAgent
from autogen_agentchat.messages import TextMessage
from autogen_core.base import CancellationToken
from autogen_ext.models import OpenAIChatCompletionClient


# Define a tool that searches the web for information.
async def web_search(query: str) -> str:
    """Find information on the web"""
    return "AutoGen is a programming framework for building multi-agent applications."


# Create an agent that uses the OpenAI GPT-4o model.
model_client = OpenAIChatCompletionClient(
    model="gpt-4o",
    # api_key="YOUR_API_KEY",
)
agent = AssistantAgent(
    name="assistant",
    model_client=model_client,
    tools=[web_search],
    system_message="Use tools to solve tasks.",
)

We can call the on_messages() method to get the agent to respond to a message.

async def assistant_run() -> None:
    response = await agent.on_messages(
        [TextMessage(content="Find information on AutoGen", source="user")],
        cancellation_token=CancellationToken(),
    )
    print(response.inner_messages)
    print(response.chat_message)


# Use asyncio.run(assistant_run()) when running in a script.
await assistant_run()

[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')])]
source='assistant' models_usage=RequestUsage(prompt_tokens=92, completion_tokens=14) content='AutoGen is a programming framework designed for building multi-agent applications.'

The call to the on_messages() method returns a Response that contains the agent’s final response in the chat_message attribute, as well as a list of inner messages in the inner_messages attribute, which stores the agent’s “thought process” that led to the final response.

Stream Messages#

We can also stream each message as it is generated by the agent by using the on_messages_stream() method.

async def assistant_run_stream() -> None:
    async for message in agent.on_messages_stream(
        [TextMessage(content="Find information on AutoGen", source="user")],
        cancellation_token=CancellationToken(),
    ):
        print(message)


# Use asyncio.run(assistant_run_stream()) when running in a script.
await assistant_run_stream()

source='assistant' models_usage=RequestUsage(prompt_tokens=61, completion_tokens=15) content=[FunctionCall(id='call_fXhM4PeZsodhhUOlNiFkoBXF', arguments='{"query":"AutoGen"}', name='web_search')]
source='assistant' models_usage=None content=[FunctionExecutionResult(content='AutoGen is a programming framework for building multi-agent applications.', call_id='call_fXhM4PeZsodhhUOlNiFkoBXF')]
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')])])

The on_messages_stream() method returns an asynchronous generator that yields each inner message generated by the agent, and the last item is the final response message in the chat_message attribute.

From the messages, you can see the assistant agent used the web_search tool to search for information and responded using the search results.

Understanding Tool Calling#

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.

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.

For more information on tool calling, refer to the documentation from OpenAI and Anthropic.

CodeExecutorAgent#

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

Note

It is recommended that the CodeExecutorAgent agent 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. Follow the installation instructions for Docker.

In this example, we show how to set up a CodeExecutorAgent agent that uses the DockerCommandLineCodeExecutor 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.

from autogen_agentchat.agents import CodeExecutorAgent
from autogen_ext.code_executors import DockerCommandLineCodeExecutor


async def run_code_executor_agent() -> None:
    # Create a code executor agent that uses a Docker container to execute code.
    code_executor = DockerCommandLineCodeExecutor(work_dir="coding")
    await code_executor.start()
    code_executor_agent = CodeExecutorAgent("code_executor", code_executor=code_executor)

    # Run the agent with a given code snippet.
    task = TextMessage(
        content="""Here is some code
```python
print('Hello world')
```
""",
        source="user",
    )
    response = await code_executor_agent.on_messages([task], CancellationToken())
    print(response.chat_message)

    # Stop the code executor.
    await code_executor.stop()


# Use asyncio.run(run_code_executor_agent()) when running in a script.
await run_code_executor_agent()

source='code_executor' models_usage=None content='Hello world\n'

This example shows the agent executing a code snippet that prints “Hello world”. The agent then returns the output of the code execution.

Build Your Own Agents#

You may have agents with behaviors that do not fall into a preset. In such cases, you can build custom agents.

All agents in AgentChat inherit from BaseChatAgent class and implement the following abstract methods and attributes:

  • 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 run(). It returns a Response object.

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

  • produced_message_types: The list of possible ChatMessage message types the agent can produce in its response.

Optionally, you can implement the the on_messages_stream() method to stream messages as they are generated by the agent. If this method is not implemented, the agent uses the default implementation of on_messages_stream() that calls the on_messages() method and yields all messages in the response.

CounterDownAgent#

In this example, we create a simple agent that counts down from a given number to zero, and produces a stream of messages with the current count.

from typing import AsyncGenerator, List, Sequence

from autogen_agentchat.agents import BaseChatAgent
from autogen_agentchat.base import Response
from autogen_agentchat.messages import AgentMessage, ChatMessage


class CountDownAgent(BaseChatAgent):
    def __init__(self, name: str, count: int = 3):
        super().__init__(name, "A simple agent that counts down.")
        self._count = count

    @property
    def produced_message_types(self) -> List[type[ChatMessage]]:
        return [TextMessage]

    async def on_messages(self, messages: Sequence[ChatMessage], cancellation_token: CancellationToken) -> Response:
        # Calls the on_messages_stream.
        response: Response | None = None
        async for message in self.on_messages_stream(messages, cancellation_token):
            if isinstance(message, Response):
                response = message
        assert response is not None
        return response

    async def on_messages_stream(
        self, messages: Sequence[ChatMessage], cancellation_token: CancellationToken
    ) -> AsyncGenerator[AgentMessage | Response, None]:
        inner_messages: List[AgentMessage] = []
        for i in range(self._count, 0, -1):
            msg = TextMessage(content=f"{i}...", source=self.name)
            inner_messages.append(msg)
            yield msg
        # The response is returned at the end of the stream.
        # It contains the final message and all the inner messages.
        yield Response(chat_message=TextMessage(content="Done!", source=self.name), inner_messages=inner_messages)

    async def on_reset(self, cancellation_token: CancellationToken) -> None:
        pass


async def run_countdown_agent() -> None:
    # Create a countdown agent.
    countdown_agent = CountDownAgent("countdown")

    # Run the agent with a given task and stream the response.
    async for message in countdown_agent.on_messages_stream([], CancellationToken()):
        if isinstance(message, Response):
            print(message.chat_message.content)
        else:
            print(message.content)


# Use asyncio.run(run_countdown_agent()) when running in a script.
await run_countdown_agent()

3...
2...
1...
Done!

UserProxyAgent#

A common use case for building a custom agent is to create an agent that acts as a proxy for the user.

In the example below we show how to implement a UserProxyAgent - an agent that asks the user to enter some text through console and then returns that message as a response.

import asyncio


class UserProxyAgent(BaseChatAgent):
    def __init__(self, name: str) -> None:
        super().__init__(name, "A human user.")

    @property
    def produced_message_types(self) -> List[type[ChatMessage]]:
        return [TextMessage]

    async def on_messages(self, messages: Sequence[ChatMessage], cancellation_token: CancellationToken) -> Response:
        user_input = await asyncio.get_event_loop().run_in_executor(None, input, "Enter your response: ")
        return Response(chat_message=TextMessage(content=user_input, source=self.name))

    async def on_reset(self, cancellation_token: CancellationToken) -> None:
        pass


async def run_user_proxy_agent() -> None:
    user_proxy_agent = UserProxyAgent(name="user_proxy_agent")
    response = await user_proxy_agent.on_messages([], CancellationToken())
    print(response.chat_message.content)


# Use asyncio.run(run_user_proxy_agent()) when running in a script.
await run_user_proxy_agent()

I am glad to be here.
On this page
Edit on GitHub
Show Source