终止条件

控制 Agent 何时停止执行

---

📖 概述

终止条件（Termination Condition）用于判断 Agent 或团队是否应该停止执行。

---

1. 内置终止条件

MaxMessageTermination

达到最大消息数时终止。

from autogen_agentchat.conditions import MaxMessageTermination

termination = MaxMessageTermination(max_messages=10)

TextMentionTermination

消息中包含特定文本时终止。

from autogen_agentchat.conditions import TextMentionTermination

termination = TextMentionTermination(termination_text="完成任务")

TimeoutTermination

超时终止。

from autogen_agentchat.conditions import TimeoutTermination

termination = TimeoutTermination(timeout_seconds=60)

ExternalTermination

外部控制终止。

from autogen_agentchat.conditions import ExternalTermination

termination = ExternalTermination()

# 外部调用
termination.set_termination()

---

2. 组合条件

OrTermination (或)

满足任一条件即终止。

from autogen_agentchat.conditions import OrTermination

termination = OrTermination([
    MaxMessageTermination(10),
    TextMentionTermination("完成"),
    TimeoutTermination(60)
])

AndTermination (与)

满足所有条件才终止。

from autogen_agentchat.conditions import AndTermination

termination = AndTermination([
    MaxMessageTermination(5),
    TextMentionTermination("完成")
])

---

3. 自定义终止条件

from autogen_agentchat.base import TerminationCondition
from autogen_agentchat.messages import BaseAgentEvent, BaseChatMessage
from typing import List, Optional

class CustomTermination(TerminationCondition):
    def __init__(self, keyword: str):
        self.keyword = keyword
        self.messages: List[str] = []

    @property
    def terminated(self) -> bool:
        # 检查是否有消息包含关键词
        return any(self.keyword in msg for msg in self.messages)

    async def check(self, messages: List[BaseAgentEvent | BaseChatMessage]) -> Optional[str]:
        for msg in messages:
            if hasattr(msg, 'content') and self.keyword in msg.content:
                return f"包含关键词 '{self.keyword}'，终止"
            self.messages.append(msg.content if hasattr(msg, 'content') else str(msg))
        return None

    def reset(self) -> None:
        self.messages = []

---

4. 使用示例

单个条件

from autogen_agentchat.teams import RoundRobinGroupChat
from autogen_agentchat.conditions import MaxMessageTermination

team = RoundRobinGroupChat(
    [agent1, agent2],
    termination_condition=MaxMessageTermination(10)
)

组合条件

from autogen_agentchat.conditions import OrTermination

team = RoundRobinGroupChat(
    [agent1, agent2],
    termination_condition=OrTermination([
        MaxMessageTermination(10),
        TextMentionTermination("任务完成"),
        TimeoutTermination(300)  # 5 分钟超时
    ])
)

---

5. 检查终止

class TerminationCondition:
    @property
    def terminated(self) -> bool:
        """是否已终止"""
        pass

    async def check(self, messages: List[Message]) -> Optional[str]:
        """检查是否应该终止，返回终止原因或 None"""
        pass

    def reset(self) -> None:
        """重置状态"""
        pass

---

6. 最佳实践

始终设置最大消息数

避免无限循环：

# ✅ 好的实践
team = SelectorGroupChat(
    [agent1, agent2],
    termination_condition=OrTermination([
        MaxMessageTermination(20),  # 安全网
        TextMentionTermination("完成")  # 正常终止
    ])
)

# ❌ 避免：没有安全网
team = SelectorGroupChat(
    [agent1, agent2],
    termination_condition=TextMentionTermination("完成")  # 可能永远不会触发
)

使用超时

对于可能长时间运行的任务：

termination = OrTermination([
    MaxMessageTermination(50),
    TimeoutTermination(600),  # 10 分钟超时
])

---

📝 练习

1. 创建一个带 MaxMessageTermination 的简单对话
2. 使用 TextMentionTermination 检测"完成"
3. 创建组合终止条件

---

🔗 相关链接

• [00-AgentChat 总览](00-AgentChat 总览)
• 02-群聊模式
• [../04-实战示例/04-多 Agent 协作](../04-实战示例/04-多 Agent 协作)