AutoGen：玩转多智能体团队协作 (Teams)
- 👉👉👉本人接办多种与AI相关的应用开发项目（涵盖大模型微调、RAG等） !!!
- 👉👉👉 如需合作，请随时私信 !!!
AutoGen 的 AgentChat 模块为我们提供了强大的复杂协作能力。通过团队功能模块, 我们可以方便地建立由多个智能体构成的合作团队。本文旨在引导你逐步掌握创建、观察、控制和管理一个 AutoGen 团队的方法, 并通过示例代码演示帮助你快速实现。
为什么要使用 Teams？
在应对需要协作与专业知识的任务时，多智能体团队是一个强有力的手段。相较于单一智能体而言，在多智能体系统中可以实现任务的并行处理，并充分挖掘各成员的专业技能从而更有效地解决复杂问题。然而，在实际应用中，团队管理确实存在一定的挑战。
注意： 对于较为简单的任务，在构建团队之前，请优先考虑配置单一智能体方案；如果该方案无法满足需求，则升级为多智能体团队方案。
创建一个 Team
在 AutoGen 系统中，默认配置下的 GroupChat 模块采用了轮询式的交互机制。其核心机制在于通过共享统一的上下文信息实现高效协作，在每一轮次中依次由各参与方完成特定任务。当某个参与者轮到其行动时，则会将发出的信息同步至其余所有参与者。这种设计不仅简化了团队管理流程，在实际运行中还能够有效提升整体系统的响应效率与一致性保障能力。
我们将由两个AssistantAgent组成的团队进行构建：一个为核心智能体、另一个为监督者智能体。此外，在监督者智能体的响应中检测到特定关键词时，则会终止该团队的任务运行。
import asyncio
from autogen_agentchat.agents import AssistantAgent
from autogen_agentchat.base import TaskResult
from autogen_agentchat.conditions import ExternalTermination, TextMentionTermination
from autogen_agentchat.teams import RoundRobinGroupChat
from autogen_agentchat.ui import Console
from autogen_core import CancellationToken
from autogen_ext.models.openai import OpenAIChatCompletionClient
# 创建一个使用 deepseek 模型的 Agent
model_client = OpenAIChatCompletionClient(
model="deepseek-chat",
base_url="https://api.deepseek.com",
api_key="sk-XXXXXXX",
model_info={
"vision": True,
"function_calling": True,
"json_output": True,
"family": "unknown",
},
)
# 创建主要智能体
primary_agent = AssistantAgent(
"primary",
model_client=model_client,
system_message="你是一个乐于助人的 AI 助手。",
)
# 创建评论员智能体
critic_agent = AssistantAgent(
"critic",
model_client=model_client,
system_message="提供建设性的反馈。当你的反馈得到解决时，回复 'APPROVE'。",
)
# 定义一个终止条件，当评论员批准时，停止任务
text_termination = TextMentionTermination("APPROVE")
# 创建一个包含主要智能体和评论员智能体的团队
team = RoundRobinGroupChat([primary_agent, critic_agent],
termination_condition=text_termination)
python
上述代码实现了“反思”模式，即一个评论员智能体评估主要智能体的响应。
Team
使用 run() 方法启动团队，并为其分配一个任务：
# 在脚本中时，使用 `asyncio.run(...)`
result = await team.run(task="写一首关于秋天的短诗。")
print(result)
python
我们观察到返回结果result是一个TextMessage消息体的集合，在其中包含了user以及我们定义的primary和critic这两个智能体。为了方便后续操作，请对result进行筛选处理以提取所需信息。
print(result.messages[1].content)
python
该团队采用智能体方案直至满足终止条件，在此案例中采用轮询顺序的智能体并识别到'APPROVE'关键词后完成操作并返回一个TaskResult对象
观察 Team
该方法允许您在团队环境中实现实时消息传输功能。通过调用 run_stream() 函数即可实现这一目标。该生成器将依次输出各智能体发送的消息，并最终提供一个 TaskResult 对象作为结果总结。
# 在脚本内部时，使用异步 main 函数并通过 `asyncio.run(...)` 调用它
await team.reset() # 为新任务重置团队
async for message in team.run_stream(task="写一首关于秋天的短诗。"): # type: ignore
if isinstance(message, TaskResult):
print("停止原因：", message.stop_reason)
else:
print(message)
python
如前所述，在此之前展示的例子中，请注意你可以通过检查该链接中的 [stop_reason] 属性来确定团队停止的原因。
该Console()方法支持一种便捷的方式以标准格式打印消息到控制台。
await team.reset() # 为新任务重置团队
await Console(team.run_stream(task="写一首关于秋天的短诗。")) # 将消息流式传输到控制台
python
重置 Team
你可以通过执行[reset()]函数来重启团队。此操作会取消所有参与者的当前状态设置并恢复到初始值。每条参与者线程都会被重置到指定初始条件并重新启动相关的逻辑流程。
await team.reset() # 为下一次重置团队
python如果下一个任务与上一个任务无关的话，在大多数情况下应重新配置团队；但如果下一个任务与上一个任务相关，则无需重新配置而可以直接恢复。
停止 Team
除了自动终止条件（例如TextMentionTermination）之外，在Microsoft Autogen中还提供了一个外部终止条件ExternalTermination。你可以通过调用该终止条件将使团队在指定的外部条件下执行停止操作。
当在[ExternalTermination]节点调用[set()]方法时，在当前智能体的回合结束后会暂停整个团队。这样，在当前智能体完成回合可能会不会立即停下。这使得当前智能体可以在完成回合击合后发送最终消息到所有成员后再执行关闭操作。这样，在完成一个完整的轮回之后才有可能发送最终消息到所有成员后再关闭系统以保证整个系统的稳定性
# 创建一个带有外部终止条件的新团队
external_termination = ExternalTermination()
team = RoundRobinGroupChat(
[primary_agent, critic_agent],
termination_condition=external_termination | text_termination, # 使用按位 OR 运算符来组合条件
)
# 在后台任务中团队
run = asyncio.create_task(Console(team.run_stream(task="写一首关于秋天的短诗。")))
# 等待一段时间
await asyncio.sleep(0.1)
# 停止团队
external_termination.set()
# 等待团队完成
await run
python
从上面的输出中可以看到, 团队因为满足外部终止条件而停滞, 但发言智能体能够在团队停滞前就位完成其回合
恢复 Team
除非你选择重置团队配置并完成初始化设置, 否则该团队将保持其初始状态, 并在每次操作后持续记录对话历史以及相关的上下文信息
你可以通过再次执行 run() 或 run_stream() 方法来重新启动团队，并无需重新开始任务。 RoundRobinGroupChat 将从轮询顺序中的下一个智能体继续运行。
await Console(team.run_stream()) # 恢复团队以继续上一个任务
python
你可以看到团队恢复到了上一次停顿的位置，并且第一个消息是由团队停止前最后一个智能体发表后发出的下一个智能体。
为了保持关于上一任务的背景信息，在同一时间段内引入新的任务以恢复团队的力量。
# 新任务是将同一首诗翻译成中文唐诗风格。
await Console(team.run_stream(task="将这首诗用中文唐诗风格写一遍。"))
python
中止 Team
你可以通过传递 cancellation_token 参数来停止对 run() 或 run_stream() 方法的调用，在执行期间实现这一功能。
这一操作将通过参数设置实现，并可参考 Autogen Agent Chat Teams 方法的相关文档以获取详细信息。
该功能允许在程序运行时主动终止特定操作的执行。
不同于停止团队，则会立即终止该操作，并在执行过程中触发带有CancelledError类型的异常
不同于停止团队，则会立即终止该操作，并在执行过程中触发带有CancelledError类型的异常
注意： 当团队中止时，调用方将收到 CancelledError 异常。
# 创建一个取消令牌
cancellation_token = CancellationToken()
# 使用另一个协程团队
run = asyncio.create_task(
team.run(
task="将这首诗翻译成西班牙语。",
cancellation_token=cancellation_token,
)
)
# 取消
cancellation_token.cancel()
try:
result = await run # 这将引发 CancelledError。
except asyncio.CancelledError:
print("任务已取消。")
python
总结
AutoGen 的 Teams 功能为我们支持一种构建多智能体协作系统的能力。学习如何搭建, 观察, 管理以及调控团队结构, 你可以充分调用 AutoGen 的 AgentChat 模块来应对多种复杂任务。希望本文能帮助您掌握 AutoGen 的 Teams 功能, 并在实际应用中充分发挥其潜力！