autogen_agentchat.agents#

class AssistantAgent(name: str, model_client: ChatCompletionClient, *, tools: List[Tool | Callable[[...], Any] | Callable[[...], Awaitable[Any]]] | None = None, handoffs: List[Handoff | str] | None = None, description: str = 'An agent that provides assistance with ability to use tools.', system_message: str | None = 'You are a helpful AI assistant. Solve tasks using your tools. Reply with TERMINATE when the task has been completed.', reflect_on_tool_use: bool = False, tool_call_summary_format: str = '{result}')[source]#

Bases: BaseChatAgent

An agent that provides assistance with tool use.

The on_messages() returns a Response in which chat_message is the final response message.

The on_messages_stream() creates an async generator that produces the inner messages as they are created, and the Response object as the last item before closing the generator.

Tool call behavior:

If the model returns no tool call, then the response is immediately returned as a TextMessage in chat_message.
When the model returns tool calls, they will be executed right away:
- When reflect_on_tool_use is False (default), the tool call results are returned as a TextMessage in chat_message. tool_call_summary_format can be used to customize the tool call summary.
- When reflect_on_tool_use is True, the another model inference is made using the tool calls and results, and the text response is returned as a TextMessage in chat_message.

Hand off behavior:

If a handoff is triggered, a HandoffMessage will be returned in chat_message.
If there are tool calls, they will also be executed right away before returning the handoff.

Note

The assistant agent is not thread-safe or coroutine-safe. It should not be shared between multiple tasks or coroutines, and it should not call its methods concurrently.

Note

By default, the tool call results are returned as response when tool calls are made. So it is recommended to pay attention to the formatting of the tools return values, especially if another agent is expecting them in a specific format. Use tool_call_summary_format to customize the tool call summary, if needed.

Note

If multiple handoffs are detected, only the first handoff is executed.

Parameters:

name (str) – The name of the agent.
model_client (ChatCompletionClient) – The model client to use for inference.
tools (List[Tool | Callable[..., Any] | Callable[..., Awaitable[Any]]] | None, optional) – The tools to register with the agent.
handoffs (List[HandoffBase | str] | None, optional) – The handoff configurations for the agent, allowing it to transfer to other agents by responding with a HandoffMessage. The transfer is only executed when the team is in Swarm. If a handoff is a string, it should represent the target agent’s name.
description (str, optional) – The description of the agent.
system_message (str, optional) – The system message for the model.
reflect_on_tool_use (bool, optional) – If True, the agent will make another model inference using the tool call and result to generate a response. If False, the tool call result will be returned as the response. Defaults to False.
tool_call_summary_format (str, optional) – The format string used to create a tool call summary for every tool call result. Defaults to “{result}”. When reflect_on_tool_use is False, a concatenation of all the tool call summaries, separated by a new line character (’n’) will be returned as the response. Available variables: {tool_name}, {arguments}, {result}. For example, “{tool_name}: {result}” will create a summary like “tool_name: result”.

Raises:

ValueError – If tool names are not unique.
ValueError – If handoff names are not unique.
ValueError – If handoff names are not unique from tool names.
ValueError – If maximum number of tool iterations is less than 1.

Examples

The following example demonstrates how to create an assistant agent with a model client and generate a response to a simple task.

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


async def main() -> None:
    model_client = OpenAIChatCompletionClient(
        model="gpt-4o",
        # api_key = "your_openai_api_key"
    )
    agent = AssistantAgent(name="assistant", model_client=model_client)

    response = await agent.on_messages(
        [TextMessage(content="What is the capital of France?", source="user")], CancellationToken()
    )
    print(response)


asyncio.run(main())

The following example demonstrates how to create an assistant agent with a model client and a tool, generate a stream of messages for a task, and print the messages to the console.

import asyncio
from autogen_ext.models.openai import OpenAIChatCompletionClient
from autogen_agentchat.agents import AssistantAgent
from autogen_agentchat.messages import TextMessage
from autogen_agentchat.ui import Console
from autogen_core import CancellationToken


async def get_current_time() -> str:
    return "The current time is 12:00 PM."


