autogen_agentchat.conditions#

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

class MaxMessageTermination(max_messages: int, include_agent_event: bool = False)[source]#

Bases: TerminationCondition, Component[MaxMessageTerminationConfig]

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

Parameters:

max_messages – The maximum number of messages allowed in the conversation.
include_agent_event – If True, include BaseAgentEvent in the message count. Otherwise, only include BaseChatMessage. Defaults to False.

component_config_schema#: alias of MaxMessageTerminationConfig

component_provider_override: ClassVar[str | None] = 'autogen_agentchat.conditions.MaxMessageTermination'#: Override the provider string for the component. This should be used to prevent internal module names being a part of the module name.

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

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

_to_config() → MaxMessageTerminationConfig[source]#

Dump the configuration that would be requite to create a new instance of a component matching the configuration of this instance.

Returns:: T – The configuration of the component.

classmethod _from_config(config: MaxMessageTerminationConfig) → Self[source]#

Create a new instance of the component from a configuration object.

Parameters:: config (T) – The configuration object.
Returns:: Self – The new instance of the component.

class TextMentionTermination(text: str, sources: Sequence[str] | None = None)[source]#

Bases: TerminationCondition, Component[TextMentionTerminationConfig]

Terminate the conversation if a specific text is mentioned.

Parameters:

text – The text to look for in the messages.
sources – Check only messages of the specified agents for the text to look for.

component_config_schema#: alias of TextMentionTerminationConfig

component_provider_override: ClassVar[str | None] = 'autogen_agentchat.conditions.TextMentionTermination'#: Override the provider string for the component. This should be used to prevent internal module names being a part of the module name.

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

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

_to_config() → TextMentionTerminationConfig[source]#

Dump the configuration that would be requite to create a new instance of a component matching the configuration of this instance.

Returns:: T – The configuration of the component.

classmethod _from_config(config: TextMentionTerminationConfig) → Self[source]#

Create a new instance of the component from a configuration object.

Parameters:: config (T) – The configuration object.
Returns:: Self – The new instance of the component.

class StopMessageTermination[source]#

Bases: TerminationCondition, Component[StopMessageTerminationConfig]

Terminate the conversation if a StopMessage is received.

component_config_schema#: alias of StopMessageTerminationConfig

component_provider_override: ClassVar[str | None] = 'autogen_agentchat.conditions.StopMessageTermination'#: Override the provider string for the component. This should be used to prevent internal module names being a part of the module name.

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

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

_to_config() → StopMessageTerminationConfig[source]#

Dump the configuration that would be requite to create a new instance of a component matching the configuration of this instance.

Returns:: T – The configuration of the component.

classmethod _from_config(config: StopMessageTerminationConfig) → Self[source]#

Create a new instance of the component from a configuration object.

Parameters:: config (T) – The configuration object.
Returns:: Self – The new instance of the component.

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

Bases: TerminationCondition, Component[TokenUsageTerminationConfig]

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.

component_config_schema#: alias of TokenUsageTerminationConfig

component_provider_override: ClassVar[str | None] = 'autogen_agentchat.conditions.TokenUsageTermination'#: Override the provider string for the component. This should be used to prevent internal module names being a part of the module name.

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

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

_to_config() → TokenUsageTerminationConfig[source]#

Dump the configuration that would be requite to create a new instance of a component matching the configuration of this instance.

Returns:: T – The configuration of the component.

classmethod _from_config(config: TokenUsageTerminationConfig) → Self[source]#

Create a new instance of the component from a configuration object.

Parameters:: config (T) – The configuration object.
Returns:: Self – The new instance of the component.

class HandoffTermination(target: str)[source]#

Bases: TerminationCondition, Component[HandoffTerminationConfig]

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

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

component_config_schema#: alias of HandoffTerminationConfig

component_provider_override: ClassVar[str | None] = 'autogen_agentchat.conditions.HandoffTermination'#: Override the provider string for the component. This should be used to prevent internal module names being a part of the module name.

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

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

_to_config() → HandoffTerminationConfig[source]#

Dump the configuration that would be requite to create a new instance of a component matching the configuration of this instance.

Returns:: T – The configuration of the component.

classmethod _from_config(config: HandoffTerminationConfig) → Self[source]#

Create a new instance of the component from a configuration object.

Parameters:: config (T) – The configuration object.
Returns:: Self – The new instance of the component.

class TimeoutTermination(timeout_seconds: float)[source]#

Bases: TerminationCondition, Component[TimeoutTerminationConfig]

Terminate the conversation after a specified duration has passed.

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

component_config_schema#: alias of TimeoutTerminationConfig

component_provider_override: ClassVar[str | None] = 'autogen_agentchat.conditions.TimeoutTermination'#: Override the provider string for the component. This should be used to prevent internal module names being a part of the module name.

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

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

_to_config() → TimeoutTerminationConfig[source]#

Dump the configuration that would be requite to create a new instance of a component matching the configuration of this instance.

Returns:: T – The configuration of the component.

classmethod _from_config(config: TimeoutTerminationConfig) → Self[source]#

