autogen_agentchat.base#

class autogen_agentchat.base.ChatAgent(*args, **kwargs)[source]#

Bases: TaskRunner, Protocol

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

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.

async on_messages(messages: Sequence[TextMessage | MultiModalMessage | StopMessage | HandoffMessage], cancellation_token: CancellationToken) → Response[source]#: Handles incoming messages and returns a response.

on_messages_stream(messages: Sequence[TextMessage | MultiModalMessage | StopMessage | HandoffMessage], cancellation_token: CancellationToken) → AsyncGenerator[TextMessage | MultiModalMessage | StopMessage | HandoffMessage | ToolCallMessage | ToolCallResultMessage | Response, None][source]#: Handles incoming messages and returns a stream of inner messages and and the final item is the response.

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

property produced_message_types: List[type[TextMessage | MultiModalMessage | StopMessage | HandoffMessage]]#: The types of messages that the agent produces.

Bases: object

A response from calling ChatAgent.on_messages().

chat_message: TextMessage | MultiModalMessage | StopMessage | HandoffMessage#: A chat message produced by the agent as the response.

inner_messages: List[TextMessage | MultiModalMessage | StopMessage | HandoffMessage | ToolCallMessage | ToolCallResultMessage] | None = None#: Inner messages produced by the agent.

Bases: object

Result of running a task.

messages: Sequence[TextMessage | MultiModalMessage | StopMessage | HandoffMessage | ToolCallMessage | ToolCallResultMessage]#: Messages produced by the task.

stop_reason: str | None = None#: The reason the task stopped.

class autogen_agentchat.base.TaskRunner(*args, **kwargs)[source]#

Bases: Protocol

A task runner.

async run(*, task: str | TextMessage | MultiModalMessage | None = None, cancellation_token: CancellationToken | None = None) → TaskResult[source]#

Run the task and return the result.

The runner is stateful and a subsequent call to this method will continue from where the previous call left off. If the task is not specified, the runner will continue with the current task.

Run the task and produces a stream of messages and the final result TaskResult as the last item in the stream.

The runner is stateful and a subsequent call to this method will continue from where the previous call left off. If the task is not specified, the runner will continue with the current task.

class autogen_agentchat.base.Team(*args, **kwargs)[source]#

Bases: TaskRunner, Protocol

async reset() → None[source]#: Reset the team and all its participants to its initial state.

exception autogen_agentchat.base.TerminatedException[source]#: Bases: BaseException

class autogen_agentchat.base.TerminationCondition[source]#

Bases: ABC

A stateful condition that determines when a conversation should be terminated.

A termination condition is a callable that takes a sequence of ChatMessage objects since the last time the condition was called, and returns a StopMessage if the conversation should be terminated, or None otherwise. Once a termination condition has been reached, it must be reset before it can be used again.

Termination conditions can be combined using the AND and OR operators.

Example

import asyncio
from autogen_agentchat.teams import MaxTurnsTermination, TextMentionTermination


async def main() -> None:
    # Terminate the conversation after 10 turns or if the text "TERMINATE" is mentioned.
    cond1 = MaxTurnsTermination(10) | TextMentionTermination("TERMINATE")

    # Terminate the conversation after 10 turns and if the text "TERMINATE" is mentioned.
    cond2 = MaxTurnsTermination(10) & TextMentionTermination("TERMINATE")

    # ...

    # Reset the termination condition.
    await cond1.reset()
    await cond2.reset()


asyncio.run(main())

abstract async reset() → None[source]#: Reset the termination condition.

abstract property terminated: bool#: Check if the termination condition has been reached