autogen_agentchat.conditions#

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

class ExternalTermination[源代码]#

基类：TerminationCondition, Component[ExternalTerminationConfig]

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

Example:

from autogen_agentchat.conditions import ExternalTermination

termination = ExternalTermination()

# 在 asyncio 任务中运行团队
...

# 从外部设置终止条件
termination.set()

classmethod _from_config(config: ExternalTerminationConfig) → Self[源代码]#

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

参数:: config (T) -- 配置对象。
Returns:: Self -- 组件的新实例。

_to_config() → ExternalTerminationConfig[源代码]#

导出当前组件实例的配置，该配置可用于创建具有相同配置的新组件实例。

Returns:: T -- 组件的配置。

component_config_schema#: ExternalTerminationConfig 的别名

component_provider_override: ClassVar[str | None] = 'autogen_agentchat.conditions.ExternalTermination'#: 覆盖组件的provider字符串。这应该用于防止内部模块名称成为模块名称的一部分。

async reset() → None[源代码]#: 重置终止条件。

set() → None[源代码]#: 将终止条件设置为已终止。

property terminated: bool#: 检查是否已达到终止条件

class FunctionCallTermination(function_name: str)[源代码]#

基类：TerminationCondition, Component[FunctionCallTerminationConfig]

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

参数:: function_name (str) -- 要在消息中查找的函数名称。
抛出:: TerminatedException -- 如果终止条件已满足。

classmethod _from_config(config: FunctionCallTerminationConfig) → Self[源代码]#

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

参数:: config (T) -- 配置对象。
Returns:: Self -- 组件的新实例。

_to_config() → FunctionCallTerminationConfig[源代码]#

导出当前组件实例的配置，该配置可用于创建具有相同配置的新组件实例。

Returns:: T -- 组件的配置。

component_config_schema#: FunctionCallTerminationConfig 的别名

component_provider_override: ClassVar[str | None] = 'autogen_agentchat.conditions.FunctionCallTermination'#: 组件配置的架构。

async reset() → None[源代码]#: 重置终止条件。

property terminated: bool#: 检查是否已达到终止条件

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:
    # 检查最后一条消息是否为停止消息
    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[源代码]#: 重置终止条件。

property terminated: bool#: 检查是否已达到终止条件

class HandoffTermination(target: str)[源代码]#

基类：TerminationCondition, Component[HandoffTerminationConfig]

当接收到带有指定目标的:class:`~autogen_agentchat.messages.HandoffMessage`时终止对话。

参数:: target (str) -- 交接消息的目标对象。

classmethod _from_config(config: HandoffTerminationConfig) → Self[源代码]#

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

参数:: config (T) -- 配置对象。
Returns:: Self -- 组件的新实例。

_to_config() → HandoffTerminationConfig[源代码]#

导出当前组件实例的配置，该配置可用于创建具有相同配置的新组件实例。

Returns:: T -- 组件的配置。

component_config_schema#: HandoffTerminationConfig 的别名

component_provider_override: ClassVar[str | None] = 'autogen_agentchat.conditions.HandoffTermination'#: 覆盖组件的provider字符串。这应该用于防止内部模块名称成为模块名称的一部分。

async reset() → None[源代码]#: 重置终止条件。

property terminated: bool#: 检查是否已达到终止条件

class MaxMessageTermination(max_messages: int, include_agent_event: bool = False)[源代码]#

基类：TerminationCondition, Component[MaxMessageTerminationConfig]

在交换达到最大消息数量后终止对话。

参数:

max_messages -- 对话中允许的最大消息数量。
include_agent_event -- 如果为 True，将 BaseAgentEvent 计入消息计数。否则仅计算 BaseChatMessage。默认为 False。

classmethod _from_config(config: MaxMessageTerminationConfig) → Self[源代码]#

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

参数:: config (T) -- 配置对象。
Returns:: Self -- 组件的新实例。

_to_config() → MaxMessageTerminationConfig[源代码]#

导出当前组件实例的配置，该配置可用于创建具有相同配置的新组件实例。

Returns:: T -- 组件的配置。

component_config_schema#: MaxMessageTerminationConfig 的别名

component_provider_override: ClassVar[str | None] = 'autogen_agentchat.conditions.MaxMessageTermination'#: 覆盖组件的provider字符串。这应该用于防止内部模块名称成为模块名称的一部分。

async reset() → None[源代码]#: 重置终止条件。

property terminated: bool#: 检查是否已达到终止条件

class SourceMatchTermination(sources: List[str])[源代码]#

基类：TerminationCondition, Component[SourceMatchTerminationConfig]

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

参数:: sources (List[str]) -- 用于终止对话的来源名称列表。
抛出:: TerminatedException -- 如果终止条件已被触发。

classmethod _from_config(config: SourceMatchTerminationConfig) → Self[源代码]#

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

参数:: config (T) -- 配置对象。
Returns:: Self -- 组件的新实例。

_to_config() → SourceMatchTerminationConfig[源代码]#

导出当前组件实例的配置，该配置可用于创建具有相同配置的新组件实例。

Returns:: T -- 组件的配置。

component_config_schema#: SourceMatchTerminationConfig 的别名

