作者 | 崔皓

審校 | 重樓

本文主要介紹了 OpenAI 推出的實驗性框架 Swarm，它是一個多代理編排框架，致力于探索管理復雜 AI 代理交互的方式。文中闡述了 AI 代理的概念、組成及發(fā)展方向，強調(diào) Swarm 的作用是協(xié)調(diào)多代理工作，通過任務分解、動態(tài)調(diào)度等機制，通過多個代碼示例，體現(xiàn)其在復雜任務處理上的優(yōu)勢，還介紹了 Swarm 的核心概念如代理和任務移交，以及通過示例代碼展示其關鍵概念和功能實現(xiàn)，最后深入原理，從 Routine 到 Agent 的轉變及 Handoff 功能的運作方式。

引言：揭開 Swarm 的神秘面紗

AI Agent 的出現(xiàn)為人類帶來了前所未有的可能性，從簡單的任務執(zhí)行到復雜問題的智能解決，Agent 正在改變我們與技術交互的方式。然而，隨著應用需求的增長，單一 Agent 的能力難以滿足復雜場景的多樣化需求。在許多情況下，解決問題不僅需要單個 Agent 的專業(yè)性，還需要多個 Agent 的協(xié)作，彼此分工明確、緊密配合，才能高效完成任務。如何協(xié)調(diào)和管理多個 Agent，讓它們各司其職又能無縫銜接，成為一個新的問題。

為此，OpenAI 提出了 Swarm ：多 Agent 協(xié)作架構，通過定義統(tǒng)一的智能體行為規(guī)范和動態(tài)任務轉交機制，為復雜場景中的 Agent 協(xié)作提供了解決方案。在接下來的內(nèi)容中，我們將深入探索這一架構，通過理論描述和代碼實踐的方式，揭示 Swarm 的設計理念與運作原理。

什么是 Swarm？

Swarm 是 OpenAI 推出的一個實驗性框架，專為幫助開發(fā)者高效編排多代理系統(tǒng)（multi-agent systems）而設計。它于 2024 年正式發(fā)布，致力于探索簡潔、可擴展的方式來管理復雜的 AI 代理交互。Swarm 是開源的，托管在 GitHub 平臺上，開發(fā)者可以輕松獲取、嘗試和貢獻。

在本質(zhì)上，Swarm 是一個多代理編排框架，旨在讓代理間的協(xié)調(diào)變得輕量化、可定制且易于測試。通過 Swarm，開發(fā)者可以構建、組織并管理多個 AI 代理（AI Agent），這些代理之間可以傳遞任務控制權，以共同完成復雜的工作流程。

各位看到這里就需要劃重點了， Swarm的作用是構建、組織并管理多個AI Agent，說白了，Swarm的作用就是協(xié)調(diào)多代理工作。

既然如此，為了搞懂Swarm 是個啥，我們就要先了解 AI 代理（AI Agent）。

如下圖所示，AI 代理（AI Agent）是能夠自主感知環(huán)境、做出決策并采取行動的系統(tǒng)，旨在實現(xiàn)更高層次的人工智能能力。在 AI 發(fā)展路徑中，AI 代理從解決單一任務（如生成文字、圖片、語音或視頻）逐步發(fā)展到能夠處理復雜任務（如金融投資、建筑設計或詩歌創(chuàng)作），最終朝著通用人工智能（AGI）的方向邁進，其目標是具備類似人類思考和行動的能力，能夠應對多樣化環(huán)境和復雜問題。

AI 代理的核心概念包括：

• 狀態(tài)（State）：代理在環(huán)境中所處的當前狀況。
• 智能體（Agent）：基于狀態(tài)信息選擇并執(zhí)行行動（Action）。
• 獎勵（Reward）：根據(jù)代理行為對目標的影響、環(huán)境給予的反饋。

AI 代理的組成主要包括：

• LLM（大語言模型）：作為智能的核心，提供強大的語言理解和生成能力。
• 記憶：包括短期和長期記憶，用于跟蹤任務狀態(tài)和歷史數(shù)據(jù)。
• 工具：涵蓋代碼執(zhí)行、搜索引擎、API 調(diào)用等功能模塊。
• 規(guī)則：定義代理行為的邊界，包括反思、自我批評和思維鏈條。

