autogen_agentchat.conditions#

该模块为控制多代理团队的行为提供了各种终止条件。

class ExternalTermination[source]#

基类: 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()
classmethod _from_config(config: ExternalTerminationConfig) Self[source]#

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

参数:

config (T) – 配置对象。

返回:

Self – 组件的新实例。

_to_config() ExternalTerminationConfig[source]#

导出配置,该配置是创建与此实例的配置匹配的组件的新实例所需的。

返回:

T – 组件的配置。

component_config_schema#

别名 ExternalTerminationConfig

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

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

async reset() None[source]#

重置终止条件。

set() None[source]#

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

property terminated: bool#

检查是否已达到终止条件

class FunctionCallTermination(function_name: str)[source]#

基类: TerminationCondition, Component[FunctionCallTerminationConfig]

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

参数:

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

引发:

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

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

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

参数:

config (T) – 配置对象。

返回:

Self – 组件的新实例。

_to_config() FunctionCallTerminationConfig[source]#

导出配置,该配置是创建与此实例的配置匹配的组件的新实例所需的。

返回:

T – 组件的配置。

component_config_schema#

别名 FunctionCallTerminationConfig

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

组件配置的模式。

async reset() None[source]#

重置终止条件。

property terminated: bool#

检查是否已达到终止条件

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

基类: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")
async reset() None[source]#

重置终止条件。

property terminated: bool#

检查是否已达到终止条件

class HandoffTermination(target: str)[source]#

基类:TerminationCondition, Component[HandoffTerminationConfig]

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

参数:

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

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

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

参数:

config (T) – 配置对象。

返回:

Self – 组件的新实例。

_to_config() HandoffTerminationConfig[source]#

导出配置,该配置是创建与此实例的配置匹配的组件的新实例所需的。

返回:

T – 组件的配置。

component_config_schema#

HandoffTerminationConfig 的别名

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

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

async reset() None[source]#

重置终止条件。

property terminated: bool#

检查是否已达到终止条件

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.

参数:
  • 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.

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

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

参数:

config (T) – 配置对象。

返回:

Self – 组件的新实例。

_to_config() MaxMessageTerminationConfig[source]#

导出配置,该配置是创建与此实例的配置匹配的组件的新实例所需的。

返回:

T – 组件的配置。

component_config_schema#

alias of MaxMessageTerminationConfig

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

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

async reset() None[source]#

重置终止条件。

property terminated: bool#

检查是否已达到终止条件

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

Bases: TerminationCondition, Component[SourceMatchTerminationConfig]

Terminate the conversation after a specific source responds.

参数:

sources (List[str]) – List of source names to terminate the conversation.

引发:

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

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

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

参数:

config (T) – 配置对象。

返回:

Self – 组件的新实例。

_to_config() SourceMatchTerminationConfig[source]#

导出配置,该配置是创建与此实例的配置匹配的组件的新实例所需的。

返回:

T – 组件的配置。

component_config_schema#

alias of SourceMatchTerminationConfig

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

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

async reset() None[source]#

重置终止条件。

property terminated: bool#

检查是否已达到终止条件

class StopMessageTermination[source]#

基类: TerminationCondition, Component[StopMessageTerminationConfig]

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

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

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

参数:

config (T) – 配置对象。

返回:

Self – 组件的新实例。

_to_config() StopMessageTerminationConfig[source]#

导出配置,该配置是创建与此实例的配置匹配的组件的新实例所需的。

返回:

T – 组件的配置。

component_config_schema#

别名: StopMessageTerminationConfig

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

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

async reset() None[source]#

重置终止条件。

property terminated: bool#

检查是否已达到终止条件

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

基类: TerminationCondition, Component[TextMentionTerminationConfig]

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

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

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

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

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

参数:

config (T) – 配置对象。

返回:

Self – 组件的新实例。

_to_config() TextMentionTerminationConfig[source]#

导出配置,该配置是创建与此实例的配置匹配的组件的新实例所需的。

返回:

T – 组件的配置。

component_config_schema#

别名: TextMentionTerminationConfig

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

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

async reset() None[source]#

重置终止条件。

property terminated: bool#

检查是否已达到终止条件

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

基类: TerminationCondition, Component[TextMessageTerminationConfig]

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

此终止条件检查消息序列中的 TextMessage 实例。 当找到 TextMessage 时,如果满足以下任一条件,它将终止对话:- 未指定来源(终止于任何 TextMessage)- 消息来源与指定的来源匹配

参数:

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

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

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

参数:

config (T) – 配置对象。

返回:

Self – 组件的新实例。

_to_config() TextMessageTerminationConfig[source]#

导出配置,该配置是创建与此实例的配置匹配的组件的新实例所需的。

返回:

T – 组件的配置。

component_config_schema#

别名: TextMessageTerminationConfig

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

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

async reset() None[source]#

重置终止条件。

property terminated: bool#

检查是否已达到终止条件

class TimeoutTermination(timeout_seconds: float)[source]#

基类: TerminationCondition, Component[TimeoutTerminationConfig]

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

参数:

timeout_seconds – 终止对话前的最大持续时间(秒)。

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

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

参数:

config (T) – 配置对象。

返回:

Self – 组件的新实例。

_to_config() TimeoutTerminationConfig[source]#

导出配置,该配置是创建与此实例的配置匹配的组件的新实例所需的。

返回:

T – 组件的配置。

component_config_schema#

别名:TimeoutTerminationConfig

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

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

async reset() None[source]#

重置终止条件。

property terminated: bool#

检查是否已达到终止条件

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

基类: TerminationCondition, Component[TokenUsageTerminationConfig]

如果达到 token 使用限制,则终止对话。

参数:
  • max_total_token – 对话中允许的最大 token 总数。

  • max_prompt_token – 对话中允许的最大 prompt token 数。

  • max_completion_token – 对话中允许的最大 completion token 数。

引发:

ValueError – 如果未提供 max_total_token、max_prompt_token 或 max_completion_token 中的任何一个。

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

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

参数:

config (T) – 配置对象。

返回:

Self – 组件的新实例。

_to_config() TokenUsageTerminationConfig[source]#

导出配置,该配置是创建与此实例的配置匹配的组件的新实例所需的。

返回:

T – 组件的配置。

component_config_schema#

别名:TokenUsageTerminationConfig

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

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

async reset() None[source]#

重置终止条件。

property terminated: bool#

检查是否已达到终止条件