AutoGen学习笔记系列（六）Tutorial

这篇文章瞄准的是AutoGen框架官方教程中的 Tutorial 章节中的 Termination 小节，主要介绍了更细粒度上图如何终止Team组内轮询的过程。

官网链接：https://microsoft.github.io/autogen/stable/user-guide/agentchat-user-guide/tutorial/termination.html# ；

Termination

之前的数篇文章介绍了如何对一个正在运行的Team进行暂停、恢复、终止、条件判断等操作，这篇文章将对终止这个操作进行更详细的介绍。

你可能会觉得那些方案已经够用了，这里怎么单独整出来一章介绍如何终止？即便是暂停、恢复都没有这待遇（这两个功能被挤在一章 "Human-in-the-loop" 中）。

提出这问题至少说明你有认真思考过，但没有经历过钱包的毒打：我们要控制Token数！！！。

特别是对于 OpanAI、DeepSeek、Gemini 这种付费 API而言，你的Team与服务器之间每一句废话都是对你钱包余额的不尊重。

AutoGen提供了下面几个自带的终止控制器，其中加粗的是在之前我们已经用过的，一定要区分这里的终止控制器和之前提到的 max_turn 参数：

MaxMessageTermination: 在Team沟通数达到条件后终止；
TextMentionTermination: 检测到指定字符串信息后终止；
TokenUsageTermination: 消耗了指定Token数后终止；
TimeoutTermination: 超过指定时间后终止；
HandoffTermination: 当控制权发生转移后终止；
SourceMatchTermination: Team中指定Agent响应后终止；
ExternalTermination: 允许外部自定义终止条件，如GUI界面的按钮动作；
StopMessageTermination: Agent生成StopMessage对象后终止；
TextMessageTermination: Agent生成 TextMessage 对象后终止；

Basic Usage

官网在这里列举了一个写诗的示例，还是一个Agent写、另一个Agent评判，但是他让写的是一个 关于巴黎天气的独特俳句，里面涉及到音律，我对这块一窍不通就只能从技术上分析这个demo。

python">from autogen_agentchat.agents import AssistantAgent
from autogen_agentchat.conditions import MaxMessageTermination, TextMentionTermination
from autogen_agentchat.teams import RoundRobinGroupChat
from autogen_agentchat.ui import Console
from autogen_ext.models.openai import OpenAIChatCompletionClient
import os, asyncioos.environ["OPENAI_API_KEY"] = "你的OpenAI API Key"
model_client = OpenAIChatCompletionClient(model="gpt-4o",temperature=1,
)# 主Agent用来生成诗句
primary_agent = AssistantAgent("primary",model_client=model_client,system_message="You are a helpful AI assistant.",
)# 评判Agent为生成的诗句进行打分
critic_agent = AssistantAgent("critic",model_client=model_client,system_message="Provide constructive feedback for every message. Respond with 'APPROVE' to when your feedbacks are addressed.",
)# 设置Team内部最大沟通信息数的终止条件
max_msg_termination = MaxMessageTermination(max_messages=3)
round_robin_team = RoundRobinGroupChat([primary_agent, critic_agent], termination_condition=max_msg_termination)asyncio.run(Console(round_robin_team.run_stream(task="Write a unique, Haiku about the weather in Paris"))
)

运行结果如下：

$ python demo.py

在这里插入图片描述

Combining Termination Conditions

关于组合终止判定条件在先前的demo中已经用到过，AutoGen允许使用逻辑运算符 AND、OR 对终止条件进行组合。

python">from autogen_agentchat.agents import AssistantAgent
from autogen_agentchat.conditions import MaxMessageTermination, TextMentionTermination
from autogen_agentchat.teams import RoundRobinGroupChat
from autogen_agentchat.ui import Console
from autogen_ext.models.openai import OpenAIChatCompletionClient
import os, asyncioos.environ["OPENAI_API_KEY"] = "你的OpenAI API Key"
model_client = OpenAIChatCompletionClient(model="gpt-4o",temperature=1,
)
primary_agent = AssistantAgent("primary",model_client=model_client,system_message="You are a helpful AI assistant.",
)
critic_agent = AssistantAgent("critic",model_client=model_client,system_message="Provide constructive feedback for every message. Respond with 'APPROVE' to when your feedbacks are addressed.",
)#-------------------------------------------------------------#
# 创建一个最大Team内轮询终止条件
max_msg_termination = MaxMessageTermination(max_messages=10)
# 创建一个文本关键字终止条件，检测到 APPROVE 后自动停止
text_termination = TextMentionTermination("APPROVE")
# 将两个终止条件以或操作符进行合并
combined_termination = max_msg_termination | text_terminationround_robin_team = RoundRobinGroupChat([primary_agent, critic_agent], termination_condition=combined_termination)asyncio.run(Console(round_robin_team.run_stream(task="Write a unique, Haiku about the weather in Paris"))
)