這種設計使得 AI 代理能夠在復雜環(huán)境中模擬人類決策過程，完成多樣化任務，并為通用人工智能的實現(xiàn)奠定基礎。

換句話說，AI 代理（Agent）也就是我們常說的智能體，指的是能夠執(zhí)行特定任務或功能的獨立單位。每個代理通常具備某種能力或工具，能夠根據(jù)輸入進行處理，并輸出結果或執(zhí)行某些操作。AI 代理不僅能夠根據(jù)預定規(guī)則和任務指令工作，還能夠在復雜的動態(tài)環(huán)境中做出決策、進行交互、協(xié)調(diào)和任務分配。

而Swarm 是一種讓智能體協(xié)作的工作方式，它讓每個智能體各司其職的同時，還能讓他們相互協(xié)作完成更加復雜的任務。OpenAI 推出Swarm的目的也是為大家在出一些復雜任務時具備一些“解題思路”。

比如說，有一家在線零售企業(yè)，客戶經(jīng)常遇到多種問題，如咨詢產(chǎn)品信息、查詢訂單狀態(tài)以及申請退換貨等。這些需求往往需要不同的專屬服務，而傳統(tǒng)客服系統(tǒng)難以高效處理多樣化問題。通過 Swarm 框架，客服系統(tǒng)可以拆解任務并分配給多個專屬 AI 智能體。例如，當客戶咨詢產(chǎn)品信息時，推薦代理可以提供精確的產(chǎn)品推薦；當客戶查詢訂單時，訂單代理可以快速返回訂單狀態(tài)；如果客戶申請退貨，退貨代理能夠解釋政策并發(fā)起流程。Swarm架構就是讓這些智能體高效協(xié)作的實踐方法。

目前，Swarm是一個實驗性示例框架，用來探索多智能體系統(tǒng)的最佳實踐，為多智能體協(xié)作提供基礎研究。

在官方的介紹文檔中提到了Swarm 的兩個核心概念：代理（Agent）與任務移交（Handoff）。

• 代理（Agent）：代表執(zhí)行特定任務或功能的單位，可以是具備特定技能或工具的獨立實體。
• 任務移交（Handoff）：允許一個代理根據(jù)當前上下文將任務委派給另一個更合適的代理。

這種任務的移交需要在生命代理之初就進行定義，除了移交以外，在初始化代理的時候還可以定義代理所具備的能力，包括函數(shù)或工具調(diào)用等。

為什么需要 Swarm？

Swarm 的核心價值在于其強大的任務分解能力、動態(tài)調(diào)度機制和多場景適配性，使其成為解決復雜問題的理想框架。對于人工智能系統(tǒng)而言，許多任務并非單一流程可以完成，而是由多個子任務組成。Swarm 的設計理念將復雜問題拆解為若干小任務，通過分工和動態(tài)任務調(diào)度，讓這些子任務相互銜接。

這種機制不僅提高了解決問題的效率，還大幅減少了上下文信息的冗余傳播。在此基礎上，Swarm 提供了強大的靈活性，可以適配從文檔分析到多模型協(xié)作的多種應用場景。這種能力的實現(xiàn)基于每個子智能體（Agent）專注于特定領域或任務，其運行嚴格遵循明確的策略與規(guī)則，同時主智能體負責協(xié)調(diào)和調(diào)度整個流程。

這么說可能太抽象，我們以航空公司客戶服務為例給大家說明。如下圖所示，當服務接收到客戶請求時，主智能體（Triage Agent）首先分析用戶輸入，并結合客戶上下文和航班信息，判斷問題的類型。例如，如果客戶詢問如何更改航班，主智能體會將任務傳遞給專注于航班修改的智能體（Flight Modification Agent）。該智能體進一步對問題進行細化，判斷用戶的需求是取消航班還是改簽航班。如果用戶希望取消航班，它會將任務傳遞給取消航班智能體（Flight Cancel），負責處理退款或生成積分的具體操作；如果用戶需要改簽航班，任務會被傳遞給改簽智能體（Flight Change），完成航班改簽流程。如果在改簽和取消航班時遇到問題，智能體還可以將請求再交回給主智能體進行處理，在圖中可以看到紅線的部分就是交回請求。