Create a new instance of the component from a configuration object.

Parameters:: config (T) – The configuration object.
Returns:: Self – The new instance of the component.

class ExternalTermination[source]#

Bases: TerminationCondition, Component[ExternalTerminationConfig]

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

component_config_schema#: alias of ExternalTerminationConfig

component_provider_override: ClassVar[str | None] = 'autogen_agentchat.conditions.ExternalTermination'#: Override the provider string for the component. This should be used to prevent internal module names being a part of the module name.

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

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

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

_to_config() → ExternalTerminationConfig[source]#

Dump the configuration that would be requite to create a new instance of a component matching the configuration of this instance.

Returns:: T – The configuration of the component.

classmethod _from_config(config: ExternalTerminationConfig) → Self[source]#

Create a new instance of the component from a configuration object.

Parameters:: config (T) – The configuration object.
Returns:: Self – The new instance of the component.

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

Bases: TerminationCondition, Component[SourceMatchTerminationConfig]

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.

component_config_schema#: alias of SourceMatchTerminationConfig

component_provider_override: ClassVar[str | None] = 'autogen_agentchat.conditions.SourceMatchTermination'#: Override the provider string for the component. This should be used to prevent internal module names being a part of the module name.

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

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

_to_config() → SourceMatchTerminationConfig[source]#

Dump the configuration that would be requite to create a new instance of a component matching the configuration of this instance.

Returns:: T – The configuration of the component.

classmethod _from_config(config: SourceMatchTerminationConfig) → Self[source]#

Create a new instance of the component from a configuration object.

Parameters:: config (T) – The configuration object.
Returns:: Self – The new instance of the component.

class TextMessageTermination(source: str | None = None)[source]#

Bases: TerminationCondition, Component[TextMessageTerminationConfig]

Terminate the conversation if a TextMessage is received.

This termination condition checks for TextMessage instances in the message sequence. When a TextMessage is found, it terminates the conversation if either: - No source was specified (terminates on any TextMessage) - The message source matches the specified source

Parameters:: source (str | None, optional) – The source name to match against incoming messages. If None, matches any source. Defaults to None.

component_config_schema#: alias of TextMessageTerminationConfig

component_provider_override: ClassVar[str | None] = 'autogen_agentchat.conditions.TextMessageTermination'#: Override the provider string for the component. This should be used to prevent internal module names being a part of the module name.

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

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

_to_config() → TextMessageTerminationConfig[source]#

Dump the configuration that would be requite to create a new instance of a component matching the configuration of this instance.

Returns:: T – The configuration of the component.

classmethod _from_config(config: TextMessageTerminationConfig) → Self[source]#

Create a new instance of the component from a configuration object.

Parameters:: config (T) – The configuration object.
Returns:: Self – The new instance of the component.

class FunctionCallTermination(function_name: str)[source]#

Bases: TerminationCondition, Component[FunctionCallTerminationConfig]

Terminate the conversation if a FunctionExecutionResult with a specific name was received.

Parameters:: function_name (str) – The name of the function to look for in the messages.
Raises:: TerminatedException – If the termination condition has already been reached.

component_config_schema#: alias of FunctionCallTerminationConfig

component_provider_override: ClassVar[str | None] = 'autogen_agentchat.conditions.FunctionCallTermination'#: The schema for the component configuration.

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

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

_to_config() → FunctionCallTerminationConfig[source]#

Dump the configuration that would be requite to create a new instance of a component matching the configuration of this instance.

Returns:: T – The configuration of the component.

classmethod _from_config(config: FunctionCallTerminationConfig) → Self[source]#

Create a new instance of the component from a configuration object.

Parameters:: config (T) – The configuration object.
Returns:: Self – The new instance of the component.

class FunctionalTermination(func: Callable[[Sequence[BaseAgentEvent | BaseChatMessage]], bool] | Callable[[Sequence[BaseAgentEvent | BaseChatMessage]], Awaitable[bool]])[source]#

Bases: TerminationCondition

Terminate the conversation if an functional expression is met.

Parameters:: func (Callable[[Sequence[BaseAgentEvent | BaseChatMessage]], bool] | Callable[[Sequence[BaseAgentEvent | BaseChatMessage]], Awaitable[bool]]) – A function that takes a sequence of messages and returns True if the termination condition is met, False otherwise. The function can be a callable or an async callable.

Example

import asyncio
from typing import Sequence

from autogen_agentchat.conditions import FunctionalTermination
from autogen_agentchat.messages import BaseAgentEvent, BaseChatMessage, StopMessage


def expression(messages: Sequence[BaseAgentEvent | BaseChatMessage]) -> bool:
    # Check if the last message is a stop message
    return isinstance(messages[-1], StopMessage)


termination = FunctionalTermination(expression)


async def run() -> None:
    messages = [
        StopMessage(source="agent1", content="Stop"),
    ]
    result = await termination(messages)
    print(result)


asyncio.run(run())

StopMessage(source="FunctionalTermination", content="Functional termination condition met")

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

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