async def main() -> None:
    model_client = OpenAIChatCompletionClient(
        model="gpt-4o",
        # api_key = "your_openai_api_key"
    )
    agent = AssistantAgent(name="assistant", model_client=model_client, tools=[get_current_time])

    await Console(
        agent.on_messages_stream(
            [TextMessage(content="What is the current time?", source="user")], CancellationToken()
        )
    )


asyncio.run(main())

The following example shows how to use o1-mini model with the assistant agent.

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


async def main() -> None:
    model_client = OpenAIChatCompletionClient(
        model="o1-mini",
        # api_key = "your_openai_api_key"
    )
    # The system message is not supported by the o1 series model.
    agent = AssistantAgent(name="assistant", model_client=model_client, system_message=None)

    response = await agent.on_messages(
        [TextMessage(content="What is the capital of France?", source="user")], CancellationToken()
    )
    print(response)


asyncio.run(main())

Note

The o1-preview and o1-mini models do not support system message and function calling. So the system_message should be set to None and the tools and handoffs should not be set. See o1 beta limitations for more details.

async load_state(state: Mapping[str, Any]) → None[source]#: Load the state of the assistant agent

async on_messages(messages: Sequence[Annotated[TextMessage | MultiModalMessage | StopMessage | HandoffMessage, FieldInfo(annotation=NoneType, required=True, discriminator='type')]], cancellation_token: CancellationToken) → Response[source]#: Handles incoming messages and returns a response.

async on_messages_stream(messages: Sequence[Annotated[TextMessage | MultiModalMessage | StopMessage | HandoffMessage, FieldInfo(annotation=NoneType, required=True, discriminator='type')]], cancellation_token: CancellationToken) → AsyncGenerator[Annotated[TextMessage | MultiModalMessage | StopMessage | HandoffMessage | ToolCallMessage | ToolCallResultMessage, FieldInfo(annotation=NoneType, required=True, discriminator='type')] | Response, None][source]#: Handles incoming messages and returns a stream of messages and and the final item is the response. The base implementation in BaseChatAgent simply calls on_messages() and yields the messages in the response.

async on_reset(cancellation_token: CancellationToken) → None[source]#: Reset the assistant agent to its initialization state.

property produced_message_types: List[type[Annotated[TextMessage | MultiModalMessage | StopMessage | HandoffMessage, FieldInfo(annotation=NoneType, required=True, discriminator='type')]]]#: The types of messages that the assistant agent produces.

async save_state() → Mapping[str, Any][source]#: Save the current state of the assistant agent.

class BaseChatAgent(name: str, description: str)[source]#

Bases: ChatAgent, ABC

Base class for a chat agent.

property description: str#: The description of the agent. This is used by team to make decisions about which agents to use. The description should describe the agent’s capabilities and how to interact with it.

async load_state(state: Mapping[str, Any]) → None[source]#: Restore agent from saved state. Default implementation for stateless agents.

property name: str#: The name of the agent. This is used by team to uniquely identify the agent. It should be unique within the team.

abstract async on_messages(messages: Sequence[Annotated[TextMessage | MultiModalMessage | StopMessage | HandoffMessage, FieldInfo(annotation=NoneType, required=True, discriminator='type')]], cancellation_token: CancellationToken) → Response[source]#: Handles incoming messages and returns a response.

async on_messages_stream(messages: Sequence[Annotated[TextMessage | MultiModalMessage | StopMessage | HandoffMessage, FieldInfo(annotation=NoneType, required=True, discriminator='type')]], cancellation_token: CancellationToken) → AsyncGenerator[Annotated[TextMessage | MultiModalMessage | StopMessage | HandoffMessage | ToolCallMessage | ToolCallResultMessage, FieldInfo(annotation=NoneType, required=True, discriminator='type')] | Response, None][source]#: Handles incoming messages and returns a stream of messages and and the final item is the response. The base implementation in BaseChatAgent simply calls on_messages() and yields the messages in the response.

abstract async on_reset(cancellation_token: CancellationToken) → None[source]#: Resets the agent to its initialization state.

abstract property produced_message_types: List[type[Annotated[TextMessage | MultiModalMessage | StopMessage | HandoffMessage, FieldInfo(annotation=NoneType, required=True, discriminator='type')]]]#: The types of messages that the agent produces.