在另一個場景中，如果客戶報告行李丟失，主智能體會將問題轉交給行李管理智能體（Lost Baggage），該模塊會立即啟動行李搜索流程。如果找到行李，系統(tǒng)會安排將其送達客戶地址；如果未能找到，也可以通過紅線將請求交回給主智能體。

這種分工明確的設計確保了每個問題都由最適合的模塊來處理，而動態(tài)調(diào)度機制則使得任務能夠以最短路徑被解決。同時，也體現(xiàn)了 Swarm 的核心優(yōu)勢：通過任務分解，系統(tǒng)將復雜的客戶服務流程劃分為一個個小模塊；通過動態(tài)調(diào)度，主智能體和子智能體之間的任務交接變得高效流暢；通過策略化引導，子智能體在完成特定任務時能夠嚴格遵循規(guī)則，確保用戶體驗的一致性與任務的準確性。每個子任務的獨立性使得系統(tǒng)易于擴展，當新的需求出現(xiàn)時，只需添加對應的子智能體即可，而無需改動現(xiàn)有架構。這不僅降低了系統(tǒng)的復雜性，還顯著提高了可維護性。

Swarm 的實戰(zhàn)

前面對Swarm 進行簡單介紹之后，我們來嘗試安裝Swarm 并通過實例體驗一下它的“魅力”。通過如下指令安裝Swarm：

pip install git+ssh://git@github.com/openai/swarm.git

接著，通過一個簡單代碼來看看它是如何讓Agent 工作的。

如下圖所示，該示例展示了多語言代理切換功能，具體而言實現(xiàn)了英語、西班牙語和中文三種語言代理之間的智能切換。英語代理作為主代理接受用戶的請求，當發(fā)現(xiàn)用戶請求的內(nèi)容是中文的時候，轉交給中文代理進行處理。需要注意的是，這里的請求移交是通過兩個預定義函數(shù)完成的，分別是transfer_to_chinese_agent 將請求轉交給中文代理，以及transfer_to_spanish_agent 將請求轉交給西班牙語代理。

詳細代碼如下：

from swarm import Swarm, Agent
# 創(chuàng)建Swarm客戶端實例
client = Swarm()
# 創(chuàng)建英語代理實例
english_agent = Agent(
    # 設置代理的名稱為"English Agent"
    name="English Agent",
    # 設置代理的指令 - 只使用英語交流
    instructions="You only speak English.",
)
# 創(chuàng)建西班牙語代理實例
spanish_agent = Agent(
    name="Spanish Agent", 
    # 設置代理只使用西班牙語交流
    instructions="You only speak Spanish.",
)
# 創(chuàng)建中文代理實例
chinese_agent = Agent(
    name="Chinese Agent",
    # 設置代理只使用中文交流
    instructions="You only speak Chinese.",
)
# 定義轉移到西班牙語代理的函數(shù)
def transfer_to_spanish_agent():
    """Transfer spanish speaking users immediately."""
    return spanish_agent
# 定義轉移到中文代理的函數(shù)
def transfer_to_chinese_agent():
    """Transfer chinese speaking users immediately."""
    return chinese_agent
# 將轉移函數(shù)添加到英語代理的功能列表中
english_agent.functions.append(transfer_to_spanish_agent)
english_agent.functions.append(transfer_to_chinese_agent)
# 創(chuàng)建用戶消息 - 這里使用中文進行測試
messages = [{"role": "user", "content": "你好， 你是誰？"}]  # 中文測試消息
# 運行代理并獲取響應
response = client.run(agent=english_agent, messages=messages)
# 打印最后一條響應消息
print(response.messages[-1]["content"])

重點代碼解析如下：

1. 代理創(chuàng)建部分

english_agent = Agent(
    name="English Agent",
    instructions="You only speak English.",
)

創(chuàng)建了一個英語代理，通過instructions指令，利用提示詞工程使大模型扮演說英語的代理角色，負責使用英語與用戶交流。中文和西班牙語的代理創(chuàng)建也是采用相同模式。

2. 切換函數(shù)定義

