AutoGen 技術博客系列八：深入剖析 Swarm—

本系列博文在掘金同步發布, 更多優質文章，請關注本人掘金賬號：
人肉推土機的掘金賬號

AutoGen系列一：基礎介紹與入門教程

AutoGen系列二：深入自定義智能體

AutoGen系列三：內置智能體的應用與實戰

AutoGen系列四：自定義智能體的高級技巧

AutoGen系列五: 智能體團隊協作的深度剖析與實踐

AutoGen 技術博客系列 (六)：SelectorGroupChat 的原理與實踐

AutoGen 技術博客系列 (七)：狀態管理與組件序列化解析

AutoGen 技術博客系列八：深入剖析 Swarm—— 智能體協作的新范式

AutoGen 技術博客系列九：從 v0.2 到 v0.4 的遷移指南

在人工智能的浩瀚星空中，AutoGen 的 Swarm 模式猶如一顆耀眼的新星，正引領著智能體協作的新潮流。今天，就讓我們一同深入探究這一強大模式的奧秘。

一、Swarm 模式：智能協作的核心引擎

Swarm 模式的設計理念獨具匠心，它打破了傳統智能體協作的桎梏，允許智能體基于自身能力靈活地將任務交接給其他合適的智能體，并且所有智能體在同一消息上下文環境下協同工作。這一特性使得智能體能夠自主地進行任務規劃和分配，極大地提升了系統的靈活性與適應性。

在其運行機制中，智能體輪流生成響應并廣播，而每次發言智能體的選定取決于上下文中最新的 HandoffMessage。這就要求每個智能體都具備生成 HandoffMessage 的能力，從而明確指示任務的交接方向。例如，在 AssistantAgent 中，我們可以通過設置 handoffs 參數來精準指定可交接的目標智能體，并利用 Handoff 進一步定制消息內容和交接行為，為智能體之間的協作提供了豐富的定制化空間。

從實現原理的深度層面來看，當團隊接收到任務時，首個發言智能體迅速展開任務處理，并根據任務的具體情況和自身的能力判斷是否需要交接任務以及交接的對象。一旦某個智能體生成 HandoffMessage，接收智能體便會無縫接管任務，繼續在相同的消息上下文環境下推進工作。這種基于消息驅動的任務交接機制，確保了任務的連貫性和高效性。

值得注意的是，AssistantAgent 依賴模型的工具調用功能來實現任務交接，這就對模型提出了較高的要求，即必須支持工具調用。若模型執行并行工具調用，可能會引發意想不到的行為。為避免此類問題，在使用 OpenAIChatCompletionClient 或 AzureOpenAIChatCompletionClient 時，我們可以通過簡單地設置 parallel_tool_calls = False 來禁用并行工具調用，確保系統的穩定運行。

二、實戰案例：從航班退款到股票研究

（一）客戶支持案例：航班退款的智能流程

在這里插入圖片描述

【圖片來源 AutoGen 官方說明文檔】

在航班退款場景中，我們構建了一個包含旅行代理（Travel Agent）和航班退款專員（Flights Refunder）的智能系統，并允許用戶在必要時參與交互。

旅行代理作為系統的入口，負責啟動對話并全面評估用戶的退款請求。當遇到退款相關任務時，它會迅速將任務交接給航班退款專員；若需要用戶提供進一步的信息，如航班號等，它會將任務交接給用戶。航班退款專員則專注于使用 refund_flight 工具處理退款事宜，在需要用戶輸入時，會暫停團隊執行，等待用戶提供關鍵信息。