async run(*, task: str | Annotated[TextMessage | MultiModalMessage | StopMessage | HandoffMessage, FieldInfo(annotation=NoneType, required=True, discriminator='type')] | None = None, cancellation_token: CancellationToken | None = None) → TaskResult[source]#: Run the agent with the given task and return the result.

async run_stream(*, task: str | Annotated[TextMessage | MultiModalMessage | StopMessage | HandoffMessage, FieldInfo(annotation=NoneType, required=True, discriminator='type')] | None = None, cancellation_token: CancellationToken | None = None) → AsyncGenerator[Annotated[TextMessage | MultiModalMessage | StopMessage | HandoffMessage | ToolCallMessage | ToolCallResultMessage, FieldInfo(annotation=NoneType, required=True, discriminator='type')] | TaskResult, None][source]#: Run the agent with the given task and return a stream of messages and the final task result as the last item in the stream.

async save_state() → Mapping[str, Any][source]#: Export state. Default implementation for stateless agents.

class CodeExecutorAgent(name: str, code_executor: CodeExecutor, *, description: str = 'A computer terminal that performs no other action than running Python scripts (provided to it quoted in ```python code blocks), or sh shell scripts (provided to it quoted in ```sh code blocks).')[source]#

Bases: BaseChatAgent

An agent that 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.

import asyncio
from autogen_agentchat.agents import CodeExecutorAgent
from autogen_agentchat.messages import TextMessage
from autogen_ext.code_executors.docker import DockerCommandLineCodeExecutor
from autogen_core import CancellationToken


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()


asyncio.run(run_code_executor_agent())

async on_messages(messages: Sequence[Annotated[TextMessage | MultiModalMessage | StopMessage | HandoffMessage, FieldInfo(annotation=NoneType, required=True, discriminator='type')]], cancellation_token: CancellationToken) → Response[source]#: Handles incoming messages and returns a response.

async on_reset(cancellation_token: CancellationToken) → None[source]#: It it’s a no-op as the code executor agent has no mutable state.

property produced_message_types: List[type[Annotated[TextMessage | MultiModalMessage | StopMessage | HandoffMessage, FieldInfo(annotation=NoneType, required=True, discriminator='type')]]]#: The types of messages that the code executor agent produces.

class CodingAssistantAgent(name: str, model_client: ChatCompletionClient, *, description: str = 'A helpful and general-purpose AI assistant that has strong language skills, Python skills, and Linux command line skills.', system_message: str = 'You are a helpful AI assistant.\nSolve tasks using your coding and language skills.\nIn the following cases, suggest python code (in a python coding block) or shell script (in a sh coding block) for the user to execute.\n 1. When you need to collect info, use the code to output the info you need, for example, browse or search the web, download/read a file, print the content of a webpage or a file, get the current date/time, check the operating system. After sufficient info is printed and the task is ready to be solved based on your language skill, you can solve the task by yourself.\n 2. When you need to perform some task with code, use the code to perform the task and output the result. Finish the task smartly.\nSolve the task step by step if you need to. If a plan is not provided, explain your plan first. Be clear which step uses code, and which step uses your language skill.\nWhen using code, you must indicate the script type in the code block. The user cannot provide any other feedback or perform any other action beyond executing the code you suggest. The user can\'t modify your code. So do not suggest incomplete code which requires users to modify. Don\'t use a code block if it\'s not intended to be executed by the user.\nIf you want the user to save the code in a file before executing it, put # filename: <filename> inside the code block as the first line. Don\'t include multiple code blocks in one response. Do not ask users to copy and paste the result. Instead, use \'print\' function for the output when relevant. Check the execution result returned by the user.\nIf the result indicates there is an error, fix the error and output the code again. Suggest the full code instead of partial code or code changes. If the error can\'t be fixed or if the task is not solved even after the code is executed successfully, analyze the problem, revisit your assumption, collect additional info you need, and think of a different approach to try.\nWhen you find an answer, verify the answer carefully. Include verifiable evidence in your response if possible.\nReply "TERMINATE" in the end when code has been executed and task is complete.')[source]#

Bases: AssistantAgent

[DEPRECATED] An agent that provides coding assistance using an LLM model client.

It responds with a StopMessage when ‘terminate’ is detected in the response.

pydantic model Handoff[source]#

Bases: Handoff