def transfer_to_chinese_agent():
    """Transfer chinese speaking users immediately."""
    return chinese_agent

該函數(shù)實現(xiàn)了向中文代理的移交請求功能，當系統(tǒng)檢測到用戶使用中文時會觸發(fā)此函數(shù)。與之相同的還有一個transfer_to_spanish_agent，它負責移交請求給西班牙語代理。

3. 功能注冊

english_agent.functions.append(transfer_to_spanish_agent)
english_agent.functions.append(transfer_to_chinese_agent)

由于我們假設英文代理作為主代理，它負責用戶請求的轉交，當發(fā)現(xiàn)是英語請求的時候它會自己處理，如果發(fā)現(xiàn)是其他兩種語言的時候，通過預定義的函數(shù)功能實現(xiàn)請求轉交。這段代碼將請求轉交功能注冊到英語代理中。

4. 測試代碼

# 創(chuàng)建用戶消息 - 這里使用中文進行測試
messages = [{"role": "user", "content": "你好， 你是誰？"}]  # 中文測試消息
# 運行代理并獲取響應
response = client.run(agent=english_agent, messages=messages)

接著，使用Swarm實例化的客戶端 client進行測試，我們輸入中文"你好， 你是誰？"并通過Swarm類的run方法傳入主代理english_agent 和messages，通過response返回測試結果如下：

你好，我是一個智能助手，可以幫助你解決各種問題。你有什么需要幫助的嗎？

顯然，英語代理識別出中文輸入，并將請求轉交給中文代理執(zhí)行。

Swarm 核心概念

通過上面Swarm 示例代碼，可以理解 Swarm 的關鍵概念及功能實現(xiàn)：

1. client.run() 方法

client.run() 是 Swarm 中的核心方法，類似于 OpenAI 的 chat.completions.create() 方法。它接受消息輸入并返回消息輸出，同時在多輪調(diào)用之間不保存狀態(tài)。除了處理消息對話，還支持以下功能：

• 執(zhí)行 Agent 的函數(shù)調(diào)用并追加結果。
• 在任務完成后轉交給其他 Agent（handoffs）。
• 動態(tài)更新上下文變量（context variables）。
• 在必要時支持多輪對話再返回結果。

例如，在代碼中，我們通過 client.run() 將初始用戶消息傳遞給英語代理（English Agent）。當系統(tǒng)檢測到輸入的語言不符合代理的要求時，會調(diào)用代理函數(shù)，切換到適合語言的代理，如 Spanish Agent 或 Chinese Agent。

2. Agents（代理）

Agent 是任務執(zhí)行的基本單元。它可以被看作一個封裝了特定指令（instructions）和功能（functions）的“智能體”。Agent 的核心字段包括：

• name：代理的名稱，用于標識。
• instructions：代理的指令，決定代理的行為方式。
• functions：代理可以調(diào)用的一組 Python 函數(shù)，用于執(zhí)行特定任務。
• handoff：代理可以通過函數(shù)切換到其他代理。

代碼中，我們創(chuàng)建了三個代理：English Agent、Spanish Agent 和 Chinese Agent，它們分別按照語言設置了特定的指令。代理通過 instructions 告知系統(tǒng)其行為規(guī)范，例如僅使用某種語言交流。

3. Functions（函數(shù)）

Swarm 的代理支持直接調(diào)用 Python 函數(shù)執(zhí)行任務。函數(shù)通常返回一個字符串（str），也可以通過返回一個代理（Agent）實現(xiàn)代理之間的切換，或通過修改上下文變量（context_variables）來動態(tài)改變對話狀態(tài)。

在示例中，transfer_to_spanish_agent 和 transfer_to_chinese_agent 是兩個函數(shù)，它們的作用是根據(jù)用戶的語言輸入，將當前任務切換到對應的語言代理。

如果一個代理調(diào)用了多個函數(shù)，Swarm 會按照順序依次執(zhí)行它們。如果函數(shù)出現(xiàn)錯誤（如參數(shù)缺失或類型錯誤），系統(tǒng)會生成一個錯誤響應，并嘗試從錯誤中恢復。

4. Handoffs and Updating Context Variables（代理切換和上下文變量更新）

