上海做b2b国际网站公司seo数据优化
这篇文章瞄准的是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。
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 对终止条件进行组合。
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 以控制组内轮询次数。
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负责），就会触发终止；