[DEPRECATED] Handoff configuration. Moved to autogen_agentchat.base.Handoff. Will remove in 0.4.0.

Show JSON schema

{
   "title": "Handoff",
   "description": "[DEPRECATED] Handoff configuration. Moved to :class:`autogen_agentchat.base.Handoff`. Will remove in 0.4.0.",
   "type": "object",
   "properties": {
      "target": {
         "title": "Target",
         "type": "string"
      },
      "description": {
         "default": "",
         "title": "Description",
         "type": "string"
      },
      "name": {
         "default": "",
         "title": "Name",
         "type": "string"
      },
      "message": {
         "default": "",
         "title": "Message",
         "type": "string"
      }
   },
   "deprecated": true,
   "required": [
      "target"
   ]
}

Fields:

Validators:

set_defaults » all fields

model_post_init(_Handoff__context: Any) → None[source]#: Override this method to perform additional initialization after __init__ and model_construct. This is useful if you want to do some validation that requires the entire model to be initialized.

class SocietyOfMindAgent(name: str, team: Team, model_client: ChatCompletionClient, *, description: str = 'An agent that uses an inner team of agents to generate responses.', task_prompt: str = '{transcript}\nContinue.', response_prompt: str = 'Here is a transcript of conversation so far:\n{transcript}\n\\Provide a response to the original request.')[source]#

Bases: BaseChatAgent

An agent that uses an inner team of agents to generate responses.

Each time the agent’s on_messages() or on_messages_stream() method is called, it runs the inner team of agents and then uses the model client to generate a response based on the inner team’s messages. Once the response is generated, the agent resets the inner team by calling Team.reset().

Parameters:

name (str) – The name of the agent.
team (Team) – The team of agents to use.
model_client (ChatCompletionClient) – The model client to use for preparing responses.
description (str, optional) – The description of the agent.

Example:

import asyncio
from autogen_agentchat.agents import AssistantAgent, SocietyOfMindAgent
from autogen_ext.models.openai import OpenAIChatCompletionClient
from autogen_agentchat.teams import RoundRobinGroupChat
from autogen_agentchat.conditions import MaxMessageTermination


async def main() -> None:
    model_client = OpenAIChatCompletionClient(model="gpt-4o")

    agent1 = AssistantAgent("assistant1", model_client=model_client, system_message="You are a helpful assistant.")
    agent2 = AssistantAgent("assistant2", model_client=model_client, system_message="You are a helpful assistant.")
    inner_termination = MaxMessageTermination(3)
    inner_team = RoundRobinGroupChat([agent1, agent2], termination_condition=inner_termination)

    society_of_mind_agent = SocietyOfMindAgent("society_of_mind", team=inner_team, model_client=model_client)

    agent3 = AssistantAgent("assistant3", model_client=model_client, system_message="You are a helpful assistant.")
    agent4 = AssistantAgent("assistant4", model_client=model_client, system_message="You are a helpful assistant.")
    outter_termination = MaxMessageTermination(10)
    team = RoundRobinGroupChat([society_of_mind_agent, agent3, agent4], termination_condition=outter_termination)

    stream = team.run_stream(task="Tell me a one-liner joke.")
    async for message in stream:
        print(message)


asyncio.run(main())

async on_messages(messages: Sequence[Annotated[TextMessage | MultiModalMessage | StopMessage | HandoffMessage, FieldInfo(annotation=NoneType, required=True, discriminator='type')]], cancellation_token: CancellationToken) → Response[source]#: Handles incoming messages and returns a response.

async on_messages_stream(messages: Sequence[Annotated[TextMessage | MultiModalMessage | StopMessage | HandoffMessage, FieldInfo(annotation=NoneType, required=True, discriminator='type')]], cancellation_token: CancellationToken) → AsyncGenerator[Annotated[TextMessage | MultiModalMessage | StopMessage | HandoffMessage | ToolCallMessage | ToolCallResultMessage, FieldInfo(annotation=NoneType, required=True, discriminator='type')] | Response, None][source]#: Handles incoming messages and returns a stream of messages and and the final item is the response. The base implementation in BaseChatAgent simply calls on_messages() and yields the messages in the response.

async on_reset(cancellation_token: CancellationToken) → None[source]#: Resets the agent to its initialization state.

