autogen_agentchat.conditions

此模块提供各种终止条件，用于控制多代理团队的行为。

class MaxMessageTermination(max_messages: int, include_agent_event: bool = False)

基类：TerminationCondition, Component[MaxMessageTerminationConfig]

在交换了最大消息数后终止对话。

参数:

• max_messages – 对话中允许的最大消息数。

• include_agent_event – 如果为 True，则在消息计数中包含 BaseAgentEvent。否则，只包含 BaseChatMessage。默认为 False。

component_config_schema

别名 MaxMessageTerminationConfig

component_provider_override: ClassVar[str | None] = 'autogen_agentchat.conditions.MaxMessageTermination'

覆盖组件的提供者字符串。这应该用于防止内部模块名称成为模块名称的一部分。

property terminated: bool

检查终止条件是否已达到

async reset() → None

重置终止条件。

_to_config() → MaxMessageTerminationConfig

转储创建与此实例配置匹配的组件新实例所需的配置。

返回:

T – 组件的配置。

classmethod _from_config(config: MaxMessageTerminationConfig) → Self

从配置对象创建组件的新实例。

参数:

config (T) – 配置对象。

返回:

Self – 组件的新实例。

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

基类：TerminationCondition, Component[TextMentionTerminationConfig]

如果提到了特定文本，则终止对话。

参数:

• text – 要在消息中查找的文本。

• sources – 仅检查指定代理的消息中是否存在要查找的文本。

component_config_schema

别名 TextMentionTerminationConfig

component_provider_override: ClassVar[str | None] = 'autogen_agentchat.conditions.TextMentionTermination'

覆盖组件的提供者字符串。这应该用于防止内部模块名称成为模块名称的一部分。

property terminated: bool

检查终止条件是否已达到

async reset() → None

重置终止条件。

_to_config() → TextMentionTerminationConfig

转储创建与此实例配置匹配的组件新实例所需的配置。

返回:

T – 组件的配置。

classmethod _from_config(config: TextMentionTerminationConfig) → Self

从配置对象创建组件的新实例。

参数:

config (T) – 配置对象。

返回:

Self – 组件的新实例。

class StopMessageTermination

基类：TerminationCondition, Component[StopMessageTerminationConfig]

如果收到 StopMessage，则终止对话。

component_config_schema

别名 StopMessageTerminationConfig

component_provider_override: ClassVar[str | None] = 'autogen_agentchat.conditions.StopMessageTermination'

覆盖组件的提供者字符串。这应该用于防止内部模块名称成为模块名称的一部分。

property terminated: bool

检查终止条件是否已达到

async reset() → None

重置终止条件。

_to_config() → StopMessageTerminationConfig

转储创建与此实例配置匹配的组件新实例所需的配置。

返回:

T – 组件的配置。

classmethod _from_config(config: StopMessageTerminationConfig) → Self

从配置对象创建组件的新实例。

参数:

config (T) – 配置对象。

返回:

Self – 组件的新实例。

class TokenUsageTermination(max_total_token: int | None = None, max_prompt_token: int | None = None, max_completion_token: int | None = None)

基类：TerminationCondition, Component[TokenUsageTerminationConfig]

如果达到令牌使用限制，则终止对话。

参数:

• max_total_token – 对话中允许的最大总令牌数。

• max_prompt_token – 对话中允许的最大提示令牌数。

• max_completion_token – 对话中允许的最大完成令牌数。

抛出:

ValueError – 如果未提供 max_total_token、max_prompt_token 或 max_completion_token。

component_config_schema

别名 TokenUsageTerminationConfig

component_provider_override: ClassVar[str | None] = 'autogen_agentchat.conditions.TokenUsageTermination'

覆盖组件的提供者字符串。这应该用于防止内部模块名称成为模块名称的一部分。

property terminated: bool

检查终止条件是否已达到

async reset() → None

重置终止条件。

_to_config() → TokenUsageTerminationConfig

转储创建与此实例配置匹配的组件新实例所需的配置。

返回:

T – 组件的配置。

classmethod _from_config(config: TokenUsageTerminationConfig) → Self

从配置对象创建组件的新实例。

参数:

config (T) – 配置对象。

返回:

Self – 组件的新实例。

class HandoffTermination(target: str)

基类：TerminationCondition, Component[HandoffTerminationConfig]

如果收到带有给定目标的 HandoffMessage，则终止对话。

参数:

target (str) – 移交消息的目标。

component_config_schema

别名 HandoffTerminationConfig

component_provider_override: ClassVar[str | None] = 'autogen_agentchat.conditions.HandoffTermination'

覆盖组件的提供者字符串。这应该用于防止内部模块名称成为模块名称的一部分。

property terminated: bool

检查终止条件是否已达到

async reset() → None

重置终止条件。

_to_config() → HandoffTerminationConfig

转储创建与此实例配置匹配的组件新实例所需的配置。

返回:

T – 组件的配置。

classmethod _from_config(config: HandoffTerminationConfig) → Self

从配置对象创建组件的新实例。

参数:

config (T) – 配置对象。

返回:

Self – 组件的新实例。

class TimeoutTermination(timeout_seconds: float)

基类：TerminationCondition, Component[TimeoutTerminationConfig]

在指定持续时间过后终止对话。

参数:

