autogen_agentchat.base#

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

async load_state(state: Mapping[str, Any]) → None[source]#: Restore agent from saved state

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[Annotated[TextMessage | MultiModalMessage | StopMessage | HandoffMessage, FieldInfo(annotation=NoneType, required=True, discriminator='type')]], cancellation_token: CancellationToken) → Response[source]#: Handles incoming messages and returns a response.

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 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[Annotated[TextMessage | MultiModalMessage | StopMessage | HandoffMessage, FieldInfo(annotation=NoneType, required=True, discriminator='type')]]]#: The types of messages that the agent produces.

async save_state() → Mapping[str, Any][source]#: Save agent state for later restoration

pydantic model Handoff[source]#

Bases: BaseModel

Handoff configuration.

Show JSON schema

{
   "title": "Handoff",
   "description": "Handoff configuration.",
   "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"
      }
   },
   "required": [
      "target"
   ]
}

Fields:

description (str)
message (str)
name (str)
target (str)

Validators:

set_defaults » all fields

field description: str = ''#

The description of the handoff such as the condition under which it should happen and the target agent’s ability. If not provided, it is generated from the target agent’s name.

Validated by:

set_defaults

field message: str = ''#

The message to the target agent. If not provided, it is generated from the target agent’s name.

Validated by:

set_defaults

field name: str = ''#

The name of this handoff configuration. If not provided, it is generated from the target agent’s name.

Validated by:

set_defaults

field target: str [Required]#

The name of the target agent to handoff to.

Validated by:

set_defaults

validator set_defaults » all fields[source]#

property handoff_tool: Tool#: Create a handoff tool from this handoff configuration.

Bases: object

A response from calling ChatAgent.on_messages().

chat_message: Annotated[TextMessage | MultiModalMessage | StopMessage | HandoffMessage, FieldInfo(annotation=NoneType, required=True, discriminator='type')]#: A chat message produced by the agent as the response.

inner_messages: List[Annotated[TextMessage | MultiModalMessage | StopMessage | HandoffMessage | ToolCallMessage | ToolCallResultMessage, FieldInfo(annotation=NoneType, required=True, discriminator='type')]] | None = None#: Inner messages produced by the agent.

Bases: object

Result of running a task.

messages: Sequence[Annotated[TextMessage | MultiModalMessage | StopMessage | HandoffMessage | ToolCallMessage | ToolCallResultMessage, FieldInfo(annotation=NoneType, required=True, discriminator='type')]]#: Messages produced by the task.

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

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

Bases: Protocol

A task runner.

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 Team(*args, **kwargs)[source]#

Bases: TaskRunner, Protocol

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

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

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

exception TerminatedException[source]#: Bases: BaseException

class 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.conditions import MaxMessageTermination, TextMentionTermination


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

    # Terminate the conversation after 10 turns and if the text "TERMINATE" is mentioned.
    cond2 = MaxMessageTermination(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