代理轉交（Handoff）在代理無法完成任務時，可以將任務交接給其他代理。例如，English Agent 調(diào)用 transfer_to_spanish_agent 函數(shù)后，返回了 Spanish Agent，實現(xiàn)了從英語代理到西班牙語代理的切換。

此外，Swarm 還支持動態(tài)更新上下文變量。通過返回一個 Result 對象，函數(shù)可以同時更新返回值、切換代理和修改上下文變量。這種能力確保了復雜任務流程中每個步驟的信息傳遞和狀態(tài)保持一致。

5. Function Schemas（函數(shù)模式）

Swarm 能夠自動將函數(shù)轉換為 JSON Schema，以便代理理解函數(shù)的功能和參數(shù)需求：

• 函數(shù)的文檔字符串（docstring）會轉化為函數(shù)的描述信息。
• 參數(shù)的類型提示（type hints）會映射到 JSON Schema 的字段類型。
• 如果函數(shù)定義了必需參數(shù)，則會自動標記為 required。

Swarm 是如何運作的 ？

Swarm 的運作核心在于通過多個智能體（Agent）的協(xié)同工作，實現(xiàn)復雜任務的高效處理。為了讓大家對這個過程有深入的了解，接下來，我們會從Routine 概念開始，逐步演化為具備智能能力的 Agent，并最終依靠 Handoff 功能實現(xiàn)動態(tài)任務分配。

Routine 是 Swarm 的基礎單元，由預定義的指令和工具/函數(shù)組成，負責描述任務的邏輯流程及完成任務所需的工具。而 Agent 是一種強化版的 Routine，通過結合大語言模型（LLM），賦予其智能理解和決策能力，使其不僅能夠執(zhí)行預定義任務，還能靈活應對復雜的用戶需求。在此基礎上，Swarm 的 Handoff 功能讓不同的 Agent 能夠根據(jù)請求的內(nèi)容動態(tài)協(xié)作，將用戶的任務無縫轉交給更適合的 Agent 處理，并完整保留對話上下文，避免用戶重復輸入。

什么是 Routine？

簡單來說，Routine 是一組預定義的指令與相應工具的組合，旨在完成特定任務。它不僅是一個執(zhí)行步驟的計劃，還包含完成任務所需的資源和工具。我們可以簡單理解Routine 就是Agent（代理）的雛形，它描述了代理需要完成的任務，同時還賦予它對應的工具/函數(shù)，只是Agent 本身還具備大模型的能力還可以進行“思考”。

如下圖所示，Routine可以拆解為兩部分：

• 指令（Instructions）：以自然語言或系統(tǒng)提示的形式描述的任務執(zhí)行步驟。
• 工具/函數(shù)（Tools/function）：完成這些步驟所需的工具或函數(shù)。需要說明的是，無論是調(diào)用工具或者函數(shù)都會通過函數(shù)調(diào)用的方式，也就是function call的方式完成。

說白了，Routine就是系統(tǒng)提示：描述任務的邏輯流程（比如詢問問題、搜索信息或處理用戶請求），加上可調(diào)用的工具/函數(shù)（用來輔助完成這些任務）。

來個具體的例子，我們?yōu)榭蛻舴沾矶x了一個例程，指示其對用戶問題進行分類，然后建議修復或提供退款。同時定義函數(shù)execute_refund和look_up_item 作為外部工具協(xié)助完成退貨以及查找訂單的工作。下面就是客戶服務例程的代碼：指令和工具/函數(shù)。

system_message = (
    "你是一名 ACME Inc. 的客戶服務代理，以下是你的工作流程：\n"
    "1. 首先，向用戶詢問更多信息以理解他們的問題（如果問題未明確）。\n"
    "2. 提出一個解決方案。\n"
    "3. 如果用戶不滿意，提供退款。\n"
    "4. 如果接受退款，查找物品 ID 并執(zhí)行退款操作。"
)
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"

如何執(zhí)行 Routine？

為了執(zhí)行 Routine，需要實現(xiàn)從用戶交互到模型調(diào)用、工具執(zhí)行的完整閉環(huán)。以下是幾個關鍵部分：

1.函數(shù)定義（Function Definition）

