最近，OpenAI 推出的 Swarm 框架引发了业界对 MultiAgent框架的热议。特别是，Swarm 中的核心原语抽象——handoff（交接）， 允许代理之间高效转移任务，以其简单直观的设计，受到了好评。
什么是 Handoff 模式？
如果想象一下四人接力赛，运动员们一棒接一棒，顺畅无比。换到 AI 的世界里，“交接”就像是代理之间互相传递工作任务。OpenAI 推出的 Swarm 框架十分灵活，这种交接模式允许一个 AI 代理把任务无缝转交给其他代理，如同在聊天中把话筒递给下一个人。
AutoGen 的回应
作为 MultiAgent 框架领域的代表 AutoGen 坐不住了，发文[1]表示，利用 AutoGen 的灵活和可扩展的系统设计，同样可以实现 Handoffs，并且还具备以下优势：
-
可扩展性：AutoGen 可以轻松扩展到分布式环境，支持更大规模的代理协作。
-
灵活性：开发者可以自定义代理的实现，根据需求打造专属的 AI 团队。
-
集成便利性：原生异步 API 可以轻松与 UI 和其他系统集成，适应各种应用场景。
-
开放性：作为开源项目，AutoGen 一样允许开发者自由扩展和优化系统。
一个客户服务场景中，我们可能会有这样的阵容：
-
客户 Agent（UserAgent）处理用户登录和 Agent 响应，并启动与聊天机器人的新会话。 -
分流Agent（Triage Agent）：负责初步评估客户问题 -
维修Agent（Issue and Repair Agent）：处理维修相关事务 -
销售Agent（Sales Agent）：负责产品推荐和销售 -
人类Agent（Human Agent）：处理复杂情况或需要人工干预的问题，确保系统的鲁棒性。
每个 AI Agent都由 AIAgent 类来处理，它们可以：
-
使用 ChatCompletionClient 生成回复
-
通过 delegate_tools 将任务委派给其他代理
-
订阅特定的主题类型，并向用户发送消息，每个 Agent 都有自己的主题类型，如 sales_agent_topic_type、issues_and_repairs_agent_topic_type 等，便于消息的精准投递。
当你发问时，初步分流 Agent 会首先接手。如果问题超出了它的处理能力，它会把任务“交接”给维修 Agent 或者销售 Agent，或者在必要时直接转给人类客服，确保你能得到及时的帮助。这种“交接”就像是办公室里的团队协作，确保没有人被落下。
下面是一些具体实现：
-
定义不同角色的 Agent
runtime = SingleThreadedAgentRuntime()
model_client = OpenAIChatCompletionClient(
model="gpt-4o-mini",
# api_key="YOUR_API_KEY",
)
# Register the triage agent.
triage_agent_type = await AIAgent.register(
runtime,
type=triage_agent_topic_type, # Using the topic type as the agent type.
factory=lambda: AIAgent(
description="A triage agent.",
system_message=SystemMessage(
content="You are a customer service bot for ACME Inc. "
"Introduce yourself. Always be very brief. "
"Gather information to direct the customer to the right department. "
"But make your questions subtle and natural."
),
model_client=model_client,
tools=[],
delegate_tools=[
transfer_to_issues_and_repairs_tool,
transfer_to_sales_agent_tool,
escalate_to_human_tool,
],
agent_topic_type=triage_agent_topic_type,
user_topic_type=user_topic_type,
),
)
# Add subscriptions for the triage agent: it will receive messages published to its own topic only.
await runtime.add_subscription(TypeSubscription(topic_type=triage_agent_topic_type, agent_type=triage_agent_type.type))
# Register the sales agent.
sales_agent_type = await AIAgent.register(
runtime,
type=sales_agent_topic_type, # Using the topic type as the agent type.
factory=lambda: AIAgent(
description="A sales agent.",
system_message=SystemMessage(
content="You are a sales agent for ACME Inc."
"Always answer in a sentence or less."
"Follow the following routine with the user:"
"1. Ask them about any problems in their life related to catching roadrunners.n"
"2. Casually mention one of ACME's crazy made-up products can help.n"
" - Don't mention price.n"
"3. Once the user is bought in, drop a ridiculous price.n"
"4. Only after everything, and if the user says yes, "
"tell them a crazy caveat and execute their order.n"
""
),
model_client=model_client,
tools=[execute_order_tool],
delegate_tools=[transfer_back_to_triage_tool],
agent_topic_type=sales_agent_topic_type,
user_topic_type=user_topic_type,
),
)
# Add subscriptions for the sales agent: it will receive messages published to its own topic only.
await runtime.add_subscription(TypeSubscription(topic_type=sales_agent_topic_type, agent_type=sales_agent_type.type))
# Register the issues and repairs agent.
issues_and_repairs_agent_type = await AIAgent.register(
runtime,
type=issues_and_repairs_agent_topic_type, # Using the topic type as the agent type.
factory=lambda: AIAgent(
description="An issues and repairs agent.",
system_message=SystemMessage(
content="You are a customer support agent for ACME Inc."
"Always answer in a sentence or less."
"Follow the following routine with the user:"
"1. First, ask probing questions and understand the user's problem deeper.n"
" - unless the user has already provided a reason.n"
"2. Propose a fix (make one up).n"
"3. ONLY if not satesfied, offer a refund.n"
"4. If accepted, search for the ID and then execute refund."
),
model_client=model_client,
tools=[
execute_refund_tool,
look_up_item_tool,
],
delegate_tools=[transfer_back_to_triage_tool],
agent_topic_type=issues_and_repairs_agent_topic_type,
user_topic_type=user_topic_type,
),
)
# Add subscriptions for the issues and repairs agent: it will receive messages published to its own topic only.
await runtime.add_subscription(
TypeSubscription(topic_type=issues_and_repairs_agent_topic_type, agent_type=issues_and_repairs_agent_type.type)
)
# Register the human agent.
human_agent_type = await HumanAgent.register(
runtime,
type=human_agent_topic_type, # Using the topic type as the agent type.
factory=lambda: HumanAgent(
description="A human agent.",
agent_topic_type=human_agent_topic_type,
user_topic_type=user_topic_type,
),
)
# Add subscriptions for the human agent: it will receive messages published to its own topic only.
await runtime.add_subscription(TypeSubscription(topic_type=human_agent_topic_type, agent_type=human_agent_type.type))
# Register the user agent.
user_agent_type = await UserAgent.register(
runtime,
type=user_topic_type,
factory=lambda: UserAgent(
description="A user agent.",
user_topic_type=user_topic_type,
agent_topic_type=triage_agent_topic_type, # Start with the triage agent.
),
)
# Add subscriptions for the user agent: it will receive messages published to its own topic only.
await runtime.add_subscription(TypeSubscription(topic_type=user_topic_type, agent_type=user_agent_type.type))
-
执行
# Start the runtime.
runtime.start()
# Create a new session for the user.
session_id = str(uuid.uuid4())
await runtime.publish_message(UserLogin(), topic_id=TopicId(user_topic_type, source=session_id))
# Run until completion.
await runtime.stop_when_idle()
# 运行日志：
--------------------------------------------------------------------------------
User login, session ID: 7a568cf5-13e7-4e81-8616-8265a01b3f2b.
--------------------------------------------------------------------------------
User:
I want a refund
--------------------------------------------------------------------------------
TriageAgent:
I can help with that! Could I ask what item you're seeking a refund for?
--------------------------------------------------------------------------------
User:
A pair of shoes I bought
--------------------------------------------------------------------------------
TriageAgent:
[FunctionCall(id='call_qPx1DXDL2NLcHs8QNo47egsJ', arguments='{}', name='transfer_to_issues_and_repairs')]
--------------------------------------------------------------------------------
TriageAgent:
Delegating to IssuesAndRepairsAgent
--------------------------------------------------------------------------------
IssuesAndRepairsAgent:
I see you're looking for a refund on a pair of shoes. Can you tell me what the issue is with the shoes?
--------------------------------------------------------------------------------
User:
The shoes are too small
--------------------------------------------------------------------------------
IssuesAndRepairsAgent:
I recommend trying a size up as a fix; would that work for you?
--------------------------------------------------------------------------------
User:
no I want a refund
--------------------------------------------------------------------------------
IssuesAndRepairsAgent:
[FunctionCall(id='call_Ytp8VUQRyKFNEU36mLE6Dkrp', arguments='{"search_query":"shoes"}', name='look_up_item')]
--------------------------------------------------------------------------------
IssuesAndRepairsAgent:
[FunctionExecutionResult(content='item_132612938', call_id='call_Ytp8VUQRyKFNEU36mLE6Dkrp')]
--------------------------------------------------------------------------------
IssuesAndRepairsAgent:
[FunctionCall(id='call_bPm6EKKBy5GJ65s9OKt9b1uE', arguments='{"item_id":"item_132612938","reason":"not provided"}', name='execute_refund')]
--------------------------------------------------------------------------------
IssuesAndRepairsAgent:
[FunctionExecutionResult(content='success', call_id='call_bPm6EKKBy5GJ65s9OKt9b1uE')]
--------------------------------------------------------------------------------
IssuesAndRepairsAgent:
Your refund has been successfully processed! If you have any other questions, feel free to ask.
--------------------------------------------------------------------------------
User:
I want to talk to your manager
--------------------------------------------------------------------------------
IssuesAndRepairsAgent:
I can help with that, let me transfer you to a supervisor.
--------------------------------------------------------------------------------
User:
Okay
--------------------------------------------------------------------------------
IssuesAndRepairsAgent:
[FunctionCall(id='call_PpmLZvwNoiDPUH8Tva3eAwHX', arguments='{}', name='transfer_back_to_triage')]
--------------------------------------------------------------------------------
IssuesAndRepairsAgent:
Delegating to TriageAgent
--------------------------------------------------------------------------------
TriageAgent:
[FunctionCall(id='call_jSL6IBm5537Dr74UbJSxaj6I', arguments='{}', name='escalate_to_human')]
--------------------------------------------------------------------------------
TriageAgent:
Delegating to HumanAgent
--------------------------------------------------------------------------------
HumanAgent:
Hello this is manager
--------------------------------------------------------------------------------
User:
Hi! Thanks for your service. I give you a 5 start!
--------------------------------------------------------------------------------
HumanAgent:
Thanks.
--------------------------------------------------------------------------------
User:
exit
--------------------------------------------------------------------------------
User session ended, session ID: 7a568cf5-13e7-4e81-8616-8265a01b3f2b.
总结
AutoGen 的 Handoff 实现不仅匹敌 OpenAI 的 Swarm，还在多个方面有所超越。它为开发者提供了一个强大、灵活且可扩展的框架，以构建复杂的多代理 AI 系统。不过，相较于 OpenAI 的 Swarm 实现，AutoGen 明显感觉概念更多，依赖更重，网络上就有人这么评价，“OpenAI 发布了一个示例，大约有 12 行代码，易于阅读，除了 swarm 模块外，没有其他依赖。对于 Autogen，我找不到类似的东西；似乎所有的示例都更复杂。”
这或许也是 AutoGen 之类成熟产品最大软肋吧！
