OpenAI Swarm - 多智能體編排框架詳細介紹
項目概述
OpenAI Swarm 是一個教育性框架，用於探索符合人體工學的輕量級多智能體編排。該項目由 OpenAI 解決方案團隊管理，旨在為開發者提供一個簡單、靈活且可控的多智能體系統構建工具。
重要更新: Swarm 現已被 OpenAI Agents SDK 替代，後者是 Swarm 的生產就緒版本。OpenAI 團隊建議所有生產用例遷移到 Agents SDK。
核心特性
1. 設計理念
- 輕量級: 專注於簡單性和易用性
- 高度可控: 提供精確的智能體控制機制
- 易於測試: 支援簡單的測試和除錯流程
- 教育導向: 作為學習多智能體編排的教育資源
2. 核心抽象概念
Swarm 透過兩個核心抽象實現智能體協調：
Agent（智能體）
- 包含指令 (instructions) 和工具 (tools)
- 可以在任何時點將對話交接給另一個智能體
- 支援動態指令和上下文變數
Handoffs（交接）
- 智能體之間的無縫轉換機制
- 支援複雜的智能體網路和工作流程
- 保持對話連續性
安裝和基本使用
系統要求
- Python 3.10+
安裝方式
pip install git+ssh://[email protected]/openai/swarm.git
# 或者
pip install git+https://github.com/openai/swarm.git
基本示例
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 的 client.run() 函數實現以下循環：
- 從當前智能體獲取完成響應
- 執行工具調用並添加結果
- 必要時切換智能體
- 更新上下文變數
- 如果沒有新的函數調用，返回結果
參數配置
| 參數 | 類型 | 描述 | 預設值 |
|---|---|---|---|
| agent | Agent | 要調用的（初始）智能體 | (必需) |
| messages | List | 消息物件列表 | (必需) |
| context_variables | dict | 上下文變數 | {} |
| max_turns | int | 最大輪次 | float("inf") |
| model_override | str | 模型覆蓋 | None |
| execute_tools | bool | 是否執行工具 | True |
| stream | bool | 是否啟用流式響應 | False |
| debug | bool | 是否啟用除錯日誌 | False |
智能體配置
Agent 類字段
| 字段 | 類型 | 描述 | 預設值 |
|---|---|---|---|
| name | str | 智能體名稱 | "Agent" |
| model | str | 使用的模型 | "gpt-4o" |
| instructions | str或函數 | 智能體指令 | "You are a helpful agent." |
| functions | List | 可調用函數列表 | [] |
| tool_choice | str | 工具選擇策略 | None |
動態指令示例
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"}
)
函數調用機制
基本函數調用
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=[greet])
智能體交接
sales_agent = Agent(name="Sales Agent")
def transfer_to_sales():
return sales_agent
agent = Agent(functions=[transfer_to_sales])
複雜結果返回
def talk_to_sales():
print("Hello, World!")
return Result(
value="Done",
agent=sales_agent,
context_variables={"department": "sales"}
)
示例項目
Swarm 提供了多個示例項目，展示不同的應用場景：
1. basic - 基礎示例
演示設定、函數調用、交接和上下文變數的基本用法
2. triage_agent - 分流智能體
簡單的分流設定，將請求轉發給合適的智能體
3. weather_agent - 天氣智能體
展示函數調用的簡單示例
4. airline - 航空客服
多智能體設定，處理航空公司不同類型的客戶服務請求
5. support_bot - 支援機器人
包括用戶界面智能體和帶有多種工具的幫助中心智能體
6. personal_shopper - 個人購物助手
幫助進行銷售和退款訂單處理的個人購物智能體
流式處理
Swarm 支援流式響應，使用與 Chat Completions API 相同的事件：
stream = client.run(agent, messages, stream=True)
for chunk in stream:
print(chunk)
新增事件類型：
{"delim":"start"}和{"delim":"end"}- 標記智能體處理單個消息的開始和結束{"response": Response}- 在流結束時返回完整的 Response 物件
測試和評估
開發測試
使用 run_demo_loop 進行命令行測試：
from swarm.repl import run_demo_loop
run_demo_loop(agent, stream=True)
評估建議
- 項目鼓勵開發者自帶評估套件
- 在 airline、weather_agent 和 triage_agent 示例中提供了評估參考
- 支援自定義性能測試指標
與其他 OpenAI 產品的區別
與 Assistants API 的對比
- Assistants API: 提供完全託管的線程和內置記憶體管理
- Swarm: 完全在客戶端運行，無狀態設計，更適合學習和實驗
技術特點
- 基於 Chat Completions API 構建
- 調用之間不保存狀態
- 輕量級客戶端實現
- 專注於教育和學習目的
適用場景
Swarm 特別適合以下情況：
- 需要處理大量獨立功能和指令的場景
- 難以編碼到單個提示中的複雜任務
- 學習多智能體編排概念
- 快速原型開發和實驗
項目狀態
注意: 雖然 Swarm 已被 OpenAI Agents SDK 替代，但它仍然是一個優秀的教育資源，幫助開發者理解多智能體系統的基本概念和實現方式。對於生產環境，建議遷移到官方的 Agents SDK。
總結
OpenAI Swarm 為多智能體系統的學習和開發提供了一個簡潔而強大的框架。透過智能體和交接這兩個核心概念，開發者可以構建複雜的 AI 工作流程，同時保持程式碼的可讀性和可維護性。雖然已被新的 SDK 替代，但其設計理念和教育價值依然重要。