chatflow.conditions

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

class agentopera.chatflow.conditions.MaxMessageTermination(max_messages: int, include_agent_event: bool = False)[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.
include_agent_event – If True, include AgentEvent in the message count. Otherwise, only include ChatMessage. Defaults to False.

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

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

class agentopera.chatflow.conditions.TextMentionTermination(text: str, sources: Sequence[str] | None = None)[source]

Bases: TerminationCondition

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.

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

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

class agentopera.chatflow.conditions.StopMessageTermination[source]

Bases: TerminationCondition

Terminate the conversation if a StopMessage is received.

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

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

class agentopera.chatflow.conditions.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.

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

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

class agentopera.chatflow.conditions.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.

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

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

class agentopera.chatflow.conditions.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.

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

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

class agentopera.chatflow.conditions.ExternalTermination[source]

Bases: TerminationCondition

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

Example:

from agentopera.chatflow.conditions import ExternalTermination

termination = ExternalTermination()

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

# Set the termination condition externally
termination.set()

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.

class agentopera.chatflow.conditions.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.

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

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

class agentopera.chatflow.conditions.TextMessageTermination(source: str | None = None)[source]

Bases: TerminationCondition

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.

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

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

class agentopera.chatflow.conditions.FunctionCallTermination(function_name: str)[source]

Bases: TerminationCondition

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.

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

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