以下是具體的代碼實現：

    from typing import Any, Dict, Listfrom autogen_agentchat.agents import AssistantAgentfrom autogen_agentchat.conditions import HandoffTermination, TextMentionTerminationfrom autogen_agentchat.messages import HandoffMessagefrom autogen_agentchat.teams import Swarmfrom autogen_agentchat.ui import Consolefrom autogen_ext.models.openai import OpenAIChatCompletionClient# 定義退款工具函數def refund_flight(flight_id: str) -> str:return f"Flight {flight_id} refunded"# 創建模型客戶端model_client = OpenAIChatCompletionClient(model="gpt-4o",# api_key="YOUR_API_KEY",)# 創建旅行代理智能體travel_agent = AssistantAgent("travel_agent",model_client=model_client,handoffs=["flights_refunder", "user"],system_message="""You are a travel agent. The flights_refunder is in charge of refunding flights. If you need information from the user, you must first send your message, then you can handoff to the user. Use TERMINATE when the travel planning is complete.""",)# 創建航班退款專員智能體flights_refunder = AssistantAgent("flights_refunder",model_client=model_client,handoffs=["travel_agent", "user"],tools=[refund_flight],system_message="""You are an agent specialized in refunding flights. You only need flight reference numbers to refund a flight. You have the ability to refund a flight using the refund_flight tool. If you need information from the user, you must first send your message, then you can handoff to the user. When the transaction is complete, handoff to the travel agent to finalize.""",)# 設置終止條件termination = HandoffTermination(target="user") | TextMentionTermination("TERMINATE")team = Swarm([travel_agent, flights_refunder], termination_condition=termination)# 定義任務task = "I need to refund my flight."async def run_team_stream():task_result = await Console(team.run_stream(task=task))last_message = task_result.messages[-1]while isinstance(last_message, HandoffMessage) and last_message.target == "user":user_message = input("User: ")task_result = await Console(team.run_stream(task=HandoffMessage(source="user", target=last_message.source, content=user_message)))last_message = task_result.messages[-1]# 運行任務# Use asyncio.run(...) if you are running this in a script.await run_team_stream()

代碼解讀：

首先，我們導入了必要的模塊和類，包括 AssistantAgent、HandoffTermination、TextMentionTermination 等，這些是構建 Swarm 團隊和實現任務交接、終止條件判斷的關鍵組件。
接著定義了 refund_flight 函數，它模擬了航班退款的實際操作，接受航班號作為參數并返回退款成功的消息。
創建 OpenAIChatCompletionClient 作為模型客戶端，用于智能體與語言模型的交互，這里指定了使用的模型為 gpt-4o。
然后分別創建了 travel_agent 和 flights_refunder 兩個智能體。travel_agent 的系統消息表明它作為旅行代理的角色和任務交接規則，flights_refunder 則定義了其作為航班退款專員的職責和工具使用方法。
設置了終止條件 termination，它由 HandoffTermination（當交接目標為用戶時觸發）和 TextMentionTermination（當消息中提及 “TERMINATE” 時觸發）組合而成，確保任務在合適的時機結束。
最后定義了任務 “I need to refund my flight.” 并通過 run_team_stream 函數運行團隊任務，在任務執行過程中，如果遇到交接給用戶的情況，會暫停等待用戶輸入，然后繼續執行任務。

運行結果如下：

當用戶輸入 “I need to refund my flight.” 時，旅行代理首先做出響應：

---------- user ----------
I need to refund my flight.
---------- travel_agent ----------
[FunctionCall(id='call_ZQ2rGjq4Z29pd0yP2sNcuyd2', arguments='{}', name='transfer_to_flights_refunder')]
[Prompt tokens: 119, Completion tokens: 14]
---------- travel_agent ----------
[FunctionExecutionResult(content='Transferred to flights_refunder, adopting the role of flights_refunder immediately.', call_id='call_ZQ2rGjq4Z29pd0yP2sNcuyd2')]
---------- travel_agent ----------
Transferred to flights_refunder, adopting the role of flights_refunder immediately.

隨后航班退款專員接手任務，并詢問用戶航班號：

---------- flights_refunder ----------
Could you please provide me with the flight reference number so I can process the refund for you?
[Prompt tokens: 191, Completion tokens: 20]
---------- flights_refunder ----------
[FunctionCall(id='call_1iRfzNpxTJhRTW2ww9aQJ8sK', arguments='{}', name='transfer_to_user')]
[Prompt tokens: 219, Completion tokens: 11]
---------- flights_refunder ----------
[FunctionExecutionResult(content='Transferred to user, adopting the role of user immediately.', call_id='call_1iRfzNpxTJhRTW2ww9aQJ8sK')]
---------- flights_refunder ----------
Transferred to user, adopting the role of user immediately.

