概覽
多智慧體功能讓模型啟動並協調多個子代理程式平行工作,再整合其成果,提供最終回應。對於需要處理複雜任務,且能受益於平行分工的應用程式,這項功能尤其有效,例如探索程式碼庫、撰寫文件及實作功能。
所有 GPT-5.6 模型皆提供多智慧體 Beta 功能。在應用程式中啟用多智慧體功能前,請先查看模型頁面。
何時使用多智慧體
任務通常可以拆分為彼此獨立的工作。單一智慧體必須依序完成這些工作,多個智慧體則能平行處理。多智慧體功能讓根智慧體將工作委派給多個子代理程式,同時執行。這可帶來多項好處:
- 平行執行。 彼此獨立的研究、分析或實作任務可同時進行,有助於加快執行速度。
- 專注的上下文。 每個子代理程式都會收到範圍明確的任務,並維護各自的上下文,減少無關工作之間的上下文干擾,提升表現。
- 由模型主導協調。 根智慧體可以建立子代理程式、向它們提供補充資訊、等待結果,並整合出最終答案,應用程式無須自行實作編排邏輯。
當任務可拆分為具體且彼此獨立的工作時,多智慧體編排最能發揮效用,例如:
- 探索大型程式碼庫的不同部分
- 比較多個提案、文件或假設
- 平行研究多個資料來源
- 實作獨立元件或撰寫獨立測試套件
- 平行調查可能導致故障的不同原因
- 同時探索解決問題的不同方法
請注意,增加子代理程式可能會增加 Token 用量。對於依賴單一循序推理流程、需要頻繁寫入共用可變狀態,或耗時主要集中於單一緩慢外部操作的任務,效益可能較小。
| 適合使用多智慧體的情況 | 較適合使用單一智慧體的情況 |
|---|---|
| 工作可拆分為彼此獨立且範圍明確的任務 | 每個步驟都直接依賴前一個步驟 |
| 獨立的上下文有助於保持專注 | 任務規模夠小,可在一次短時間執行內完成 |
| 平行探索可縮短實際經過的時間 | 智慧體會爭用同一個可變資源 |
| 比較各自獨立取得的發現有助於擴大涵蓋範圍 | 你需要固定且具確定性的執行圖 |
快速入門
Python 與 JavaScript 範例使用 Beta 版 Responses SDK。對於 HTTP
請求,請使用 client.beta.responses,並透過
betas 引數傳入 responses_multi_agent=v1。對於原始 HTTP 請求與 WebSocket 連線,請在請求或連線標頭中傳入
OpenAI-Beta: responses_multi_agent=v1。
多智慧體功能處於 Beta 階段期間,項目的結構描述可能會變更。
在 Responses API 請求中使用 multi_agent.enabled 啟用多智慧體功能。當 multi_agent.enabled 為 true 時,根智慧體便可建立樹狀的子代理程式結構。子代理程式共用請求所指定的模型與可用工具,而智慧體則透過建立、傳訊及等待等基本協作操作相互協調(請參閱多智慧體的運作方式)。根智慧體負責整合子代理程式的回應,並提供最終回應。
from openai import OpenAI
client = OpenAI()
def review_pull_request(diff: str) -> str:
response = client.beta.responses.create(
model="gpt-5.6-sol",
input=(
"Review the pull-request diff below with three agents: one for "
"correctness, one for security, and one for missing tests. "
"Reconcile duplicate or conflicting findings, then return a "
"prioritized review with file and line references.\n\n"
f"<diff>\n{diff}\n</diff>"
),
multi_agent={
"enabled": True,
"max_concurrent_subagents": 3,
},
betas=["responses_multi_agent=v1"],
)
return "".join(
part.text
for item in response.output
if (
item.type == "message"
and item.agent is not None
and item.agent.agent_name == "/root"
and item.phase == "final_answer"
)
for part in item.content
if part.type == "output_text"
)max_concurrent_subagents 設定整個智慧體樹中可同時處於活動狀態的子代理程式數量上限。計算範圍包括所有後代,也就是子層、孫層及更深層的子代理程式,但不包含根智慧體。
API 並未對此設定施加固定上限。預設值為 3,建議大多數工作負載使用此值。多智慧體執行也不限制樹的深度,或單次執行期間建立的子代理程式總數。
你可以新增開發者訊息,調整根模型應在何時建立子代理程式。這則開發者訊息會補充系統為根智慧體與子代理程式注入的指示。
開發者訊息範例如下:
- 「除非使用者明確要求使用子代理程式、委派工作或讓智慧體平行工作,否則不要建立子代理程式。」
- 「已啟用主動多智慧體委派。當平行工作能明顯提升速度或品質時,請使用子代理程式。」
多智慧體的運作方式
Responses API 為根智慧體與子代理程式的模型提供託管編排動作,以及使用這些動作的指示。根智慧體的名稱為 /root。建立的子代理程式使用階層式路徑,例如:
/root
├── /root/researcher
├── /root/reviewer
└── /root/reviewer/tester
多智慧體功能並未對子代理程式總數或樹的深度施加固定限制。大多數任務請使用 max_concurrent_subagents 的預設值 3。此設定限制整個樹中同時進行的子代理程式回合數,包含子層及更深層的後代。
啟用多智慧體模式時,Responses API 會提供六種託管協作動作。這些動作可能以 multi_agent_call 項目的形式出現。應用程式不應執行這些動作,也不應提交其輸出。
| 動作 | 用途 |
|---|---|
spawn_agent | 建立子代理程式並指派其初始任務。 |
send_message | 將訊息加入現有智慧體的佇列,但不啟動新回合。 |
followup_task | 將更多工作指派給現有的非根智慧體,並啟動或恢復其回合。 |
wait_agent | 等待發出呼叫的智慧體信箱收到更新。 |
interrupt_agent | 中斷另一個智慧體正在進行的回合,但不刪除其上下文。 |
list_agents | 傳回目前的智慧體樹、狀態,以及每個智慧體的 last_task_message。 |
開發者自訂工具呼叫的處理方式,與未啟用多智慧體功能時相同。樹中的任何智慧體都可能發出 function_call。應用程式必須執行該呼叫,並提交對應的 function_call_output。
請注意,樹中的所有智慧體都能使用 API 請求內模型呼叫所設定的工具。
在 Responses API 中使用多智慧體
HTTP 與 WebSocket 效能比較
HTTP 與 WebSocket 支援相同的多智慧體能力,但對於大量使用工具或長時間執行的工作流程,建議使用 WebSocket。其持續連線讓應用程式能在函式輸出可用時立即傳回,減少接續執行的額外負擔,並縮短智慧體的等待時間。
使用 HTTP 時,必須等到所有活動中的智慧體都已完成工作,或暫停以等待用戶端執行函式呼叫,回應才會完成。接著,應用程式會執行所有待處理的函式呼叫,並在新的 Responses API 請求中提交其輸出,讓暫停的智慧體繼續執行。
使用 WebSocket 時,應用程式可在每個函式輸出可用時,立即將其注入回應,無須等待進行中的回應完成。等待中的智慧體可立即恢復執行,其他智慧體則能繼續工作。當智慧體在不同時間完成工作或請求使用工具時,這可減少協調延遲,並避免額外的請求往返。
對於需要呼叫多個託管工具的工作流程,例如平行網頁搜尋,或僅需單次請求且函式呼叫不多的工作流程,HTTP 可能已足夠。對大多數多智慧體工作流程而言,WebSocket 通常能提供較低的延遲與更好的端對端效能。
HTTP 函式呼叫的執行流程