timeout_seconds – 终止对话前的最长持续时间（秒）。

component_config_schema

别名 TimeoutTerminationConfig

component_provider_override: ClassVar[str | None] = 'autogen_agentchat.conditions.TimeoutTermination'

覆盖组件的提供者字符串。这应该用于防止内部模块名称成为模块名称的一部分。

property terminated: bool

检查终止条件是否已达到

async reset() → None

重置终止条件。

_to_config() → TimeoutTerminationConfig

转储创建与此实例配置匹配的组件新实例所需的配置。

返回:

T – 组件的配置。

classmethod _from_config(config: TimeoutTerminationConfig) → Self

从配置对象创建组件的新实例。

参数:

config (T) – 配置对象。

返回:

Self – 组件的新实例。

class ExternalTermination

基类：TerminationCondition, Component[ExternalTerminationConfig]

一种通过调用 set() 方法进行外部控制的终止条件。

示例

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

别名 ExternalTerminationConfig

component_provider_override: ClassVar[str | None] = 'autogen_agentchat.conditions.ExternalTermination'

覆盖组件的提供者字符串。这应该用于防止内部模块名称成为模块名称的一部分。

property terminated: bool

检查终止条件是否已达到

set() → None

将终止条件设置为已终止。

async reset() → None

重置终止条件。

_to_config() → ExternalTerminationConfig

转储创建与此实例配置匹配的组件新实例所需的配置。

返回:

T – 组件的配置。

classmethod _from_config(config: ExternalTerminationConfig) → Self

从配置对象创建组件的新实例。

参数:

config (T) – 配置对象。

返回:

Self – 组件的新实例。

class SourceMatchTermination(sources: List[str])

基类：TerminationCondition, Component[SourceMatchTerminationConfig]

在特定来源响应后终止对话。

参数:

sources (List[str]) – 用于终止对话的来源名称列表。

抛出:

TerminatedException – 如果终止条件已达到。

component_config_schema

别名 SourceMatchTerminationConfig

component_provider_override: ClassVar[str | None] = 'autogen_agentchat.conditions.SourceMatchTermination'

覆盖组件的提供者字符串。这应该用于防止内部模块名称成为模块名称的一部分。

property terminated: bool

检查终止条件是否已达到

async reset() → None

重置终止条件。

_to_config() → SourceMatchTerminationConfig

转储创建与此实例配置匹配的组件新实例所需的配置。

返回:

T – 组件的配置。

classmethod _from_config(config: SourceMatchTerminationConfig) → Self

从配置对象创建组件的新实例。

参数:

config (T) – 配置对象。

返回:

Self – 组件的新实例。

class TextMessageTermination(source: str | None = None)

基类：TerminationCondition, Component[TextMessageTerminationConfig]

如果收到 TextMessage，则终止对话。

此终止条件检查消息序列中的 TextMessage 实例。当找到 TextMessage 时，如果满足以下任一条件，它将终止对话：- 未指定来源（在任何 TextMessage 上终止）- 消息来源与指定来源匹配

参数:

source (str | None, optional) – 要与传入消息匹配的来源名称。如果为 None，则匹配任何来源。默认为 None。

component_config_schema

别名 TextMessageTerminationConfig

component_provider_override: ClassVar[str | None] = 'autogen_agentchat.conditions.TextMessageTermination'

覆盖组件的提供者字符串。这应该用于防止内部模块名称成为模块名称的一部分。

property terminated: bool

检查终止条件是否已达到

async reset() → None

重置终止条件。

_to_config() → TextMessageTerminationConfig

转储创建与此实例配置匹配的组件新实例所需的配置。

返回:

T – 组件的配置。

classmethod _from_config(config: TextMessageTerminationConfig) → Self

从配置对象创建组件的新实例。

参数:

config (T) – 配置对象。

返回:

Self – 组件的新实例。

class FunctionCallTermination(function_name: str)

基类：TerminationCondition, Component[FunctionCallTerminationConfig]

如果收到具有特定名称的 FunctionExecutionResult，则终止对话。

参数:

function_name (str) – 要在消息中查找的函数名称。

抛出:

TerminatedException – 如果终止条件已达到。

component_config_schema

别名 FunctionCallTerminationConfig

component_provider_override: ClassVar[str | None] = 'autogen_agentchat.conditions.FunctionCallTermination'

组件配置的 schema。

property terminated: bool

检查终止条件是否已达到

async reset() → None

重置终止条件。

_to_config() → FunctionCallTerminationConfig

转储创建与此实例配置匹配的组件新实例所需的配置。

返回:

T – 组件的配置。

classmethod _from_config(config: FunctionCallTerminationConfig) → Self

从配置对象创建组件的新实例。

参数:

config (T) – 配置对象。

返回:

Self – 组件的新实例。

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

基类：TerminationCondition

如果满足功能表达式，则终止对话。

参数:

func (Callable[[Sequence[BaseAgentEvent | BaseChatMessage]], bool] | Callable[[Sequence[BaseAgentEvent | BaseChatMessage]], Awaitable[bool]]) – 一个函数，它接受一个消息序列，如果满足终止条件则返回 True，否则返回 False。该函数可以是可调用对象或异步可调用对象。

示例

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

检查终止条件是否已达到

async reset() → None

重置终止条件。