运行结果如下：

$ python demo.py

在这里插入图片描述

Custom Termination Condition

通常情况下内置的终止判断功能已经能够满足要求，但AutoGen也提供了自定义终止判断函数的接口 TerminationCondition；

【注】：这里官方给的demo运行后会抛出异常，并且会持续循环下去，需要对其进行修正再添加一个 MaxMessageTermination 以控制组内轮询次数。

python">from typing import Sequencefrom autogen_agentchat.base import TerminatedException, TerminationCondition
from autogen_agentchat.conditions import MaxMessageTermination
from autogen_agentchat.messages import AgentEvent, ChatMessage, StopMessage, ToolCallExecutionEvent
from autogen_core import Component
from pydantic import BaseModel
from typing_extensions import Self
import os, asynciofrom autogen_agentchat.agents import AssistantAgent
from autogen_agentchat.teams import RoundRobinGroupChat
from autogen_agentchat.ui import Console
from autogen_ext.models.openai import OpenAIChatCompletionClientos.environ["OPENAI_API_KEY"] = "你的OpenAI API Key"def approve() -> None:"""Approve the message when all feedbacks have been addressed."""pass
model_client = OpenAIChatCompletionClient(model="gpt-4o",temperature=1,
)
primary_agent = AssistantAgent("primary",model_client=model_client,system_message="You are a helpful AI assistant.",
)
critic_agent = AssistantAgent("critic",model_client=model_client,tools=[approve],  # Register the approve function as a tool.system_message="Provide constructive feedback. Use the approve tool to approve when all feedbacks are addressed.",
)#-------------------------------------------------------------------------#class FunctionCallTerminationConfig(BaseModel):"""Configuration for the termination condition to allow for serializationand deserialization of the component."""function_name: strclass FunctionCallTermination(TerminationCondition, Component[FunctionCallTerminationConfig]):"""Terminate the conversation if a FunctionExecutionResult with a specific name is received."""component_config_schema = FunctionCallTerminationConfig"""The schema for the component configuration."""def __init__(self, function_name: str) -> None:self._terminated = Falseself._function_name = function_name@propertydef terminated(self) -> bool:return self._terminatedasync def __call__(self, messages: Sequence[AgentEvent | ChatMessage]) -> StopMessage | None:if self._terminated:raise TerminatedException("Termination condition has already been reached")for message in messages:if isinstance(message, ToolCallExecutionEvent):for execution in message.content:if getattr(execution, "tool_name", None) == self._function_name:  # 改为 tool_nameself._terminated = Truereturn StopMessage(content=f"Function '{self._function_name}' was executed.",source="FunctionCallTermination",)return Noneasync def reset(self) -> None:self._terminated = Falsedef _to_config(self) -> FunctionCallTerminationConfig:return FunctionCallTerminationConfig(function_name=self._function_name,)@classmethoddef _from_config(cls, config: FunctionCallTerminationConfig) -> Self:return cls(function_name=config.function_name,)#-------------------------------------------------------------------------#
max_messages_termination = MaxMessageTermination(max_messages=10)
function_call_termination = FunctionCallTermination(function_name="approve") | max_messages_termination
round_robin_team = RoundRobinGroupChat([primary_agent, critic_agent], termination_condition=function_call_termination)
asyncio.run(Console(round_robin_team.run_stream(task="Write a unique, Haiku about the weather in Paris"))
)

运行结果如下：

$ python demo.py

在这里插入图片描述

因为代码比较长，这里对其进行拆解分析：

`approve()` 函数：

这个函数没有实际逻辑，但它是 critic_agent 可以调用的一个工具；
一旦 critic_agent 调用了 approve()，对话就会终止（由 FunctionCallTermination 监测）；

`FunctionCallTermination` 对象：

这个类定义了 FunctionCallTermination 配置，用于存储function_name 变量来表示哪一个函数的执行会触发终止；

`FunctionCallTermination` 对象：

这个类监听 ToolCallExecutionEvent；
如果某个Agent调用了 approve() 方法（由 critic_agent 负责），就会触发终止；