此時系統暫停，等待用戶輸入航班號。假設用戶輸入 “507811”，則航班退款專員繼續執行任務：

---------- user ----------
Sure, it's 507811
---------- flights_refunder ----------
[FunctionCall(id='call_UKCsoEBdflkvpuT9Bi2xlvTd', arguments='{"flight_id":"507811"}', name='refund_flight')]
[Prompt tokens: 266, Completion tokens: 18]
---------- flights_refunder ----------
[FunctionExecutionResult(content='Flight 507811 refunded', call_id='call_UKCsoEBdflkvpuT9Bi2xlvTd')]
---------- flights_refunder ----------
Tool calls:
refund_flight({"flight_id":"507811"}) = Flight 507811 refunded

完成退款后，航班退款專員將任務交接回旅行代理：

---------- flights_refunder ----------
[FunctionCall(id='call_MQ2CXR8UhVtjNc6jG3wSQp2W', arguments='{}', name='transfer_to_travel_agent')]
[Prompt tokens: 303, Completion tokens: 13]
---------- flights_refunder ----------
[FunctionExecutionResult(content='Transferred to travel_agent, adopting the role of travel_agent immediately.', call_id='call_MQ2CXR8UhVtjNc6jG3wSQp2W')]
---------- flights_refunder ----------
Transferred to travel_agent, adopting the role of travel_agent immediately.

最后旅行代理確認退款成功并終止流程：

---------- travel_agent ----------
Your flight with reference number 507811 has been successfully refunded. If you need anything else, feel free to let me know. Safe travels! TERMINATE
[Prompt tokens: 272, Completion tokens: 32]

從這個案例中，我們可以清晰地看到 Swarm 模式在客戶支持場景中的高效應用，智能體之間的任務交接順暢自然，有效地解決了用戶的問題。

（二）股票研究案例：多智能體協同的智慧結晶

在股票研究領域，我們設計了一個由四個智能體組成的強大系統，分別是規劃者（Planner）、金融分析師（Financial Analyst）、新聞分析師（News Analyst）和撰寫者（Writer）。
在這里插入圖片描述

【圖片來源： AutoGen 官網文檔】

規劃者作為系統的核心協調者，負責根據任務需求有條不紊地將具體任務分配給各個專業智能體，并確保整個工作流程的高效運行。金融分析師專注于使用 get_stock_data 工具深入分析股票數據和財務指標，為研究提供堅實的數據支持。新聞分析師則利用 get_news 工具廣泛收集和總結與股票相關的最新新聞資訊，并提取關鍵市場洞察。撰寫者負責將金融和新聞分析的結果精心整理成一份邏輯嚴密、內容詳實的最終報告。

每個智能體在完成自身任務后，都會將控制權交回給規劃者，規劃者根據整體進展情況決定是否繼續分配任務或終止流程。這種循環往復的協作機制，充分發揮了每個智能體的專業優勢，實現了高效的股票研究。