函數(shù)是 Routine 的核心工具。它們可以是具體的業(yè)務邏輯實現(xiàn)，例如處理退款或查詢信息。這也就是前面描述的execute_refund和look_up_item 兩個函數(shù)的定義，代碼在上面已經(jīng)展示過了， 這里就不贅述了。需要說明的是，look_up_item 執(zhí)行訂單查詢，execute_refund 負責退款操作。

2.函數(shù)接口（Function Schema）

當語言模型接到用戶請求的時候，需要調(diào)用函數(shù)返回對應的結果，調(diào)用函數(shù)時需要使用Function Schema。它包括：函數(shù)的名稱、參數(shù)及其類型等信息。為此需要將 Python 函數(shù)定義轉化為標準化的 Schema（模式）。為了實現(xiàn)從函數(shù)到函數(shù)接口的轉換需要實現(xiàn)如下代碼：

import inspect
def function_to_schema(func) -> dict:
    """將函數(shù)定義轉化為 Schema 格式。"""
    type_map = {
        str: "string",
        int: "integer",
        float: "number",
        bool: "boolean",
        list: "array",
        dict: "object",
        type(None): "null",
    }

    # 獲取函數(shù)簽名
    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,
            },
        },
    }

這個代碼主要通過輸入函數(shù)句柄，提取函數(shù)名稱、描述、輸入?yún)?shù)等信息，從而生成函數(shù)接口（Function schema）。我們可以用execute_refund 函數(shù)測試其效果。代碼如下：

# 示例輸出
schema = function_to_schema(execute_refund)
print(schema)
結果：
{
  "type": "function",
  "function": {
    "name": "execute_refund",
    "description": "執(zhí)行退款操作。",
    "parameters": {
      "type": "object",
      "properties": {
        "item_id": { "type": "string" },
        "reason": { "type": "string" }
      },
      "required": ["item_id"]
    }
  }
}

3. 函數(shù)調(diào)用（Function Call）

完成函數(shù)定義和調(diào)用函數(shù)接口之后，接下來就是利用函數(shù)接口去調(diào)用對應的函數(shù)。需要將 Schema 注冊為工具，供模型生成調(diào)用指令。然后，根據(jù)調(diào)用結果執(zhí)行對應函數(shù)，最后將結果返回給模型。執(zhí)行代碼如下：

# 定義可用的工具函數(shù)列表
tools = [execute_refund, look_up_item]

def run_full_turn(system_message, tools, messages):
    # 記錄初始消息數(shù)量
    num_init_messages = len(messages)
    # 復制消息列表以避免修改原始數(shù)據(jù)
    messages = messages.copy()

    while True:
        # 將Python函數(shù)轉換為OpenAI工具模式
        tool_schemas = [function_to_schema(tool) for tool in tools]
        # 創(chuàng)建工具名稱到函數(shù)的映射字典
        tools_map = {tool.__name__: tool for tool in tools}

        # 調(diào)用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)

        # 如果沒有工具調(diào)用請求則結束循環(huán)
        if not message.tool_calls:
            break

        # 處理工具調(diào)用
        for tool_call in message.tool_calls:
            # 執(zhí)行工具調(diào)用并獲取結果
            result = execute_tool_call(tool_call, tools_map)

            # 將工具調(diào)用結果添加到消息歷史
            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):
    # 獲取要調(diào)用的函數(shù)名
    name = tool_call.function.name
    # 解析函數(shù)參數(shù)
    args = json.loads(tool_call.function.arguments)

    # 打印調(diào)用信息
    print(f"Assistant: {name}({args})")

    # 執(zhí)行函數(shù)調(diào)用并返回結果
    return tools_map[name](**args)

# 主循環(huán)
messages = []
while True:
    # 獲取用戶輸入
    user = input("User: ")
    # 添加用戶消息到歷史
    messages.append({"role": "user", "content": user})

    # 執(zhí)行一輪對話并獲取新消息
    new_messages = run_full_turn(system_message, tools, messages)
    # 將新消息添加到歷史中
    messages.extend(new_messages)

在這段代碼中，通過大模型和Routine構建了一個交互系統(tǒng)，用來讓客服代理與用戶進行對話。我們截取部分重要段落給大家進行拆解如下：

