autogen_agentchat.conditions#

This module provides various termination conditions for controlling the behavior of multi-agent teams.

class ExternalTermination[source]#

Bases: TerminationCondition

A termination condition that is externally controlled by calling the set() method.

Example:

from autogen_agentchat.conditions import ExternalTermination

termination = ExternalTermination()

# Run the team in an asyncio task.
...

# Set the termination condition externally
termination.set()

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

set() → None[source]#: Set the termination condition to terminated.

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

class HandoffTermination(target: str)[source]#

Bases: TerminationCondition

Terminate the conversation if a HandoffMessage with the given target is received.

Parameters:: target (str) – The target of the handoff message.

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

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

class MaxMessageTermination(max_messages: int)[source]#

Bases: TerminationCondition

Terminate the conversation after a maximum number of messages have been exchanged.

Parameters:: max_messages – The maximum number of messages allowed in the conversation.

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

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

class SourceMatchTermination(sources: List[str])[source]#

Bases: TerminationCondition

Terminate the conversation after a specific source responds.

Parameters:: sources (List[str]) – List of source names to terminate the conversation.
Raises:: TerminatedException – If the termination condition has already been reached.

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

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

class StopMessageTermination[source]#

Bases: TerminationCondition

Terminate the conversation if a StopMessage is received.

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

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

class TextMentionTermination(text: str)[source]#

Bases: TerminationCondition

Terminate the conversation if a specific text is mentioned.

Parameters:: text – The text to look for in the messages.

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

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

class TimeoutTermination(timeout_seconds: float)[source]#

Bases: TerminationCondition

Terminate the conversation after a specified duration has passed.

Parameters:: timeout_seconds – The maximum duration in seconds before terminating the conversation.

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

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

class TokenUsageTermination(max_total_token: int | None = None, max_prompt_token: int | None = None, max_completion_token: int | None = None)[source]#

Bases: TerminationCondition

Terminate the conversation if a token usage limit is reached.

Parameters:

max_total_token – The maximum total number of tokens allowed in the conversation.
max_prompt_token – The maximum number of prompt tokens allowed in the conversation.
max_completion_token – The maximum number of completion tokens allowed in the conversation.

Raises:

ValueError – If none of max_total_token, max_prompt_token, or max_completion_token is provided.

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

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