终止条件组合模式

终止条件组合模式

学习目标

本章要解决什么问题：AutoGen 如何通过简洁的 API 实现复杂的多条件终止逻辑。你将学到：

• TerminationCondition 抽象接口
• 内置终止条件的类型
• AND/OR 运算符组合的实现
• 外部终止与竞争条件处理

项目实践

TerminationCondition 抽象

class TerminationCondition(ABC, ComponentBase[BaseModel]):
    component_type = "termination"

    @property
    @abstractmethod
    def terminated(self) -> bool:
        """是否已达到终止条件"""

    @abstractmethod
    async def __call__(
        self, messages: Sequence[BaseAgentEvent | BaseChatMessage]
    ) -> StopMessage | None:
        """检查是否应终止，返回 StopMessage 或 None"""

    @abstractmethod
    async def reset(self) -> None:
        """重置状态以便复用"""

关键设计：

• TerminationCondition 是可调用对象（callable），直接通过 condition(messages) 调用
• 返回 StopMessage 表示终止，包含终止原因
• 返回 None 表示继续
• 一旦终止，必须调用 reset() 才能再次使用

内置终止条件

条件	触发规则	适用场景
MaxMessageTermination(n)	消息数达到 n	防止无限对话
TextMentionTermination(text)	消息中包含指定文本	用户说 “结束” 或 “TERMINATE”
FunctionCallTermination()	包含特定函数调用	工具执行完毕后终止
ExternalTermination()	外部调用 set() 方法	GUI 点击停止按钮
HandoffTermination(target)	发生指向特定目标的 handoff	控制权转移时终止
SourceMatchTermination(sources)	消息来自特定 Agent	某个 Agent 完成工作后

AND/OR 运算符组合

AutoGen 通过运算符重载实现了优雅的条件组合：

from autogen_agentchat.conditions import MaxMessageTermination, TextMentionTermination

# OR：达到 10 条消息 OR 提到 "TERMINATE"
cond1 = MaxMessageTermination(10) | TextMentionTermination("TERMINATE")

# AND：达到 10 条消息 AND 提到 "TERMINATE"
cond2 = MaxMessageTermination(10) & TextMentionTermination("TERMINATE")

# 组合使用
cond3 = (MaxMessageTermination(20) | TextMentionTermination("STOP")) & ExternalTermination()

内部实现：__or__ 和 __and__ 方法返回组合对象（_OrTermination / _AndTermination），调用时分别检查子条件。

实际使用

from autogen_agentchat.conditions import (
    MaxMessageTermination,
    TextMentionTermination,
    ExternalTermination,
)

termination = (
    MaxMessageTermination(20)
    | TextMentionTermination("TERMINATE")
    | ExternalTermination()
)

team = RoundRobinGroupChat(
    [agent_a, agent_b],
    termination_condition=termination,
)

问题与规避

组合条件的短路行为

陷阱：OR 组合中，第一个子条件满足后即终止，后续子条件不会被检查。

规避：理解 OR 和 AND 的语义差异。如果需要”任一条件满足即终止”用 OR，如果需要”所有条件都满足才终止”用 AND。

外部终止的竞争条件

陷阱：ExternalTermination 的 set() 方法可能在消息处理间隙被调用，存在错过检查的风险。

规避：

• ExternalTermination 在每次 __call__ 时检查状态
• 如果已 set()，下次消息到达时立即终止

重置的必要性

陷阱：终止后未调用 reset() 就再次使用，条件始终返回 terminated = True。

规避：

• 每次 team.run() 之前创建新的终止条件实例
• 或在每次使用后调用 reset()

设计取舍

为什么用运算符重载而非方法调用？

**运算符重载（& / |）**的优势：

• 语法简洁：cond1 | cond2 比 TerminationCondition.or_(cond1, cond2) 更易读
• 直观：| 天然表示”或”的逻辑关系
• 可链式组合：a | b | c & d

方法调用的优势：

• 对不熟悉运算符重载的用户更友好
• IDE 自动补全支持更好

替代方案

方案	可读性	灵活性	类型安全	示例
运算符重载	高（数学直觉）	高	中	AutoGen
工厂方法	中	高	高	自定义组合类
DSL（如 YAML）	低（学习成本）	中	低	LangChain
回调函数	高	最高	最低	自定义 lambda

参考来源

• 源码验证：autogen-agentchat/base/_termination.py（终止条件抽象）
• 源码验证：autogen-agentchat/conditions/（内置实现）