一個教育框架,探索符合人體工程學,輕巧的多機構編排。
警告
Swarm目前是一個實驗樣本框架,旨在探索多代理系統的人體工程學接口。它不打算用於生產,因此沒有官方支持。 (這也意味著我們不會審查PRS或問題!)
Swarm的主要目標是展示在編排代理商中探索的交接和例行圖案:Handoffs&Requines Cookbook。它並不是作為獨立圖書館,主要是出於教育目的。
需要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專注於使代理協調和執行輕巧,高度可控且易於測試。
它通過兩個原始的抽象來實現這一目標: Agent
和交接。 Agent
包含instructions
和tools
,並且可以隨時選擇將對話交給另一個Agent
。
這些原語的功能足夠強大,可以在工具和代理網絡之間表達丰富的動態,從而使您可以構建可擴展的現實解決方案,同時避免使用陡峭的學習曲線。
筆記
群體代理與助手API中的助手無關。為了方便起見,它們的命名類似,但否則完全無關。 Swarm完全由聊天完成API提供動力,因此在呼叫之間無狀態。
Swarm探索了輕巧,可擴展和高度可定制設計的模式。類似於群的方法最適合處理大量獨立功能和說明,這些功能和說明很難將其編碼成一個提示。
助手API是尋求完全託管線程並內置內存管理和檢索的開發人員的絕佳選擇。但是,Swarm是一種教育資源,對於開發人員來說,好奇地了解多代理編排。 Swarm(幾乎)完全在客戶端上運行,就像聊天完成API一樣,不會在呼叫之間存儲狀態。
查看/examples
以獲取靈感!了解有關每個人中每個人的更多信息。
basic
:基本原理的簡單示例,例如設置,功能呼叫,交接和上下文變量triage_agent
:設置基本分類步驟的簡單示例,將其交給正確的代理weather_agent
:功能調用的簡單示例airline
:用於在航空公司環境下處理不同客戶服務請求的多代理設置。support_bot
:一個包括用戶界面代理和使用多個工具的幫助中心代理的客戶服務機器人personal_shopper
:可以幫助銷售和退款訂單的個人購物代理首先實例化群客戶端(僅在內部實例化OpenAI
客戶端)開始。
from swarm import Swarm
client = Swarm ()
client.run()
swarm的run()
函數類似於chat.completions.create()
在聊天完成API中的功能 - 它接收messages
並返回messages
,並且在呼叫之間沒有保存狀態。但是,重要的是,它還處理代理功能執行,交接,上下文變量引用,並且在返回用戶之前可以進行多個轉彎。
Swarm的client.run()
以其核心實現了以下循環:
爭論 | 類型 | 描述 | 預設 |
---|---|---|---|
代理人 | Agent | 要調用的(初始)代理。 | (必需的) |
消息 | List | 消息對象的列表,與聊天完整messages 相同 | (必需的) |
context_variables | dict | 其他上下文變量的字典,可用於功能和代理說明 | {} |
max_turns | int | 允許的最大對話回合數 | float("inf") |
model_override | str | 可選字符串以覆蓋代理使用的模型 | None |
execute_tools | bool | 如果為False ,請中斷執行並立即返回tool_calls 消息,當代理試圖調用函數時 | True |
溪流 | bool | 如果是True ,請啟用流響應 | False |
偵錯 | bool | 如果為True ,請啟用調試記錄 | False |
client.run()
完成後(在可能對代理和工具進行多次調用之後),它將返回包含所有相關更新狀態的Response
。具體來說, messages
,要調用的最後一個Agent
以及最新的上下文context_variables
。您可以將這些值(加上新用戶消息)傳遞到client.run()
的下一個執行中,以繼續其關閉的交互 - 就像chat.completions.create()
一樣。 ( run_demo_loop
函數實現/swarm/repl/repl.py
中的完整執行循環的示例。)
Response
字段場地 | 類型 | 描述 |
---|---|---|
消息 | List | 對話期間生成的消息對象的列表。與聊天完成messages 非常相似,但是帶有sender 字段,指示該消息源自哪個Agent 。 |
代理人 | Agent | 處理消息的最後一個代理。 |
context_variables | dict | 與輸入變量相同,加上任何更改。 |
Agent
只需將一組具有一組functions
的instructions
封裝(以及下面的一些其他設置),並有能力將執行移交給另一個Agent
。
雖然誘人將Agent
化為“做X的人”,但它也可以用來表示由一組instructions
和functions
定義的非常具體的工作流程或步驟(例如,一組步驟,一個複雜的檢索,單一的步驟數據轉換等)。這允許Agent
S組成“代理”,“工作流”和“任務”的網絡,這些網絡均以相同的原始形式表示。
Agent
字段場地 | 類型 | 描述 | 預設 |
---|---|---|---|
姓名 | str | 代理的名稱。 | "Agent" |
模型 | str | 代理使用的模型。 | "gpt-4o" |
指示 | str 或func() -> str | 代理的說明可以是字符串或可可返回字符串的說明。 | "You are a helpful agent." |
功能 | List | 代理可以調用的功能列表。 | [] |
tool_choice | str | 代理商的工具選擇(如果有)。 | None |
Agent
instructions
將直接轉換為對話的system
提示(作為第一個消息)。只有在任何給定時間出現活動Agent
的instructions
(例如,如果有Agent
交接, system
提示將會更改,但聊天歷史記錄不會。)
agent = Agent (
instructions = "You are a helpful agent."
)
instructions
可以是常規str
,也可以是返回str
函數。該函數可以選擇接收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?
Agent
S可以直接調用Python函數。str
(將嘗試以str
形式施放值)。Agent
,則執行將轉移到該Agent
。context_variables
參數,則將通過傳遞到client.run()
上下文context_variables
填充它。 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 ]
)
client . run (
agent = agent ,
messages = [{ "role" : "user" , "content" : "Usa greet() por favor." }],
context_variables = { "user_name" : "John" }
)
Hola, John!
Agent
函數調用有錯誤(丟失函數,錯誤的參數,錯誤),則將將錯誤響應附加到聊天中,以便Agent
可以優雅地恢復。Agent
調用多個功能,則將按該順序執行它們。Agent
可以通過將其返回function
中的另一個代理來移交給另一個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架構,該功能將傳遞到聊天完成tools
中。
description
。required
。type
(默認為string
)。 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.nnArgs: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" ]
}
}
}
stream = client . run ( agent , messages , stream = True )
for chunk in stream :
print ( chunk )
使用與聊天完成API流相同的事件。請參見/swarm/repl/repl.py
中的process_and_print_streaming_response
作為示例。
添加了兩種新的事件類型:
{"delim":"start"}
和{"delim":"end"}
,以每次Agent
處理單個消息(響應或函數調用)發出信號。這有助於識別Agent
之間的開關。{"response": Response}
將在流的末尾返回一個帶有匯總(完整)響應Response
對象,以方便起見。評估對任何項目都至關重要,我們鼓勵開發人員攜帶自己的評估套件來測試其群體的表現。作為參考,我們提供了一些示例,涉及如何在airline
, weather_agent
和triage_agent
Quickstart示例中評估群。有關更多詳細信息,請參見READMES。
使用run_demo_loop
測試您的群!這將在您的命令行上運行一個重複。支持流。
from swarm . repl import run_demo_loop
...
run_demo_loop ( agent , stream = True )