component_provider_override: ClassVar[str | None] = 'autogen_agentchat.conditions.SourceMatchTermination'#: 覆盖组件的provider字符串。这应该用于防止内部模块名称成为模块名称的一部分。

async reset() → None[源代码]#: 重置终止条件。

property terminated: bool#: 检查是否已达到终止条件

class StopMessageTermination[源代码]#

基类：TerminationCondition, Component[StopMessageTerminationConfig]

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

classmethod _from_config(config: StopMessageTerminationConfig) → Self[源代码]#

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

参数:: config (T) -- 配置对象。
Returns:: Self -- 组件的新实例。

_to_config() → StopMessageTerminationConfig[源代码]#

导出当前组件实例的配置，该配置可用于创建具有相同配置的新组件实例。

Returns:: T -- 组件的配置。

component_config_schema#: StopMessageTerminationConfig 的别名

component_provider_override: ClassVar[str | None] = 'autogen_agentchat.conditions.StopMessageTermination'#: 覆盖组件的provider字符串。这应该用于防止内部模块名称成为模块名称的一部分。

async reset() → None[源代码]#: 重置终止条件。

property terminated: bool#: 检查是否已达到终止条件

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

基类：TerminationCondition, Component[TextMentionTerminationConfig]

如果消息中提到特定文本则终止对话。

参数:

text -- 需要在消息中查找的文本。
sources -- 仅在指定代理的消息中检查目标文本。

classmethod _from_config(config: TextMentionTerminationConfig) → Self[源代码]#

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

参数:: config (T) -- 配置对象。
Returns:: Self -- 组件的新实例。

_to_config() → TextMentionTerminationConfig[源代码]#

导出当前组件实例的配置，该配置可用于创建具有相同配置的新组件实例。

Returns:: T -- 组件的配置。

component_config_schema#: TextMentionTerminationConfig 的别名

component_provider_override: ClassVar[str | None] = 'autogen_agentchat.conditions.TextMentionTermination'#: 覆盖组件的provider字符串。这应该用于防止内部模块名称成为模块名称的一部分。

async reset() → None[源代码]#: 重置终止条件。

property terminated: bool#: 检查是否已达到终止条件

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

基类：TerminationCondition, Component[TextMessageTerminationConfig]

如果收到 TextMessage 则终止对话。

此终止条件会检查消息序列中的 TextMessage 实例。当发现 TextMessage 时，在以下任一情况下会终止对话： - 未指定来源（任何 TextMessage 都会触发终止） - 消息来源与指定的来源匹配

参数:: source (str | None, optional) -- 用于匹配传入消息的来源名称。若为 None 则匹配所有来源。默认为 None。

classmethod _from_config(config: TextMessageTerminationConfig) → Self[源代码]#

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

参数:: config (T) -- 配置对象。
Returns:: Self -- 组件的新实例。

_to_config() → TextMessageTerminationConfig[源代码]#

导出当前组件实例的配置，该配置可用于创建具有相同配置的新组件实例。

Returns:: T -- 组件的配置。

component_config_schema#: TextMessageTerminationConfig 的别名

component_provider_override: ClassVar[str | None] = 'autogen_agentchat.conditions.TextMessageTermination'#: 覆盖组件的provider字符串。这应该用于防止内部模块名称成为模块名称的一部分。

async reset() → None[源代码]#: 重置终止条件。

property terminated: bool#: 检查是否已达到终止条件

class TimeoutTermination(timeout_seconds: float)[源代码]#

基类：TerminationCondition, Component[TimeoutTerminationConfig]

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

参数:: timeout_seconds -- 终止对话前的最大持续时间（秒）。

classmethod _from_config(config: TimeoutTerminationConfig) → Self[源代码]#

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

参数:: config (T) -- 配置对象。
Returns:: Self -- 组件的新实例。

_to_config() → TimeoutTerminationConfig[源代码]#

导出当前组件实例的配置，该配置可用于创建具有相同配置的新组件实例。

Returns:: T -- 组件的配置。

component_config_schema#: TimeoutTerminationConfig 的别名

component_provider_override: ClassVar[str | None] = 'autogen_agentchat.conditions.TimeoutTermination'#: 覆盖组件的provider字符串。这应该用于防止内部模块名称成为模块名称的一部分。

async reset() → None[源代码]#: 重置终止条件。

property terminated: bool#: 检查是否已达到终止条件

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中的任何一个。

classmethod _from_config(config: TokenUsageTerminationConfig) → Self[源代码]#

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

参数:: config (T) -- 配置对象。
Returns:: Self -- 组件的新实例。

_to_config() → TokenUsageTerminationConfig[源代码]#

导出当前组件实例的配置，该配置可用于创建具有相同配置的新组件实例。

Returns:: T -- 组件的配置。

component_config_schema#: TokenUsageTerminationConfig 的别名

component_provider_override: ClassVar[str | None] = 'autogen_agentchat.conditions.TokenUsageTermination'#: 覆盖组件的provider字符串。这应该用于防止内部模块名称成为模块名称的一部分。

async reset() → None[源代码]#: 重置终止条件。

property terminated: bool#: 检查是否已达到终止条件