首先，定義一組工具函數(shù)（如 execute_refund 和 look_up_item），這些工具提供了解決具體業(yè)務需求的能力。它們被集中存儲在一個列表中，隨后通過工具模式（Schema）轉化為模型可以識別和調(diào)用的接口描述，從而成為語言模型的可用擴展。

tools = [execute_refund, look_up_item]
tool_schemas = [function_to_schema(tool) for tool in tools]

function_to_schema 方法將 Python 函數(shù)轉換為模型可理解的工具描述，方便系統(tǒng)動態(tài)調(diào)用這些函數(shù)。

對話流程的核心由 run_full_turn 函數(shù)實現(xiàn)，它負責完成一輪從用戶輸入到工具調(diào)用再到生成助手回復的整個邏輯。其實現(xiàn)的關鍵在于利用 OpenAI 模型生成回復并解析是否需要工具調(diào)用。每次交互開始時，系統(tǒng)會加載當前的消息歷史以及可用工具的模式列表，然后通過調(diào)用模型接口獲取助手的回復。如果模型未請求調(diào)用工具，則直接返回助手的回復；否則系統(tǒng)將根據(jù)工具調(diào)用的描述執(zhí)行相應的函數(shù)。

response = client.chat.completions.create(
    model="gpt-4o-mini",
    messages=[{"role": "system", "content": system_message}] + messages,
    tools=tool_schemas or None,
)

工具調(diào)用的具體執(zhí)行邏輯由 execute_tool_call 函數(shù)實現(xiàn)。每當模型發(fā)出工具調(diào)用請求時，該函數(shù)會根據(jù)請求中描述的工具名稱和參數(shù)，查找與之對應的 Python 函數(shù)并執(zhí)行操作。例如，若助手要求退款操作，則函數(shù)會解析出對應的函數(shù)名稱和參數(shù)，然后調(diào)用 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)

最后，通過一個主循環(huán)實現(xiàn)用戶與助手的持續(xù)交互。用戶的輸入被實時加入消息歷史，隨后調(diào)用 run_full_turn 生成助手回復，并在必要時調(diào)用外部工具函數(shù)。所有新增的消息和工具調(diào)用結果都會被動態(tài)更新到對話歷史中，保證了上下文的連貫性。

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形態(tài)。這里我們將智能體（Agent）理解為一種強化版的 Routine，即 Routine 加上了語言模型（LLM）的智能能力。Routine 負責定義工具和執(zhí)行步驟，而 Agent 則通過結合 LLM 具備了智能理解和處理能力。具體來說，Agent 包括以下幾個核心要素：

• 名稱（Name）：用于標識 Agent 的職責。
• 模型（Model）：用來理解用戶請求，處理工具函數(shù)調(diào)用，并給予用戶反饋。
• 指令（Instructions）：定義 Agent 如何執(zhí)行其任務。
• 工具（Tools）：該 Agent 可以調(diào)用的一組函數(shù)。

我們可以通過如下代碼來定義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],
)

上面代碼定義了退貨和銷售代理，分別針對兩個代理都定義了工具/函數(shù)，協(xié)助他們完成工作。

Handoff轉交功能

好！目前，我們已經(jīng)理解了Routine 并且順利從Routine的概念過渡到了Agent，還記得我們在“什么是Swarm”章節(jié)中介紹Swarm結構的核心就是Agent和Handoff， 接著我們就來介紹Handoff 功能。前面的內(nèi)容中也提到了， 在當前代理無法處理對應請求的時候，該代理會將請求轉交給其他代理處理。

于是，我們修改執(zhí)行代碼如下：

