盘锦做网站谁家好,王烨辉简历,pc门户网站是什么意思,ppt公司简介页面设计这篇文章瞄准的是AutoGen框架官方教程中的 Tutorial 章节中的 Termination 小节#xff0c;主要介绍了更细粒度上图如何终止Team组内轮询的过程。
官网链接#xff1a;https://microsoft.github.io/autogen/stable/user-guide/agentchat-user-guide/tutorial/termination.ht…这篇文章瞄准的是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(modelgpt-4o,temperature1,
)# 主Agent用来生成诗句
primary_agent AssistantAgent(primary,model_clientmodel_client,system_messageYou are a helpful AI assistant.,
)# 评判Agent为生成的诗句进行打分
critic_agent AssistantAgent(critic,model_clientmodel_client,system_messageProvide constructive feedback for every message. Respond with APPROVE to when your feedbacks are addressed.,
)# 设置Team内部最大沟通信息数的终止条件
max_msg_termination MaxMessageTermination(max_messages3)
round_robin_team RoundRobinGroupChat([primary_agent, critic_agent], termination_conditionmax_msg_termination)asyncio.run(Console(round_robin_team.run_stream(taskWrite a unique, Haiku about the weather in Paris))
)运行结果如下
$ python demo.pyCombining 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(modelgpt-4o,temperature1,
)
primary_agent AssistantAgent(primary,model_clientmodel_client,system_messageYou are a helpful AI assistant.,
)
critic_agent AssistantAgent(critic,model_clientmodel_client,system_messageProvide constructive feedback for every message. Respond with APPROVE to when your feedbacks are addressed.,
)#-------------------------------------------------------------#
# 创建一个最大Team内轮询终止条件
max_msg_termination MaxMessageTermination(max_messages10)
# 创建一个文本关键字终止条件检测到 APPROVE 后自动停止
text_termination TextMentionTermination(APPROVE)
# 将两个终止条件以或操作符进行合并
combined_termination max_msg_termination | text_terminationround_robin_team RoundRobinGroupChat([primary_agent, critic_agent], termination_conditioncombined_termination)asyncio.run(Console(round_robin_team.run_stream(taskWrite a unique, Haiku about the weather in Paris))
)运行结果如下
$ python demo.pyCustom 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 Keydef approve() - None:Approve the message when all feedbacks have been addressed.pass
model_client OpenAIChatCompletionClient(modelgpt-4o,temperature1,
)
primary_agent AssistantAgent(primary,model_clientmodel_client,system_messageYou are a helpful AI assistant.,
)
critic_agent AssistantAgent(critic,model_clientmodel_client,tools[approve], # Register the approve function as a tool.system_messageProvide 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 FunctionCallTerminationConfigThe schema for the component configuration.def __init__(self, function_name: str) - None:self._terminated Falseself._function_name function_namepropertydef 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(contentfFunction {self._function_name} was executed.,sourceFunctionCallTermination,)return Noneasync def reset(self) - None:self._terminated Falsedef _to_config(self) - FunctionCallTerminationConfig:return FunctionCallTerminationConfig(function_nameself._function_name,)classmethoddef _from_config(cls, config: FunctionCallTerminationConfig) - Self:return cls(function_nameconfig.function_name,)#-------------------------------------------------------------------------#
max_messages_termination MaxMessageTermination(max_messages10)
function_call_termination FunctionCallTermination(function_nameapprove) | max_messages_termination
round_robin_team RoundRobinGroupChat([primary_agent, critic_agent], termination_conditionfunction_call_termination)
asyncio.run(Console(round_robin_team.run_stream(taskWrite 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 负责就会触发终止