property produced_message_types: List[type[Annotated[TextMessage | MultiModalMessage | StopMessage | HandoffMessage, FieldInfo(annotation=NoneType, required=True, discriminator='type')]]]#: The types of messages that the agent produces.

class ToolUseAssistantAgent(name: str, model_client: ChatCompletionClient, registered_tools: List[Tool | Callable[[...], Any] | Callable[[...], Awaitable[Any]]], *, description: str = 'An agent that provides assistance with ability to use tools.', system_message: str = "You are a helpful AI assistant. Solve tasks using your tools. Reply with 'TERMINATE' when the task has been completed.")[source]#

Bases: AssistantAgent

[DEPRECATED] An agent that provides assistance with tool use.

It responds with a StopMessage when ‘terminate’ is detected in the response.

Parameters:

name (str) – The name of the agent.
model_client (ChatCompletionClient) – The model client to use for inference.
registered_tools (List[Tool | Callable[..., Any] | Callable[..., Awaitable[Any]]) – The tools to register with the agent.
description (str, optional) – The description of the agent.
system_message (str, optional) – The system message for the model.

class UserProxyAgent(name: str, *, description: str = 'A human user', input_func: Callable[[str], str] | Callable[[str, CancellationToken | None], Awaitable[str]] | None = None)[source]#

Bases: BaseChatAgent

An agent that can represent a human user through an input function.

This agent can be used to represent a human user in a chat system by providing a custom input function.

Parameters:

name (str) – The name of the agent.
description (str, optional) – A description of the agent.
input_func (Optional[Callable[[str], str]], Callable[[str, Optional[CancellationToken]], Awaitable[str]]) – A function that takes a prompt and returns a user input string.

Note

Using UserProxyAgent puts a running team in a temporary blocked state until the user responds. So it is important to time out the user input function and cancel using the CancellationToken if the user does not respond. The input function should also handle exceptions and return a default response if needed.

For typical use cases that involve slow human responses, it is recommended to use termination conditions such as HandoffTermination or SourceMatchTermination to stop the running team and return the control to the application. You can run the team again with the user input. This way, the state of the team can be saved and restored when the user responds.

See Pause for User Input for more information.

Example

Simple usage case:

import asyncio
from autogen_core import CancellationToken
from autogen_agentchat.agents import UserProxyAgent
from autogen_agentchat.messages import TextMessage


async def simple_user_agent():
    agent = UserProxyAgent("user_proxy")
    response = await asyncio.create_task(
        agent.on_messages(
            [TextMessage(content="What is your name? ", source="user")],
            cancellation_token=CancellationToken(),
        )
    )
    print(f"Your name is {response.chat_message.content}")

Example

Cancellable usage case:

import asyncio
from typing import Any
from autogen_core import CancellationToken
from autogen_agentchat.agents import UserProxyAgent
from autogen_agentchat.messages import TextMessage


token = CancellationToken()
agent = UserProxyAgent("user_proxy")


async def timeout(delay: float):
    await asyncio.sleep(delay)


def cancellation_callback(task: asyncio.Task[Any]):
    token.cancel()


async def cancellable_user_agent():
    try:
        timeout_task = asyncio.create_task(timeout(3))
        timeout_task.add_done_callback(cancellation_callback)
        agent_task = asyncio.create_task(
            agent.on_messages(
                [TextMessage(content="What is your name? ", source="user")],
                cancellation_token=CancellationToken(),
            )
        )
        response = await agent_task
        print(f"Your name is {response.chat_message.content}")
    except Exception as e:
        print(f"Exception: {e}")
    except BaseException as e:
        print(f"BaseException: {e}")

async on_messages(messages: Sequence[Annotated[TextMessage | MultiModalMessage | StopMessage | HandoffMessage, FieldInfo(annotation=NoneType, required=True, discriminator='type')]], cancellation_token: CancellationToken | None = None) → Response[source]#: Handle incoming messages by requesting user input.

async on_reset(cancellation_token: CancellationToken | None = None) → None[source]#: Reset agent state.

property produced_message_types: List[type[Annotated[TextMessage | MultiModalMessage | StopMessage | HandoffMessage, FieldInfo(annotation=NoneType, required=True, discriminator='type')]]]#: Message types this agent can produce.