OpenAI今天Open了一下：開源多智能體框架Swarm

機器之心報道

編輯：Panda

毫無疑問，多智能體肯定是 OpenAI 未來重要的研究方向之一，前些天 OpenAI 著名研究科學家 Noam Brown 還在 X 上爲 OpenAI 正在組建的一個新的多智能體研究團隊招募機器學習工程師。

就在幾個小時前，這個或許還沒有組建完成的新研究團隊就已經開源發佈了一項重量級研究成果：Swarm。這是一個實驗性質的多智能體編排框架，主打特徵是工效（ergonomic）與輕量（lightweight）。

Swarm 開源後引起了熱烈討論，有網友表示這能幫助簡化許多潛在的多智能體用例的工作流程。

我們先來看一個例子。首先安裝 Swarm，很簡單：

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

裝好這個框架之後，用起來也很方便。以下代碼定義了 2 個智能體，而用戶的指令是與智能體 B 交談：

from swarm import Swarm, Agent

client = Swarm()

def transfer_to_agent_b():

return agent_b

agent_a = Agent(

name="Agent A",

instructions="You are a helpful agent.",

functions=[transfer_to_agent_b],

agent_b = Agent(

name="Agent B",

instructions="Only speak in Haikus.",

response = client.run(

agent=agent_a,

messages=[{"role": "user", "content": "I want to talk to agent B."}],

print(response.messages[-1]["content"])

輸出消息：

Hope glimmers brightly,

New paths converge gracefully,

What can I assist?

下面我們就來簡單介紹一下這個開源項目。

首先，需要再次強調，Swarm 是一個實驗性質的多智能體框架，並不是爲生產目的開發的，因此團隊表示不會提供任何官方支持。

Swarm 概況

Swarm 關注的重點是讓智能體協作和執行變得輕量、高度可控且易於測試。

爲此，它使用了兩種原語抽象：智能體（agent）和交接（handoff）。其中，智能體包含指令和工具，並且在任何時間都可以選擇將對話交接給另一個智能體。

該團隊表示，這些原語很強大，「足以表達工具和智能體網絡之間的豐富動態，讓你可以針對真實世界問題構建可擴展的解決方案，同時避免陡峭的學習曲線。」

另外，該團隊指出，請注意 Swarm 智能體與 Assistants API 中的 Assistants 無關。之所以名字相似，只是爲了方便。Swarm 完全由 Chat Completions API 提供支持，因此在調用之間是無狀態的。

爲什麼要使用 Swarm？

在設計上，Swarm 是輕量級、可擴展且高度可定製的。它最適合處理存在大量獨立功能和指令的情況——這些功能和指令很難編碼成單個提示詞。

如果開發者想要尋求完全託管的線程以及內置的內存管理和檢索，那麼 Assistants API 就已經是很好的選擇了。但如果開發者想要完全的透明度，並且能夠細粒度地控制上下文、步驟和工具調用，那麼 Swarm 纔是最佳選擇。Swarm （幾乎）完全運行在客戶端，與 Chat Completions API 非常相似，不會在調用之間存儲狀態。

該團隊還展示了一個應用示例，包括天氣查詢智能體、用於在航空公司環境中處理不同客戶服務請求的多智能體設置、客服機器人、可以幫助銷售和退款的個人智能體等。具體示例請訪問 Swarm 代碼庫。

簡單的天氣查詢智能體示例，問題先經過篩選智能體處理，再轉交給天氣智能體解答

Swarm 的核心組件

Swarm 的核心組件包括client（客戶端）、Agent（智能體）、Function（函數）。

運行 Swarm 就是從實例化一個 client 開始的（其就是在內部實例化一個 OpenAI 客戶端）。

from swarm import Swarm

client = Swarm()

client.run()

Swarm 的 run() 函數類似於 Chat Completions API 中的 chat.completions.create() 函數——接收消息並返回消息，並且在調用之間不保存任何狀態。但重點在於，它還處理 Agent 函數執行、交接、上下文變量引用，並且可以在返回給用戶之前進行多輪執行。

究其核心，Swarm 的 client.run() 是實現以下循環：

參數

client.run() 的參數包括：

client.run() 完成後（可能進行過多次智能體和工具調用），會返回一個響應，其中包含所有相關的已更新狀態。具體來說，即包含新消息、最後調用的智能體、最新的上下文變量。你可以將這些值（加上新的用戶消息）傳遞給 client.run() 的下一次執行，以繼續上次的交互——就像是 chat.completions.create()

響應字段

Agent

Agent（智能體）就是將一組指令與一組函數封裝在一起（再加上一些額外的設置），並且其有能力將執行過程交接給另一個 Agent。

Agent 字段

指令（instructions）

Agent instructions 會直接轉換成對話的系統提示詞（作爲第一條消息）。只有當前活動的 Agent 的指令會被使用（當發生智能體交接時，系統提示詞會變化，但聊天曆史不會）。

agent = Agent(

instructions="You are a helpful agent."

instructions 可以是常規字符串，也可以是返回字符串的函數。該函數可以選擇性地接收 context_variables 參數，該參數將由傳入 client.run() 的 context_variables 填充。

def instructions(context_variables):

user_name = context_variables["user_name"]

return f"Help the user, {user_name}, do whatever they want."

agent = Agent(

instructions=instructions

response = client.run(

agent=agent,

messages=[{"role":"user", "content": "Hi!"}],

context_variables={"user_name":"John"}

print(response.messages[-1]["content"])

輸出消息：

Hi John, how can I assist you today?

Function

def greet(context_variables, language):

user_name = context_variables["user_name"]

greeting = "Hola" if language.lower() == "spanish" else "Hello"

print(f"{greeting}, {user_name}!")

return "Done"

agent = Agent(

functions=[print_hello]

client.run(

agent=agent,

messages=[{"role": "user", "content": "Usa greet() por favor."}],

context_variables={"user_name": "John"}

輸出消息：

Hola, John!

如果某個 Agent 函數調用出錯（缺少函數、參數錯誤等），則會在聊天之中附加一條報錯響應，以便 Agent 恢復正常。

如果 Ageny 調用多個函數，則按順序執行它們。

交接和更新上下文變量

通過在返回的函數中包含一個 Agent，可將執行過程交接給這個 Agent。

sales_agent = Agent(name="Sales Agent")

def transfer_to_sales():

return sales_agent

agent = Agent(functions=[transfer_to_sales])

response = client.run(agent, [{"role":"user", "content":"Transfer me to sales."}])

print(response.agent.name)

輸出消息：

Sales Agent

它還可以通過返回更完整的 Result 對象來更新 context_variables。如果你希望用單個函數返回一個值、更新智能體並更新上下文變量（或三者中的任何組合），它還可以包含一個 value 和一個 agent。

sales_agent = Agent(name="Sales Agent")

def talk_to_sales():

print("Hello, World!")

return Result(

value="Done",

agent=sales_agent,

context_variables={"department": "sales"}

agent = Agent(functions=[talk_to_sales])

response = client.run(

agent=agent,

messages=[{"role": "user", "content": "Transfer me to sales"}],

context_variables={"user_name": "John"}

print(response.agent.name)

print(response.context_variables)

輸出消息：

Sales Agent

{'department': 'sales', 'user_name': 'John'}

注意：如果一個 Agent 調用了多個交接 Agent 的函數，則僅使用最後一個交接函數。

函數模式

Swarm 會自動將函數轉換爲 JSON 模式，然後將其傳遞給聊天補全工具。

def greet(name, age: int, location: str = "New York"):

"""Greets the user. Make sure to get their name and age before calling.

Args:

name: Name of the user.

age: Age of the user.

location: Best place on earth.

print(f"Hello {name}, glad you are {age} in {location}!")

"type": "function",

"function": {

"name": "greet",

"description": "Greets the user. Make sure to get their name and age before calling.\n\nArgs:\n name: Name of the user.\n age: Age of the user.\n location: Best place on earth.",

"parameters": {

"type": "object",

"properties": {

"name": {"type": "string"},

"age": {"type": "integer"},

"location": {"type": "string"}

},

"required": ["name", "age"]

流式處理

Swarm 也支持流式處理。

stream = client.run(agent, messages, stream=True)

for chunk in stream:

print(chunk)

使用了與 Chat Completions API streaming 一樣的事件，但添加了兩個事件類型：

核心貢獻者

Swarm 的核心貢獻者均就職於 OpenAI，他們分別是（右側爲對應的 GitHub 用戶名）：