以下是具體的代碼實現：

    from typing import Any, Dict, Listfrom autogen_agentchat.agents import AssistantAgentfrom autogen_agentchat.conditions import HandoffTermination, TextMentionTerminationfrom autogen_agentchat.messages import HandoffMessagefrom autogen_agentchat.teams import Swarmfrom autogen_agentchat.ui import Consolefrom autogen_ext.models.openai import OpenAIChatCompletionClient# 定義獲取股票數據工具函數async def get_stock_data(symbol: str) -> Dict[str, Any]:return {"price": 180.25, "volume": 1000000, "pe_ratio": 65.4, "market_cap": "700B"}# 定義獲取新聞工具函數async def get_news(query: str) -> List[Dict[str, str]]:return [{"title": "Tesla Expands Cybertruck Production","date": "2024-03-20","summary": "Tesla ramps up Cybertruck manufacturing capacity at Gigafactory Texas, aiming to meet strong demand.",},{"title": "Tesla FSD Beta Shows Promise","date": "2024-03-19","summary": "Latest Full Self-Driving beta demonstrates significant improvements in urban navigation and safety features.",},{"title": "Model Y Dominates Global EV Sales","date": "2024-03-18","summary": "Tesla's Model Y becomes best-selling electric vehicle worldwide, capturing significant market share.",},]# 創建模型客戶端model_client = OpenAIChatCompletionClient(model="gpt-4o",# api_key="YOUR_API_KEY",)# 創建規劃者智能體planner = AssistantAgent("planner",model_client=model_client,handoffs=["financial_analyst", "news_analyst", "writer"],system_message="""You are a research planning coordinator. Coordinate market research by delegating to specialized agents: - Financial Analyst: For stock data analysis - News Analyst: For news gathering and analysis - Writer: For compiling final report Always send your plan first, then handoff to appropriate agent. Always handoff to a single agent at a time. Use TERMINATE when research is complete.""",)# 創建金融分析師智能體financial_analyst = AssistantAgent("financial_analyst",model_client=model_client,handoffs=["planner"],tools=[get_stock_data],system_message="""You are a financial analyst. Analyze stock market data using the get_stock_data tool. Provide insights on financial metrics. Always handoff back to planner when analysis is complete.""",)# 創建新聞分析師智能體news_analyst = AssistantAgent("news_analyst",model_client=model_client,handoffs=["planner"],tools=[get_news],system_message="""You are a news analyst. Gather and analyze relevant news using the get_news tool. Summarize key market insights from news. Always handoff back to planner when analysis is complete.""",)# 創建撰寫者智能體writer = AssistantAgent("writer",model_client=model_client,handoffs=["planner"],system_message="""You are a financial report writer. Compile research findings into clear, concise reports. Always handoff back to planner when writing is complete.""",)# 設置終止條件text_termination = TextMentionTermination("TERMINATE")termination = text_terminationresearch_team = Swarm(participants=[planner, financial_analyst, news_analyst, writer], termination_condition=termination)# 定義任務task = "Conduct market research for TSLA stock"await Console(research_team.run_stream(task=task))

代碼解讀：

同樣先導入所需模塊和類，然后定義了 get_stock_data 和 get_news 兩個工具函數，分別模擬獲取股票數據和新聞資訊的操作。
創建 OpenAIChatCompletionClient 模型客戶端，并指定模型為 gpt-4o。
接著創建了四個智能體：planner 作為規劃協調者，其系統消息明確了任務分配規則和流程；financial_analyst 配備了 get_stock_data 工具用于股票數據分析；news_analyst 利用 get_news 工具進行新聞收集和分析；writer 負責撰寫報告。每個智能體都設置了相應的任務交接目標和系統消息。
設置終止條件為 TextMentionTermination("TERMINATE")，當消息中提及 “TERMINATE” 時任務結束。
最后定義任務 “Conduct market research for TSLA stock” 并運行研究團隊任務，智能體之間按照設定的流程進行協作。

運行結果如下：

當用戶輸入 “Conduct market research for TSLA stock” 時，規劃者首先啟動任務分配：

---------- user ----------
Conduct market research for TSLA stock
---------- planner ----------
[FunctionCall(id='call_BX5QaRuhmB8CxTsBlqCUIXPb', arguments='{}', name='transfer_to_financial_analyst')]
[Prompt tokens: 169, Completion tokens: 166]
---------- planner ----------
[FunctionExecutionResult(content='Transferred to financial_analyst, adopting the role of financial_analyst immediately.', call_id='call_BX5QaRuhmB8CxTsBlqCUIXPb')]
---------- planner ----------
Transferred to financial_analyst, adopting the role of financial_analyst immediately.

金融分析師接收任務并調用工具獲取股票數據：

---------- financial_analyst ----------
[FunctionCall(id='call_SAXy1ebtA9mnaZo4ztp