WebSocket 函式呼叫的執行流程

HTTP
這些範例需要使用提供 Beta 版 Responses API 的 Beta 版 SDK。若要使用 HTTP 串流,請呼叫 client.beta.responses.create,並透過 betas 引數傳入 responses_multi_agent=v1,以啟用 Beta 版型別與自動完成。在 Python 中加入型別註記時,請從 openai.types.beta 匯入 Beta 版回應項目型別。
用戶端程式碼範例:
from __future__ import annotations
import json
import sys
from openai import OpenAI
from openai.types.beta import BetaResponseOutputItem
client = OpenAI()
ROOT = "/root"
PROPOSALS = {
"alpha": {"estimated_weeks": 6, "risk": "medium"},
"beta": {"estimated_weeks": 8, "risk": "low"},
}
tools = [
{
"type": "function",
"name": "get_proposal",
"description": "Return details for a proposal that the agents should compare.",
"parameters": {
"type": "object",
"properties": {
"proposal": {
"type": "string",
"enum": ["alpha", "beta"],
}
},
"required": ["proposal"],
"additionalProperties": False,
},
"strict": True,
}
]
history = [
{
"role": "user",
"content": "Compare proposal alpha and proposal beta.",
}
]
def agent_name(item: BetaResponseOutputItem) -> str:
return item.agent.agent_name if item.agent else ROOT
def render_to_user(delta: str) -> None:
print(delta, end="", flush=True)
def log_subagent_text(agent: str, delta: str) -> None:
print(f"[{agent}] {delta}", end="", file=sys.stderr, flush=True)
def process_tool_call(name: str, arguments: str) -> str:
if name != "get_proposal":
raise ValueError(f"Unknown tool: {name}")
parsed_arguments = json.loads(arguments)
return json.dumps(PROPOSALS[parsed_arguments["proposal"]])
while True:
output_items = []
pending_calls = []
item_agents: dict[int, str] = {}
stream = client.beta.responses.create(
model="gpt-5.6-sol",
input=history,
tools=tools,
store=False,
multi_agent={
"enabled": True,
"max_concurrent_subagents": 3,
},
stream=True,
betas=["responses_multi_agent=v1"],
)
for event in stream:
if event.type == "response.output_item.added":
item_agents[event.output_index] = agent_name(event.item)
elif event.type == "response.output_text.delta":
agent = item_agents.get(event.output_index, ROOT)
if agent == ROOT:
render_to_user(event.delta)
else:
log_subagent_text(agent, event.delta)
elif event.type == "response.output_item.done":
output_items.append(event.item)
if event.item.type == "function_call":
# Handle function calls from both the root agent and subagents.
pending_calls.append(event.item)
elif event.type == "response.completed":
print(f"\nUsage: {event.response.usage}", file=sys.stderr)
break
elif event.type in {
"error",
"response.failed",
"response.incomplete",
}:
raise RuntimeError(event)
history.extend(output_items)
for call in pending_calls:
history.append(
{
"type": "function_call_output",
"call_id": call.call_id,
"output": process_tool_call(call.name, call.arguments),
}
)
if not pending_calls:
break如果一個或多個智慧體呼叫開發人員定義的函式,請執行所有待處理的呼叫,並建立包含其輸出的接續請求。
WebSocket
在 WebSocket 模式中,當智慧體呼叫開發人員定義的函式時,請在應用程式中執行該函式,並透過 response.inject 事件將結果傳送至進行中的回應。等待中的智慧體隨後便可繼續執行,無須等待整個多智慧體回應完成。
{
"type": "response.inject",
"response_id": "resp_123",
"input": [
{
"type": "function_call_output",
"call_id": "call_123",
"output": "{\"temperature\":72}"
}
]
}
對於有效的 response.inject 請求,伺服器會回傳下列兩種事件之一:
response.inject.created:輸入已通過驗證,並已獲接受以供注入response.inject.failed:輸入未注入;請檢查error.code
{
"type": "response.inject.created",
"sequence_number": 42,
"response_id": "resp_123"
}
{
"type": "response.inject.failed",
"sequence_number": 43,
"response_id": "resp_123",
"input": [
{
"type": "function_call_output",
"call_id": "call_123",
"output": "{\"temperature\":72}"
}
],
"error": {
"code": "response_already_completed",
"message": "Response 'resp_123' has already completed."
}
}
如果請求不符合 response.inject 結構描述,伺服器會傳送狀態碼為 400 的一般錯誤,並關閉 WebSocket 連線。請修正請求並開啟新的 WebSocket 連線,再傳送下一個事件。
Python Beta 版 SDK 透過 client.beta.responses.connect 提供 WebSocket 模式,TypeScript Beta 版 SDK 則透過 ResponsesWS 提供。請在連線標頭中傳入 OpenAI-Beta: responses_multi_agent=v1;與 HTTP 串流不同,WebSocket 連接器目前尚不接受 betas 引數。
請儲存 response.created 事件中的回應 ID,並在針對該回應傳送的每個 response.inject 事件中包含此 ID。傳送注入項目後,請持續從 WebSocket 讀取,直到回應完成,且每次注入都已產生 response.inject.created 或 response.inject.failed 事件。
from __future__ import annotations
import json
from openai import OpenAI
client = OpenAI()
PROPOSALS = {
"alpha": {"estimated_weeks": 6, "risk": "medium"},
"beta": {"estimated_weeks": 8, "risk": "low"},
}
tools = [
{
"type": "function",
"name": "get_proposal",
"description": "Return details for a proposal that the agents should compare.",
"parameters": {
"type": "object",
"properties": {
"proposal": {
"type": "string",
"enum": ["alpha", "beta"],
}
},
"required": ["proposal"],
"additionalProperties": False,
},
"strict": True,
}
]
def process_tool_call(name: str, arguments: str) -> str:
if name != "get_proposal":
raise ValueError(f"Unknown tool: {name}")
parsed_arguments = json.loads(arguments)
return json.dumps(PROPOSALS[parsed_arguments["proposal"]])
def run_multi_agent(connection):
previous_response_id: str | None = None
pending_input: list[dict[str, object]] = [{"role": "user", "content": input()}]
while pending_input:
request = {
"type": "response.create",
"model": "gpt-5.6-sol",
"store": True,
"multi_agent": {"enabled": True},
"tools": tools,
"input": pending_input,
}
if previous_response_id is not None:
request["previous_response_id"] = previous_response_id
connection.send(request)
next_input: list[dict[str, object]] = []
completed_response = None
response_id: str | None = None
pending_injections = 0
for event in connection:
event_type = event.type
if event_type == "response.created":
response_id = event.response.id
elif event_type == "response.output_item.done":
item = event.item
if item.type == "function_call":
if response_id is None:
raise RuntimeError(
"Received a function call before response.created"
)
output = {
"type": "function_call_output",
"call_id": item.call_id,
"output": process_tool_call(item.name, item.arguments),
}
pending_injections += 1
connection.send(
{
"type": "response.inject",
"response_id": response_id,
"input": [output],
}
)
elif event_type == "response.inject.created":
pending_injections -= 1
elif event_type == "response.inject.failed":
pending_injections -= 1
if event.error.code != "response_already_completed":
raise RuntimeError(event.error)
next_input.extend(item.model_dump(mode="json") for item in event.input)
elif event_type == "response.completed":
completed_response = event.response
elif event_type in {
"error",
"response.failed",
"response.incomplete",
}:
raise RuntimeError(event)
if completed_response is not None and pending_injections == 0:
break
if completed_response is None:
raise RuntimeError("Connection ended before response.completed")
if not next_input:
return completed_response
previous_response_id = completed_response.id
pending_input = next_input
with client.beta.responses.connect(
extra_headers={"OpenAI-Beta": "responses_multi_agent=v1"},
) as connection:
run_multi_agent(connection)傳送 response.inject 事件後,請持續從 WebSocket 讀取,並處理確認訊息:
response.inject.created:函式輸出已加入進行中的回應。請繼續讀取該回應的事件。response.inject.failed且錯誤碼為response_already_completed:函式輸出尚未加入,回應就已完成。請取出失敗事件回傳的input,放入新的response.create請求中傳送,以接續已完成的回應。response.inject.failed且錯誤碼為response_not_found:伺服器找不到response_id所識別的回應。請確認使用的是從response.created收到的 ID。
一次多智慧體執行可能跨越多個 Responses API 請求。使用 HTTP 時,當智慧體呼叫開發人員定義的函式,應用程式會執行該函式,並透過新的 response.create 呼叫提交其輸出。使用 WebSocket 時,應用程式則會將函式輸出注入進行中的回應。
新增的多智慧體輸出項目
多智慧體回應可包含三種額外的輸出項目型別:
multi_agent_call:記錄託管的多智慧體動作,例如spawn_agent。multi_agent_call_output:包含託管動作的執行結果。agent_message:承載一個智慧體傳送給另一個智慧體的加密訊息。
call_id 欄位會將每個 multi_agent_call 連結至對應的 multi_agent_call_output。
每個項目也包含 agent 屬性。對於 agent_message,agent.agent_name 用來識別接收訊息的智慧體。請使用 author 和 recipient 追蹤訊息的傳遞方向。
應用程式收到 multi_agent_call 時,請勿將其當作函式呼叫執行,也不要回傳結果。Responses API 會執行託管動作,並回傳對應的 multi_agent_call_output。如果應用程式需要這兩個項目來重播或追蹤,請將它們一併保留。
[
{
"type": "multi_agent_call",
"id": "mac_123",
"call_id": "call_spawn_a",
"action": "spawn_agent",
"arguments": "{\"task_name\":\"agent_a\",\"fork_turns\":\"all\",\"message\":\"enc_...\"}",
"agent": { "agent_name": "/root" }
},
{
"type": "multi_agent_call_output",
"id": "maco_123",
"call_id": "call_spawn_a",
"action": "spawn_agent",
"output": [
{
"type": "output_text",
"text": "{\"task_name\":\"/root/agent_a\"}",
"annotations": [],
"logprobs": []
}
],
"agent": { "agent_name": "/root" }
},
{
"type": "agent_message",
"id": "amsg_123",
"author": "/root/agent_a",
"recipient": "/root",
"content": [
{
"type": "encrypted_content",
"encrypted_content": "enc_..."
}
],
"agent": { "agent_name": "/root" }
}
]
歸屬於特定智慧體的 SSE 事件會包含頂層的 agent 屬性。對於 agent_message 事件,agent.agent_name 用來識別接收訊息的智慧體。response.created 和 response.completed 等回應生命週期事件描述的是整體回應,而非個別智慧體,因此不包含 agent 屬性。
{
"type": "response.output_item.done",
"agent": { "agent_name": "/root" },
"item": {
"type": "agent_message",
"id": "amsg_123",
"author": "/root/agent_a",
"recipient": "/root",
"content": [
{
"type": "encrypted_content",
"encrypted_content": "enc_..."
}
],
"agent": { "agent_name": "/root" }
}
}
限制
- 壓縮:
- 啟用多智慧體時,不支援
/responses/compact端點。 - 當
multi_agent.enabled設為true時,即使請求未設定context_management,系統也會自動啟用伺服器端自動壓縮。根智慧體與每個子代理程式會各自獨立進行壓縮,保留彼此獨立的上下文。使用者仍可在請求中明確設定context_management.compact_threshold,以覆寫compact_threshold。
- 啟用多智慧體時,不支援
- 啟用多智慧體時,不支援
reasoning.summary。 - 啟用多智慧體時,不支援
max_tool_calls。 max_concurrent_subagents的預設值為3,這也是建議的設定。
提示詞指引
啟用多智慧體時,我們的系統會以新的開發人員訊息,將這些指示自動附加至根智慧體與子代理程式。你無法編輯或移除這些指示;撰寫自己的開發人員指示時,應將其定位為這些自動注入指示的補充。
根智慧體
You are `/root`, the primary agent in a team of agents collaborating to fulfill the user's goals.
At the start of your turn, you are the active agent.
You can spawn sub-agents to handle subtasks, and those sub-agents can spawn their own sub-agents.
All agents in the team, including the agents that you can assign tasks to, are equally intelligent and capable, and have access to the same set of tools.
You can use `spawn_agent` to create a new agent, `followup_task` to give an existing agent a new task and trigger a turn, and `send_message` to pass a message to a running agent without triggering a turn.
Child agents can also spawn their own sub-agents.
You can decide how much context you want to propagate to your sub-agents with the `fork_turns` parameter.
You will receive messages in the form:
```
Message Type: MESSAGE | FINAL_ANSWER
Task name: <recipient>
Sender: <author>
Payload:
<payload text>
```
They may be addressed as to=/root
There are {max_concurrent_subagents + 1} available concurrency slots, meaning that up to {max_concurrent_subagents + 1} agents can be active at once, including you.
子代理程式
You are an agent in a team of agents collaborating to complete a task.
You can spawn sub-agents to handle subtasks, and those sub-agents can spawn their own sub-agents. All agents in the team, including the agents that you can assign tasks to, are equally intelligent and capable, and have access to the same set of tools.
You can use `spawn_agent` to create a new agent, `followup_task` to give an existing agent a new task and trigger a turn, and `send_message` to pass a message to a running agent.
Child agents can also spawn their own sub-agents.
When you provide a response in the final channel, that content is immediately delivered back to your parent agent.
You will receive messages in the form:
```
Message Type: NEW_TASK | MESSAGE | FINAL_ANSWER
Task name: <recipient>
Sender: <author>
Payload:
<payload text>
```
You may also see them addressed as to=/root/..., which indicates your identity is /root/...
There are {max_concurrent_subagents + 1} available concurrency slots, meaning that up to {max_concurrent_subagents + 1} agents can be active at once, including you.