def run_full_turn(agent, messages):
    # 初始化當前智能體為傳入的 agent
    current_agent = agent
    # 記錄初始消息數(shù)量，用于之后返回新增消息
    num_init_messages = len(messages)
    # 復制消息列表，防止對原始消息數(shù)據(jù)造成影響
    messages = messages.copy()

    while True:
        # 將當前智能體的工具列表轉化為工具模式（schemas），用于 API 調(diào)用
        tool_schemas = [function_to_schema(tool) for tool in current_agent.tools]
        # 構造工具名稱到工具函數(shù)的映射
        tools = {tool.__name__: tool for tool in current_agent.tools}

        # === 1. 調(diào)用 OpenAI 接口生成回復 ===
        response = client.chat.completions.create(
            model=agent.model,  # 當前智能體使用的模型
            messages=[{"role": "system", "content": current_agent.instructions}]
            + messages,  # 包括系統(tǒng)消息和歷史對話內(nèi)容
            tools=tool_schemas or None,  # 提供工具的模式定義
        )
        # 獲取生成的消息
        message = response.choices[0].message
        # 將回復消息添加到歷史記錄中
        messages.append(message)

        # 如果消息包含文本內(nèi)容，則打印當前智能體的回復
        if message.content:
            print(f"{current_agent.name}:", message.content)

        # 如果沒有工具調(diào)用請求，則退出循環(huán)
        if not message.tool_calls:
            break

        # === 2. 處理工具調(diào)用 ===
        for tool_call in message.tool_calls:
            # 執(zhí)行工具調(diào)用，并返回結果
            result = execute_tool_call(tool_call, tools, current_agent.name)

            # 如果工具調(diào)用結果是一個新的智能體對象，則進行切換
            if type(result) is Agent:
                current_agent = result  # 更新當前智能體為新的智能體
                result = (
                    f"Transfered to {current_agent.name}. Adopt persona immediately."
                )  # 生成切換通知

            # 將工具調(diào)用的結果作為消息添加到歷史記錄中
            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
    # 解析工具調(diào)用的參數(shù)
    args = json.loads(tool_call.function.arguments)

    # 打印工具調(diào)用的詳細信息
    print(f"{agent_name}:", f"{name}({args})")

    # 調(diào)用對應的工具函數(shù)并返回其結果
    return tools[name](**args)

我們把目光放到與Handoff 相關的細節(jié)上，工具調(diào)用返回智能體對象。在處理工具調(diào)用時，execute_tool_call 的返回結果是 Agent 對象，說明該工具無法完成用戶的請求，通過返回Agent的方式讓請求切換到其他Agent中去。這里需要注意的是，我們將Handoff的動作也封裝成了函數(shù)，該函數(shù)會直接返回要切換到的Agent對象，他的具體應用就在此處。

在run_full_turn函數(shù)中通過 type(result) 判斷，如果返回的是 Agent 類型，則說明需要進行智能體切換。

# 執(zhí)行工具調(diào)用，并返回結果
result = execute_tool_call(tool_call, tools, current_agent.name)
if type(result) is Agent:  # 如果工具調(diào)用返回一個新的 Agent
    current_agent = result  # 更新當前智能體
    result = (
        f"Transfered to {current_agent.name}. Adopt persona immediately."
    )  # 通知用戶智能體切換

除此之外，我們還需要動態(tài)更新 current_agent 的工具和說明，確保切換后的行為符合目標智能體的能力。為了讓對話記錄延續(xù)，需要通過消息列表的復制和擴展（messages.copy()），保留了所有上下文信息。

總結

Swarm 作為 OpenAI 的實驗性框架，旨在解決復雜場景下多代理協(xié)作問題。AI 代理具備自主感知、決策和行動能力，Swarm 則讓多個代理協(xié)同工作。其優(yōu)勢在于強大的任務分解和動態(tài)調(diào)度，適用于多種場景，如航空公司客戶服務等。在實戰(zhàn)示例中，通過代碼展示了多語言代理切換功能，體現(xiàn)了 Swarm 的核心方法、代理、函數(shù)等概念及功能。深入原理部分，Routine 是基礎單元，執(zhí)行 Routine 需實現(xiàn)函數(shù)定義、接口和調(diào)用等環(huán)節(jié)，Agent 是強化版 Routine，結合了大語言模型的智能能力，Handoff 功能可在代理無法處理請求時進行任務轉交，通過一系列操作確保對話上下文延續(xù)及智能體切換后的行為符合要求。

作者介紹

崔皓，51CTO社區(qū)編輯，資深架構師，擁有18年的軟件開發(fā)和架構經(jīng)驗，10年分布式架構經(jīng)驗。