作者 | 崔皓

審校 | 重樓

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

引言：揭開 Swarm 的神秘面紗

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

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

什么是 Swarm？

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

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

各位看到這里就需要?jiǎng)澲攸c(diǎn)了， Swarm的作用是構(gòu)建、組織并管理多個(gè)AI Agent，說白了，Swarm的作用就是協(xié)調(diào)多代理工作。

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

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

AI 代理的核心概念包括：

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

AI 代理的組成主要包括：

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

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

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

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

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

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

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

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

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

為什么需要 Swarm？

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

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

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

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

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

Swarm 的實(shí)戰(zhàn)

前面對(duì)Swarm 進(jìn)行簡(jiǎn)單介紹之后，我們來嘗試安裝Swarm 并通過實(shí)例體驗(yàn)一下它的“魅力”。通過如下指令安裝Swarm：

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

接著，通過一個(gè)簡(jiǎn)單代碼來看看它是如何讓Agent 工作的。

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

詳細(xì)代碼如下：

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

重點(diǎn)代碼解析如下：

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

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

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

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

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

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

3. 功能注冊(cè)

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

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

4. 測(cè)試代碼

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

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

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

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

Swarm 核心概念

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

1. client.run() 方法

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

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

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

2. Agents（代理）

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

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

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

3. Functions（函數(shù)）

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

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

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

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

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

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

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

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

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

Swarm 是如何運(yùn)作的 ？

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

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

什么是 Routine？

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

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

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

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

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

system_message = (
    "你是一名 ACME Inc. 的客戶服務(wù)代理，以下是你的工作流程：\n"
    "1. 首先，向用戶詢問更多信息以理解他們的問題（如果問題未明確）。\n"
    "2. 提出一個(gè)解決方案。\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，需要實(shí)現(xiàn)從用戶交互到模型調(diào)用、工具執(zhí)行的完整閉環(huán)。以下是幾個(gè)關(guān)鍵部分：

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

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

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

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

import inspect
def function_to_schema(func) -> dict:
    """將函數(shù)定義轉(zhuǎn)化為 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,
            },
        },
    }

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

# 示例輸出
schema = function_to_schema(execute_refund)
print(schema)
結(jié)果：
{
  "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)用對(duì)應(yīng)的函數(shù)。需要將 Schema 注冊(cè)為工具，供模型生成調(diào)用指令。然后，根據(jù)調(diào)用結(jié)果執(zhí)行對(duì)應(yīng)函數(shù)，最后將結(jié)果返回給模型。執(zhí)行代碼如下：

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

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

    while True:
        # 將Python函數(shù)轉(zhuǎn)換為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獲取回復(fù)
        response = client.chat.completions.create(
            model="gpt-4o-mini",
            messages=[{"role": "system", "content": system_message}] + messages,
            tools=tool_schemas or None,
        )
        # 獲取AI助手的回復(fù)
        message = response.choices[0].message
        # 將助手回復(fù)添加到消息歷史
        messages.append(message)

        # 如果有文本回復(fù)則打印
        if message.content:
            print("Assistant:", message.content)

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

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

            # 將工具調(diào)用結(jié)果添加到消息歷史
            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)用并返回結(jié)果
    return tools_map[name](**args)

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

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

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

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

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

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

對(duì)話流程的核心由 run_full_turn 函數(shù)實(shí)現(xiàn)，它負(fù)責(zé)完成一輪從用戶輸入到工具調(diào)用再到生成助手回復(fù)的整個(gè)邏輯。其實(shí)現(xiàn)的關(guān)鍵在于利用 OpenAI 模型生成回復(fù)并解析是否需要工具調(diào)用。每次交互開始時(shí)，系統(tǒng)會(huì)加載當(dāng)前的消息歷史以及可用工具的模式列表，然后通過調(diào)用模型接口獲取助手的回復(fù)。如果模型未請(qǐng)求調(diào)用工具，則直接返回助手的回復(fù)；否則系統(tǒng)將根據(jù)工具調(diào)用的描述執(zhí)行相應(yīng)的函數(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ù)實(shí)現(xiàn)。每當(dāng)模型發(fā)出工具調(diào)用請(qǐng)求時(shí)，該函數(shù)會(huì)根據(jù)請(qǐng)求中描述的工具名稱和參數(shù)，查找與之對(duì)應(yīng)的 Python 函數(shù)并執(zhí)行操作。例如，若助手要求退款操作，則函數(shù)會(huì)解析出對(duì)應(yīng)的函數(shù)名稱和參數(shù)，然后調(diào)用 execute_refund 完成任務(wù)，返回結(jié)果后更新對(duì)話歷史。

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)

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

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，它可以理解為指令和工具的集合，為了讓它獨(dú)立工作還需要加入大語言模型（LLM），這樣才能過渡到Swarm中的Agent形態(tài)。這里我們將智能體（Agent）理解為一種強(qiáng)化版的 Routine，即 Routine 加上了語言模型（LLM）的智能能力。Routine 負(fù)責(zé)定義工具和執(zhí)行步驟，而 Agent 則通過結(jié)合 LLM 具備了智能理解和處理能力。具體來說，Agent 包括以下幾個(gè)核心要素：

• 名稱（Name）：用于標(biāo)識(shí) Agent 的職責(zé)。
• 模型（Model）：用來理解用戶請(qǐng)求，處理工具函數(shù)調(diào)用，并給予用戶反饋。
• 指令（Instructions）：定義 Agent 如何執(zhí)行其任務(wù)。
• 工具（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 = []

順著這個(gè)思路，我們可以定義更多的代理如下：

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],
)

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

Handoff轉(zhuǎn)交功能

好！目前，我們已經(jīng)理解了Routine 并且順利從Routine的概念過渡到了Agent，還記得我們?cè)凇笆裁词荢warm”章節(jié)中介紹Swarm結(jié)構(gòu)的核心就是Agent和Handoff， 接著我們就來介紹Handoff 功能。前面的內(nèi)容中也提到了， 在當(dāng)前代理無法處理對(duì)應(yīng)請(qǐng)求的時(shí)候，該代理會(huì)將請(qǐng)求轉(zhuǎn)交給其他代理處理。

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

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

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

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

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

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

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

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

            # 將工具調(diào)用的結(jié)果作為消息添加到歷史記錄中
            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)用的詳細(xì)信息
    print(f"{agent_name}:", f"{name}({args})")

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

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

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

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

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

總結(jié)

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

作者介紹

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