探秘 OpenAI Swarm：重塑 AI 代理協作新模式 原創
本文主要介紹了 OpenAI 推出的實驗性框架 Swarm，它是一個多代理編排框架，致力于探索管理復雜 AI 代理交互的方式。文中闡述了 AI 代理的概念、組成及發展方向，強調 Swarm 的作用是協調多代理工作，通過任務分解、動態調度等機制，通過多個代碼示例，體現其在復雜任務處理上的優勢，還介紹了 Swarm 的核心概念如代理和任務移交，以及通過示例代碼展示其關鍵概念和功能實現，最后深入原理，從 Routine 到 Agent 的轉變及 Handoff 功能的運作方式。
引言：揭開 Swarm 的神秘面紗
AI Agent 的出現為人類帶來了前所未有的可能性，從簡單的任務執行到復雜問題的智能解決，Agent 正在改變我們與技術交互的方式。然而，隨著應用需求的增長，單一 Agent 的能力難以滿足復雜場景的多樣化需求。在許多情況下，解決問題不僅需要單個 Agent 的專業性，還需要多個 Agent 的協作，彼此分工明確、緊密配合，才能高效完成任務。如何協調和管理多個 Agent，讓它們各司其職又能無縫銜接，成為一個新的問題。
為此，OpenAI 提出了 Swarm ：多 Agent 協作架構，通過定義統一的智能體行為規范和動態任務轉交機制，為復雜場景中的 Agent 協作提供了解決方案。在接下來的內容中，我們將深入探索這一架構，通過理論描述和代碼實踐的方式，揭示 Swarm 的設計理念與運作原理。
什么是 Swarm？
Swarm 是 OpenAI 推出的一個實驗性框架，專為幫助開發者高效編排多代理系統（multi-agent systems）而設計。它于 2024 年正式發布，致力于探索簡潔、可擴展的方式來管理復雜的 AI 代理交互。Swarm 是開源的，托管在 GitHub 平臺上，開發者可以輕松獲取、嘗試和貢獻。
在本質上，Swarm 是一個多代理編排框架，旨在讓代理間的協調變得輕量化、可定制且易于測試。通過 Swarm，開發者可以構建、組織并管理多個 AI 代理（AI Agent），這些代理之間可以傳遞任務控制權，以共同完成復雜的工作流程。
各位看到這里就需要劃重點了， Swarm的作用是構建、組織并管理多個AI Agent，說白了，Swarm的作用就是協調多代理工作。
既然如此，為了搞懂Swarm 是個啥，我們就要先了解 AI 代理（AI Agent）。
如下圖所示，AI 代理（AI Agent）是能夠自主感知環境、做出決策并采取行動的系統，旨在實現更高層次的人工智能能力。在 AI 發展路徑中，AI 代理從解決單一任務（如生成文字、圖片、語音或視頻）逐步發展到能夠處理復雜任務（如金融投資、建筑設計或詩歌創作），最終朝著通用人工智能（AGI）的方向邁進，其目標是具備類似人類思考和行動的能力，能夠應對多樣化環境和復雜問題。
AI 代理的核心概念包括：
- 狀態（State）：代理在環境中所處的當前狀況。
- 智能體（Agent）：基于狀態信息選擇并執行行動（Action）。
- 獎勵（Reward）：根據代理行為對目標的影響、環境給予的反饋。
AI 代理的組成主要包括：
- LLM（大語言模型）：作為智能的核心，提供強大的語言理解和生成能力。
- 記憶：包括短期和長期記憶，用于跟蹤任務狀態和歷史數據。
- 工具：涵蓋代碼執行、搜索引擎、API 調用等功能模塊。
- 規則：定義代理行為的邊界，包括反思、自我批評和思維鏈條。
這種設計使得 AI 代理能夠在復雜環境中模擬人類決策過程，完成多樣化任務，并為通用人工智能的實現奠定基礎。
換句話說，AI 代理（Agent）也就是我們常說的智能體，指的是能夠執行特定任務或功能的獨立單位。每個代理通常具備某種能力或工具，能夠根據輸入進行處理，并輸出結果或執行某些操作。AI 代理不僅能夠根據預定規則和任務指令工作，還能夠在復雜的動態環境中做出決策、進行交互、協調和任務分配。
而Swarm 是一種讓智能體協作的工作方式，它讓每個智能體各司其職的同時，還能讓他們相互協作完成更加復雜的任務。OpenAI 推出Swarm的目的也是為大家在出一些復雜任務時具備一些“解題思路”。
比如說，有一家在線零售企業，客戶經常遇到多種問題，如咨詢產品信息、查詢訂單狀態以及申請退換貨等。這些需求往往需要不同的專屬服務，而傳統客服系統難以高效處理多樣化問題。通過 Swarm 框架，客服系統可以拆解任務并分配給多個專屬 AI 智能體。例如，當客戶咨詢產品信息時，推薦代理可以提供精確的產品推薦；當客戶查詢訂單時，訂單代理可以快速返回訂單狀態；如果客戶申請退貨，退貨代理能夠解釋政策并發起流程。Swarm架構就是讓這些智能體高效協作的實踐方法。
目前，Swarm是一個實驗性示例框架，用來探索多智能體系統的最佳實踐，為多智能體協作提供基礎研究。
在官方的介紹文檔中提到了Swarm 的兩個核心概念：代理（Agent）與任務移交（Handoff）。
- 代理（Agent）：代表執行特定任務或功能的單位，可以是具備特定技能或工具的獨立實體。
- 任務移交（Handoff）：允許一個代理根據當前上下文將任務委派給另一個更合適的代理。
這種任務的移交需要在生命代理之初就進行定義，除了移交以外，在初始化代理的時候還可以定義代理所具備的能力，包括函數或工具調用等。
為什么需要 Swarm？
Swarm 的核心價值在于其強大的任務分解能力、動態調度機制和多場景適配性，使其成為解決復雜問題的理想框架。對于人工智能系統而言，許多任務并非單一流程可以完成，而是由多個子任務組成。Swarm 的設計理念將復雜問題拆解為若干小任務，通過分工和動態任務調度，讓這些子任務相互銜接。
這種機制不僅提高了解決問題的效率，還大幅減少了上下文信息的冗余傳播。在此基礎上，Swarm 提供了強大的靈活性，可以適配從文檔分析到多模型協作的多種應用場景。這種能力的實現基于每個子智能體（Agent）專注于特定領域或任務，其運行嚴格遵循明確的策略與規則，同時主智能體負責協調和調度整個流程。
這么說可能太抽象，我們以航空公司客戶服務為例給大家說明。如下圖所示，當服務接收到客戶請求時，主智能體（Triage Agent）首先分析用戶輸入，并結合客戶上下文和航班信息，判斷問題的類型。例如，如果客戶詢問如何更改航班，主智能體會將任務傳遞給專注于航班修改的智能體（Flight Modification Agent）。該智能體進一步對問題進行細化，判斷用戶的需求是取消航班還是改簽航班。如果用戶希望取消航班，它會將任務傳遞給取消航班智能體（Flight Cancel），負責處理退款或生成積分的具體操作；如果用戶需要改簽航班，任務會被傳遞給改簽智能體（Flight Change），完成航班改簽流程。如果在改簽和取消航班時遇到問題，智能體還可以將請求再交回給主智能體進行處理，在圖中可以看到紅線的部分就是交回請求。
在另一個場景中，如果客戶報告行李丟失，主智能體會將問題轉交給行李管理智能體（Lost Baggage），該模塊會立即啟動行李搜索流程。如果找到行李，系統會安排將其送達客戶地址；如果未能找到，也可以通過紅線將請求交回給主智能體。
這種分工明確的設計確保了每個問題都由最適合的模塊來處理，而動態調度機制則使得任務能夠以最短路徑被解決。同時，也體現了 Swarm 的核心優勢：通過任務分解，系統將復雜的客戶服務流程劃分為一個個小模塊；通過動態調度，主智能體和子智能體之間的任務交接變得高效流暢；通過策略化引導，子智能體在完成特定任務時能夠嚴格遵循規則，確保用戶體驗的一致性與任務的準確性。每個子任務的獨立性使得系統易于擴展，當新的需求出現時，只需添加對應的子智能體即可，而無需改動現有架構。這不僅降低了系統的復雜性，還顯著提高了可維護性。
Swarm 的實戰
前面對Swarm 進行簡單介紹之后，我們來嘗試安裝Swarm 并通過實例體驗一下它的“魅力”。通過如下指令安裝Swarm：
pip install git+ssh://git@github.com/openai/swarm.git接著，通過一個簡單代碼來看看它是如何讓Agent 工作的。
如下圖所示，該示例展示了多語言代理切換功能，具體而言實現了英語、西班牙語和中文三種語言代理之間的智能切換。英語代理作為主代理接受用戶的請求，當發現用戶請求的內容是中文的時候，轉交給中文代理進行處理。需要注意的是，這里的請求移交是通過兩個預定義函數完成的，分別是transfer_to_chinese_agent 將請求轉交給中文代理，以及transfer_to_spanish_agent 將請求轉交給西班牙語代理。
詳細代碼如下：
from swarm import Swarm, Agent
# 創建Swarm客戶端實例
client = Swarm()
# 創建英語代理實例
english_agent = Agent(
# 設置代理的名稱為"English Agent"
name="English Agent",
# 設置代理的指令 - 只使用英語交流
instructions="You only speak English.",
)
# 創建西班牙語代理實例
spanish_agent = Agent(
name="Spanish Agent",
# 設置代理只使用西班牙語交流
instructions="You only speak Spanish.",
)
# 創建中文代理實例
chinese_agent = Agent(
name="Chinese Agent",
# 設置代理只使用中文交流
instructions="You only speak Chinese.",
)
# 定義轉移到西班牙語代理的函數
def transfer_to_spanish_agent():
"""Transfer spanish speaking users immediately."""
return spanish_agent
# 定義轉移到中文代理的函數
def transfer_to_chinese_agent():
"""Transfer chinese speaking users immediately."""
return chinese_agent
# 將轉移函數添加到英語代理的功能列表中
english_agent.functions.append(transfer_to_spanish_agent)
english_agent.functions.append(transfer_to_chinese_agent)
# 創建用戶消息 - 這里使用中文進行測試
messages = [{"role": "user", "content": "你好， 你是誰？"}] # 中文測試消息
# 運行代理并獲取響應
response = client.run(agent=english_agent, messages=messages)
# 打印最后一條響應消息
print(response.messages[-1]["content"])重點代碼解析如下：
1. 代理創建部分
english_agent = Agent(
name="English Agent",
instructions="You only speak English.",
)創建了一個英語代理，通過instructions指令，利用提示詞工程使大模型扮演說英語的代理角色，負責使用英語與用戶交流。中文和西班牙語的代理創建也是采用相同模式。
2. 切換函數定義
def transfer_to_chinese_agent():
"""Transfer chinese speaking users immediately."""
return chinese_agent該函數實現了向中文代理的移交請求功能，當系統檢測到用戶使用中文時會觸發此函數。與之相同的還有一個transfer_to_spanish_agent，它負責移交請求給西班牙語代理。
3. 功能注冊
english_agent.functions.append(transfer_to_spanish_agent)
english_agent.functions.append(transfer_to_chinese_agent)由于我們假設英文代理作為主代理，它負責用戶請求的轉交，當發現是英語請求的時候它會自己處理，如果發現是其他兩種語言的時候，通過預定義的函數功能實現請求轉交。這段代碼將請求轉交功能注冊到英語代理中。
4. 測試代碼
# 創建用戶消息 - 這里使用中文進行測試
messages = [{"role": "user", "content": "你好， 你是誰？"}] # 中文測試消息
# 運行代理并獲取響應
response = client.run(agent=english_agent, messages=messages)接著，使用Swarm實例化的客戶端 client進行測試，我們輸入中文"你好， 你是誰？"并通過Swarm類的run方法傳入主代理english_agent 和messages，通過response返回測試結果如下：
你好，我是一個智能助手，可以幫助你解決各種問題。你有什么需要幫助的嗎？顯然，英語代理識別出中文輸入，并將請求轉交給中文代理執行。
Swarm 核心概念
通過上面Swarm 示例代碼，可以理解 Swarm 的關鍵概念及功能實現：
1. client.run() 方法
client.run() 是 Swarm 中的核心方法，類似于 OpenAI 的 chat.completions.create() 方法。它接受消息輸入并返回消息輸出，同時在多輪調用之間不保存狀態。除了處理消息對話，還支持以下功能：
- 執行 Agent 的函數調用并追加結果。
- 在任務完成后轉交給其他 Agent（handoffs）。
- 動態更新上下文變量（context variables）。
- 在必要時支持多輪對話再返回結果。
例如，在代碼中，我們通過 client.run() 將初始用戶消息傳遞給英語代理（English Agent）。當系統檢測到輸入的語言不符合代理的要求時，會調用代理函數，切換到適合語言的代理，如 Spanish Agent 或 Chinese Agent。
2. Agents（代理）
Agent 是任務執行的基本單元。它可以被看作一個封裝了特定指令（instructions）和功能（functions）的“智能體”。Agent 的核心字段包括：
- name：代理的名稱，用于標識。
- instructions：代理的指令，決定代理的行為方式。
- functions：代理可以調用的一組 Python 函數，用于執行特定任務。
- handoff：代理可以通過函數切換到其他代理。
代碼中，我們創建了三個代理：English Agent、Spanish Agent 和 Chinese Agent，它們分別按照語言設置了特定的指令。代理通過 instructions 告知系統其行為規范，例如僅使用某種語言交流。
3. Functions（函數）
Swarm 的代理支持直接調用 Python 函數執行任務。函數通常返回一個字符串（str），也可以通過返回一個代理（Agent）實現代理之間的切換，或通過修改上下文變量（context_variables）來動態改變對話狀態。
在示例中，transfer_to_spanish_agent 和 transfer_to_chinese_agent 是兩個函數，它們的作用是根據用戶的語言輸入，將當前任務切換到對應的語言代理。
如果一個代理調用了多個函數，Swarm 會按照順序依次執行它們。如果函數出現錯誤（如參數缺失或類型錯誤），系統會生成一個錯誤響應，并嘗試從錯誤中恢復。
4. Handoffs and Updating Context Variables（代理切換和上下文變量更新）
代理轉交（Handoff）在代理無法完成任務時，可以將任務交接給其他代理。例如，English Agent 調用 transfer_to_spanish_agent 函數后，返回了 Spanish Agent，實現了從英語代理到西班牙語代理的切換。
此外，Swarm 還支持動態更新上下文變量。通過返回一個 Result 對象，函數可以同時更新返回值、切換代理和修改上下文變量。這種能力確保了復雜任務流程中每個步驟的信息傳遞和狀態保持一致。
5. Function Schemas（函數模式）
Swarm 能夠自動將函數轉換為 JSON Schema，以便代理理解函數的功能和參數需求：
- 函數的文檔字符串（docstring）會轉化為函數的描述信息。
- 參數的類型提示（type hints）會映射到 JSON Schema 的字段類型。
- 如果函數定義了必需參數，則會自動標記為 required。
Swarm 是如何運作的 ？
Swarm 的運作核心在于通過多個智能體（Agent）的協同工作，實現復雜任務的高效處理。為了讓大家對這個過程有深入的了解，接下來，我們會從Routine 概念開始，逐步演化為具備智能能力的 Agent，并最終依靠 Handoff 功能實現動態任務分配。
Routine 是 Swarm 的基礎單元，由預定義的指令和工具/函數組成，負責描述任務的邏輯流程及完成任務所需的工具。而 Agent 是一種強化版的 Routine，通過結合大語言模型（LLM），賦予其智能理解和決策能力，使其不僅能夠執行預定義任務，還能靈活應對復雜的用戶需求。在此基礎上，Swarm 的 Handoff 功能讓不同的 Agent 能夠根據請求的內容動態協作，將用戶的任務無縫轉交給更適合的 Agent 處理，并完整保留對話上下文，避免用戶重復輸入。
什么是 Routine？
簡單來說，Routine 是一組預定義的指令與相應工具的組合，旨在完成特定任務。它不僅是一個執行步驟的計劃，還包含完成任務所需的資源和工具。我們可以簡單理解Routine 就是Agent（代理）的雛形，它描述了代理需要完成的任務，同時還賦予它對應的工具/函數，只是Agent 本身還具備大模型的能力還可以進行“思考”。
如下圖所示，Routine可以拆解為兩部分：
- 指令（Instructions）：以自然語言或系統提示的形式描述的任務執行步驟。
- 工具/函數（Tools/function）：完成這些步驟所需的工具或函數。需要說明的是，無論是調用工具或者函數都會通過函數調用的方式，也就是function call的方式完成。
說白了，Routine就是系統提示：描述任務的邏輯流程（比如詢問問題、搜索信息或處理用戶請求），加上可調用的工具/函數（用來輔助完成這些任務）。
來個具體的例子，我們為客戶服務代理定義了一個例程，指示其對用戶問題進行分類，然后建議修復或提供退款。同時定義函數execute_refund和look_up_item 作為外部工具協助完成退貨以及查找訂單的工作。下面就是客戶服務例程的代碼：指令和工具/函數。
system_message = (
"你是一名 ACME Inc. 的客戶服務代理，以下是你的工作流程：\n"
"1. 首先，向用戶詢問更多信息以理解他們的問題（如果問題未明確）。\n"
"2. 提出一個解決方案。\n"
"3. 如果用戶不滿意，提供退款。\n"
"4. 如果接受退款，查找物品 ID 并執行退款操作。"
)
def look_up_item(search_query):
"""Use to find item ID.
Search query can be a description or keywords."""
# return hard-coded item ID - in reality would be a lookup
return "item_132612938"
def execute_refund(item_id, reason="not provided"):
print("Summary:", item_id, reason) # lazy summary
return "success"如何執行 Routine？
為了執行 Routine，需要實現從用戶交互到模型調用、工具執行的完整閉環。以下是幾個關鍵部分：
1.函數定義（Function Definition）
函數是 Routine 的核心工具。它們可以是具體的業務邏輯實現，例如處理退款或查詢信息。這也就是前面描述的execute_refund和look_up_item 兩個函數的定義，代碼在上面已經展示過了， 這里就不贅述了。需要說明的是，look_up_item 執行訂單查詢，execute_refund 負責退款操作。
2.函數接口（Function Schema）
當語言模型接到用戶請求的時候，需要調用函數返回對應的結果，調用函數時需要使用Function Schema。它包括：函數的名稱、參數及其類型等信息。為此需要將 Python 函數定義轉化為標準化的 Schema（模式）。為了實現從函數到函數接口的轉換需要實現如下代碼：
import inspect
def function_to_schema(func) -> dict:
"""將函數定義轉化為 Schema 格式。"""
type_map = {
str: "string",
int: "integer",
float: "number",
bool: "boolean",
list: "array",
dict: "object",
type(None): "null",
}
# 獲取函數簽名
signature = inspect.signature(func)
parameters = {}
for param in signature.parameters.values():
param_type = type_map.get(param.annotation, "string")
parameters[param.name] = {"type": param_type}
required = [
param.name
for param in signature.parameters.values()
if param.default == inspect._empty
]
return {
"type": "function",
"function": {
"name": func.__name__,
"description": (func.__doc__ or "").strip(),
"parameters": {
"type": "object",
"properties": parameters,
"required": required,
},
},
}這個代碼主要通過輸入函數句柄，提取函數名稱、描述、輸入參數等信息，從而生成函數接口（Function schema）。我們可以用execute_refund 函數測試其效果。代碼如下：
# 示例輸出
schema = function_to_schema(execute_refund)
print(schema)
結果：
{
"type": "function",
"function": {
"name": "execute_refund",
"description": "執行退款操作。",
"parameters": {
"type": "object",
"properties": {
"item_id": { "type": "string" },
"reason": { "type": "string" }
},
"required": ["item_id"]
}
}
}
3. 函數調用（Function Call）
完成函數定義和調用函數接口之后，接下來就是利用函數接口去調用對應的函數。需要將 Schema 注冊為工具，供模型生成調用指令。然后，根據調用結果執行對應函數，最后將結果返回給模型。執行代碼如下：
# 定義可用的工具函數列表
tools = [execute_refund, look_up_item]
def run_full_turn(system_message, tools, messages):
# 記錄初始消息數量
num_init_messages = len(messages)
# 復制消息列表以避免修改原始數據
messages = messages.copy()
while True:
# 將Python函數轉換為OpenAI工具模式
tool_schemas = [function_to_schema(tool) for tool in tools]
# 創建工具名稱到函數的映射字典
tools_map = {tool.__name__: tool for tool in tools}
# 調用OpenAI API獲取回復
response = client.chat.completions.create(
model="gpt-4o-mini",
messages=[{"role": "system", "content": system_message}] + messages,
tools=tool_schemas or None,
)
# 獲取AI助手的回復
message = response.choices[0].message
# 將助手回復添加到消息歷史
messages.append(message)
# 如果有文本回復則打印
if message.content:
print("Assistant:", message.content)
# 如果沒有工具調用請求則結束循環
if not message.tool_calls:
break
# 處理工具調用
for tool_call in message.tool_calls:
# 執行工具調用并獲取結果
result = execute_tool_call(tool_call, tools_map)
# 將工具調用結果添加到消息歷史
result_message = {
"role": "tool",
"tool_call_id": tool_call.id,
"content": result,
}
messages.append(result_message)
# 返回新增的消息
return messages[num_init_messages:]
def execute_tool_call(tool_call, tools_map):
# 獲取要調用的函數名
name = tool_call.function.name
# 解析函數參數
args = json.loads(tool_call.function.arguments)
# 打印調用信息
print(f"Assistant: {name}({args})")
# 執行函數調用并返回結果
return tools_map[name](**args)
# 主循環
messages = []
while True:
# 獲取用戶輸入
user = input("User: ")
# 添加用戶消息到歷史
messages.append({"role": "user", "content": user})
# 執行一輪對話并獲取新消息
new_messages = run_full_turn(system_message, tools, messages)
# 將新消息添加到歷史中
messages.extend(new_messages)在這段代碼中，通過大模型和Routine構建了一個交互系統，用來讓客服代理與用戶進行對話。我們截取部分重要段落給大家進行拆解如下：
首先，定義一組工具函數（如 execute_refund 和 look_up_item），這些工具提供了解決具體業務需求的能力。它們被集中存儲在一個列表中，隨后通過工具模式（Schema）轉化為模型可以識別和調用的接口描述，從而成為語言模型的可用擴展。
tools = [execute_refund, look_up_item]
tool_schemas = [function_to_schema(tool) for tool in tools]function_to_schema 方法將 Python 函數轉換為模型可理解的工具描述，方便系統動態調用這些函數。
對話流程的核心由 run_full_turn 函數實現，它負責完成一輪從用戶輸入到工具調用再到生成助手回復的整個邏輯。其實現的關鍵在于利用 OpenAI 模型生成回復并解析是否需要工具調用。每次交互開始時，系統會加載當前的消息歷史以及可用工具的模式列表，然后通過調用模型接口獲取助手的回復。如果模型未請求調用工具，則直接返回助手的回復；否則系統將根據工具調用的描述執行相應的函數。
response = client.chat.completions.create(
model="gpt-4o-mini",
messages=[{"role": "system", "content": system_message}] + messages,
tools=tool_schemas or None,
)工具調用的具體執行邏輯由 execute_tool_call 函數實現。每當模型發出工具調用請求時，該函數會根據請求中描述的工具名稱和參數，查找與之對應的 Python 函數并執行操作。例如，若助手要求退款操作，則函數會解析出對應的函數名稱和參數，然后調用 execute_refund 完成任務，返回結果后更新對話歷史。
def execute_tool_call(tool_call, tools_map):
name = tool_call.function.name
args = json.loads(tool_call.function.arguments)
return tools_map[name](**args)最后，通過一個主循環實現用戶與助手的持續交互。用戶的輸入被實時加入消息歷史，隨后調用 run_full_turn 生成助手回復，并在必要時調用外部工具函數。所有新增的消息和工具調用結果都會被動態更新到對話歷史中，保證了上下文的連貫性。
while True:
user = input("User: ")
messages.append({"role": "user", "content": user})
new_messages = run_full_turn(system_message, tools, messages)
messages.extend(new_messages)從Routine到Agent
到目前為止，我們得到了Routine，它可以理解為指令和工具的集合，為了讓它獨立工作還需要加入大語言模型（LLM），這樣才能過渡到Swarm中的Agent形態。這里我們將智能體（Agent）理解為一種強化版的 Routine，即 Routine 加上了語言模型（LLM）的智能能力。Routine 負責定義工具和執行步驟，而 Agent 則通過結合 LLM 具備了智能理解和處理能力。具體來說，Agent 包括以下幾個核心要素：
- 名稱（Name）：用于標識 Agent 的職責。
- 模型（Model）：用來理解用戶請求，處理工具函數調用，并給予用戶反饋。
- 指令（Instructions）：定義 Agent 如何執行其任務。
- 工具（Tools）：該 Agent 可以調用的一組函數。
我們可以通過如下代碼來定義Agent：
class Agent(BaseModel):
name: str = "Agent"
model: str = "gpt-4o-mini"
instructions: str = "You are a helpful Agent"
tools: list = []順著這個思路，我們可以定義更多的代理如下：
def execute_refund(item_name):
return "success"
refund_agent = Agent(
name="Refund Agent",
instructions="You are a refund agent. Help the user with refunds.",
tools=[execute_refund],
)
def place_order(item_name):
return "success"
sales_assistant = Agent(
name="Sales Assistant",
instructions="You are a sales assistant. Sell the user a product.",
tools=[place_order],
)上面代碼定義了退貨和銷售代理，分別針對兩個代理都定義了工具/函數，協助他們完成工作。
Handoff轉交功能
好！目前，我們已經理解了Routine 并且順利從Routine的概念過渡到了Agent，還記得我們在“什么是Swarm”章節中介紹Swarm結構的核心就是Agent和Handoff， 接著我們就來介紹Handoff 功能。前面的內容中也提到了， 在當前代理無法處理對應請求的時候，該代理會將請求轉交給其他代理處理。
于是，我們修改執行代碼如下：
def run_full_turn(agent, messages):
# 初始化當前智能體為傳入的 agent
current_agent = agent
# 記錄初始消息數量，用于之后返回新增消息
num_init_messages = len(messages)
# 復制消息列表，防止對原始消息數據造成影響
messages = messages.copy()
while True:
# 將當前智能體的工具列表轉化為工具模式（schemas），用于 API 調用
tool_schemas = [function_to_schema(tool) for tool in current_agent.tools]
# 構造工具名稱到工具函數的映射
tools = {tool.__name__: tool for tool in current_agent.tools}
# === 1. 調用 OpenAI 接口生成回復 ===
response = client.chat.completions.create(
model=agent.model, # 當前智能體使用的模型
messages=[{"role": "system", "content": current_agent.instructions}]
+ messages, # 包括系統消息和歷史對話內容
tools=tool_schemas or None, # 提供工具的模式定義
)
# 獲取生成的消息
message = response.choices[0].message
# 將回復消息添加到歷史記錄中
messages.append(message)
# 如果消息包含文本內容，則打印當前智能體的回復
if message.content:
print(f"{current_agent.name}:", message.content)
# 如果沒有工具調用請求，則退出循環
if not message.tool_calls:
break
# === 2. 處理工具調用 ===
for tool_call in message.tool_calls:
# 執行工具調用，并返回結果
result = execute_tool_call(tool_call, tools, current_agent.name)
# 如果工具調用結果是一個新的智能體對象，則進行切換
if type(result) is Agent:
current_agent = result # 更新當前智能體為新的智能體
result = (
f"Transfered to {current_agent.name}. Adopt persona immediately."
) # 生成切換通知
# 將工具調用的結果作為消息添加到歷史記錄中
result_message = {
"role": "tool",
"tool_call_id": tool_call.id,
"content": result,
}
messages.append(result_message)
# ==== 3. 返回最后使用的智能體和新增的消息記錄 ====
return Response(agent=current_agent, messages=messages[num_init_messages:])
def execute_tool_call(tool_call, tools, agent_name):
# 提取工具名稱
name = tool_call.function.name
# 解析工具調用的參數
args = json.loads(tool_call.function.arguments)
# 打印工具調用的詳細信息
print(f"{agent_name}:", f"{name}({args})")
# 調用對應的工具函數并返回其結果
return tools[name](**args)我們把目光放到與Handoff 相關的細節上，工具調用返回智能體對象。在處理工具調用時，execute_tool_call 的返回結果是 Agent 對象，說明該工具無法完成用戶的請求，通過返回Agent的方式讓請求切換到其他Agent中去。這里需要注意的是，我們將Handoff的動作也封裝成了函數，該函數會直接返回要切換到的Agent對象，他的具體應用就在此處。
在run_full_turn函數中通過 type(result) 判斷，如果返回的是 Agent 類型，則說明需要進行智能體切換。
# 執行工具調用，并返回結果
result = execute_tool_call(tool_call, tools, current_agent.name)
if type(result) is Agent: # 如果工具調用返回一個新的 Agent
current_agent = result # 更新當前智能體
result = (
f"Transfered to {current_agent.name}. Adopt persona immediately."
) # 通知用戶智能體切換除此之外，我們還需要動態更新 current_agent 的工具和說明，確保切換后的行為符合目標智能體的能力。為了讓對話記錄延續，需要通過消息列表的復制和擴展（messages.copy()），保留了所有上下文信息。
總結
Swarm 作為 OpenAI 的實驗性框架，旨在解決復雜場景下多代理協作問題。AI 代理具備自主感知、決策和行動能力，Swarm 則讓多個代理協同工作。其優勢在于強大的任務分解和動態調度，適用于多種場景，如航空公司客戶服務等。在實戰示例中，通過代碼展示了多語言代理切換功能，體現了 Swarm 的核心方法、代理、函數等概念及功能。深入原理部分，Routine 是基礎單元，執行 Routine 需實現函數定義、接口和調用等環節，Agent 是強化版 Routine，結合了大語言模型的智能能力，Handoff 功能可在代理無法處理請求時進行任務轉交，通過一系列操作確保對話上下文延續及智能體切換后的行為符合要求。
作者介紹
崔皓，51CTO社區編輯，資深架構師，擁有18年的軟件開發和架構經驗，10年分布式架構經驗。
