- 👉👉👉本人承接各类AI相关应用开发项目(包括但不限于大模型微调、RAG、AI智能体、NLP、机器学习算法、运筹优化算法、数据分析EDA等) !!!
- 👉👉👉 有意愿请私信!!!
AutoGen 的 AgentChat 模块为我们提供了强大的多智能体协作能力，通过 Teams 功能，我们可以轻松构建一个由多个智能体组成的团队，共同完成复杂的任务。本文将带你一步步了解如何创建、运行、观察、控制和管理一个 AutoGen 团队，并结合示例代码，让你快速上手实践。
为什么要使用 Teams？
在处理需要协作和专业知识的任务时，多智能体团队是一个强大的工具。与单个智能体相比，团队可以并行处理任务，利用每个智能体的独特优势，从而更有效地解决复杂问题。然而，团队也需要更多的管理和协调，AutoGen 旨在简化这个过程。
注意： 对于简单的任务，首先考虑使用单个智能体。只有当单个智能体无法胜任时，再转向多智能体团队。在创建团队之前，确保已经优化了单个智能体，并赋予了它所需的工具和指令。
创建一个 Team
AutoGen 中的 RoundRobinGroupChat 是一种简单而有效的团队配置，它允许所有智能体共享相同的上下文，并以轮询方式依次响应。每个智能体在轮到自己时，将响应广播给所有其他智能体，确保团队保持一致的上下文。
我们将创建一个包含两个 AssistantAgent 的团队：一个主要智能体和一个评论员智能体。同时，我们使用 TextMentionTermination 条件，当评论员智能体的响应中出现特定词语时，停止团队的运行。
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)上述代码实现了“反思”模式，即一个评论员智能体评估主要智能体的响应。
运行 Team
使用 run() 方法启动团队，并为其分配一个任务：
# 在脚本中运行时，使用 `asyncio.run(...)`
result = await team.run(task="写一首关于秋天的短诗。")
print(result)
我们看到返回结果result是一个TextMessage消息体的集合，里面包含了 user,和我们定义的primary,critic 这两个智能体，下面我们过滤一下result的内容，以便提取我们想要的结果：
print(result.messages[1].content)
团队会运行智能体，直到满足终止条件。在本例中，团队以轮询顺序运行智能体，直到在智能体的响应中检测到“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)
如上例所示，你可以通过检查 stop_reason 属性来确定团队停止的原因。
Console()方法提供了一种方便的方式，可以将消息以适当的格式打印到控制台。
await team.reset() # 为新任务重置团队
await Console(team.run_stream(task="写一首关于秋天的短诗。")) # 将消息流式传输到控制台 
重置 Team
你可以通过调用 reset() 方法重置团队。此方法将清除团队的状态，包括所有智能体。它将调用每个智能体的 on_reset()方法来清除智能体的状态。
await team.reset() # 为下一次运行重置团队如果下一个任务与上一个任务无关，通常最好重置团队。但是，如果下一个任务与上一个任务相关，则无需重置，而是可以恢复团队。
停止 Team
除了自动终止条件（例如 TextMentionTermination）之外，你还可以通过使用 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
从上面的输出中，你可以看到团队因满足外部终止条件而停止，但发言智能体能够在团队停止之前完成其回合。
恢复 Team
除非你重置团队，否则团队是有状态的，并在每次运行后保持对话历史记录和上下文。
你可以通过再次调用 run() 或 run_stream() 方法来恢复团队，从而从上次停止的地方继续，而无需提供新任务。 RoundRobinGroupChat 将从轮询顺序中的下一个智能体继续。
await Console(team.run_stream()) # 恢复团队以继续上一个任务
你可以看到团队从上次停止的地方恢复，并且第一条消息来自团队停止之前发言的最后一个智能体之后的下一个智能体。
让我们在保留有关上一个任务的上下文的同时，使用新任务再次恢复团队。
# 新任务是将同一首诗翻译成中文唐诗风格。
await Console(team.run_stream(task="将这首诗用中文唐诗风格写一遍。")) 
中止 Team
你可以在执行期间通过设置传递给 cancellation_token 参数的 CancellationToken 来中止对 run() 或 run_stream() 的调用。
与停止团队不同，中止团队将立即停止团队并引发 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("任务已取消。")
总结
AutoGen 的 Teams 功能为我们提供了一种构建多智能体协作系统的强大方式。通过了解如何创建、运行、观察、控制和管理团队，你可以充分利用 AutoGen 的 AgentChat 模块，解决各种复杂的任务。希望本文能帮助你快速上手 AutoGen 的 Teams 功能，并在实践中发挥其强大的潜力！
