當你將 AI 代理加入工作流程時，必須將它包裝在執行器中，讓工作流程引擎能將訊息路由到它、管理其會話狀態並處理輸出。 代理執行者 是內建的執行者，負責處理此調整。
概觀
代理執行者（Agent Executor）則彌合代理抽象與工作流程執行模型之間的鴻溝。 這樣做：
- 從工作流程圖接收輸入訊息，並將其轉發給底層代理。
- 管理代理在不同執行階段中的連線和對話狀態。
- 根據工作流程執行模式（串流或非串流）調整其行為。
- 將輸出事件（
AgentResponse或AgentResponseUpdate）傳送給工作流程呼叫者進行觀察。 - 將訊息傳送給連接的下游執行器，以便在圖中繼續處理。
- 支援長期執行工作流程的檢查點。
運作方式
在 C# 中，工作流程引擎會在工作流程中為每個AIAgentHostExecutor新增的項目建立一個AIAgent。 此專用執行者擴展 ChatProtocolExecutor 並使用 循環令牌 模式：
-
訊息快取 — 當其他執行者收到訊息時，代理執行者會收集這些訊息。 如果
ForwardIncomingMessages啟用（預設），收到的訊息也會轉發給下游執行器。 -
回合權杖觸發 — 代理僅在收到
TurnToken後才處理其暫存的訊息。 -
代理呼叫 — 執行者呼叫
RunAsync底層代理（非串流）或RunStreamingAsync（串流）。 - 輸出產生 — 如果啟用串流事件，每個增量
AgentResponseUpdate都會以工作流程輸出形式產生。 若啟用了EmitAgentResponseEvents，則聚合後的AgentResponse也會作為工作流程的輸出結果。 - 下游訊息 — 代理的回應訊息會傳送給連接的下游執行者。
-
回合代幣通過 ——執行者完成回合後，會向下送一個新的
TurnToken代幣，讓鏈中的下一個代理人開始處理。
小提示
某些情境可能需要更專門的代理執行器;例如，銜接協調使用帶有自訂路由邏輯的專用HandoffAgentExecutor系統。
隱性創造與顯性創造
當你將一個 AIAgent 傳遞到 WorkflowBuilder時，框架會自動將其包裹在 AIAgentBinding 中，這會產生底層 AIAgentHostExecutor。 你不需要直接實例化代理執行器。
AIAgent writerAgent = /* create your agent */;
AIAgent reviewerAgent = /* create your agent */;
// Agents are automatically wrapped — no manual executor creation required
var workflow = new WorkflowBuilder(writerAgent)
.AddEdge(writerAgent, reviewerAgent)
.Build();
你也可以在AgentWorkflowBuilder上使用輔助工具方法來處理常見設計模式。
// Build a sequential pipeline of agents
var workflow = AgentWorkflowBuilder.BuildSequential(writerAgent, reviewerAgent);
自訂設定
若要自訂代理執行者的行為，請使用 BindAsExecutor ：AIAgentHostOptions
var options = new AIAgentHostOptions
{
EmitAgentUpdateEvents = true,
EmitAgentResponseEvents = true,
ReassignOtherAgentsAsUsers = true,
ForwardIncomingMessages = true,
};
ExecutorBinding writerBinding = writerAgent.BindAsExecutor(options);
var workflow = new WorkflowBuilder(writerBinding)
.AddEdge(writerBinding, reviewerAgent)
.Build();
輸入類型
C# 中的代理執行者接受多種輸入類型：string、、 ChatMessageIEnumerable<ChatMessage>和 。 字串輸入會自動轉換成具有 ChatMessage 角色的 User 實例。 所有收到的訊息會被累積，直到收到 a TurnToken 為止，執行者才會處理批次。 當 ReassignOtherAgentsAsUsers 啟用時（預設值），其他代理的訊息會重新分配到該 User 角色，底層模型將其視為使用者輸入，而目前代理的訊息則保留該 Assistant 角色。
輸出與鏈結
代理程式完成回合後，執行程式：
- 將代理的回應訊息傳送給所有連接的下游執行者。
- 轉發一個新
TurnToken郵件，讓下一個代理人能開始處理。
這使得鏈化代理變得簡單——只需用邊連接它們：
var workflow = new WorkflowBuilder(frenchTranslator)
.AddEdge(frenchTranslator, spanishTranslator)
.AddEdge(spanishTranslator, englishTranslator)
.Build();
串流行為
串流行為可以透過 EmitAgentUpdateEvents 上的 AIAgentHostOptions 選項進行控制，也可以動態地使用 TurnToken 來控制。
-
啟用時 ，執行者會呼叫
RunStreamingAsync代理，並將每個AgentResponseUpdate事件作為工作流程輸出事件。 這提供逐個代幣的即時更新。 -
當被停用 時，執行者會呼叫
RunAsync並產生單一完整回應。
// Enable streaming events at the configuration level
var options = new AIAgentHostOptions
{
EmitAgentUpdateEvents = true,
};
// Or enable streaming dynamically via TurnToken
await run.TrySendMessageAsync(new TurnToken(emitEvents: true));
共享會議
每個代理執行者預設保持自己的會話。 若要在代理間共享會話，請先將代理與共用的會話提供者配置，再將它們加入工作流程。
配置選項
AIAgentHostOptions 控制執行代理人的行為：
| Option | 預設值 | 說明 |
|---|---|---|
EmitAgentUpdateEvents |
null |
執行時會發出串流更新事件。
TurnToken 若設定為 則優先。 如果兩者皆是 null，串流會被關閉。 |
EmitAgentResponseEvents |
false |
將彙整的代理回應作為工作流程輸出事件發出。 |
InterceptUserInputRequests |
false |
攔截 UserInputRequestContent 並將其作為工作流程訊息進行路由以便處理。 |
InterceptUnterminatedFunctionCalls |
false |
攔截 FunctionCallContent，若無對應的結果，則將其作為工作流程訊息路由。 |
ReassignOtherAgentsAsUsers |
true |
將其他代理的訊息重新指派到 User 角色，以便模型將它們視為使用者輸入。 |
ForwardIncomingMessages |
true |
在代理產生的訊息之前，將收到的訊息轉發給下游執行者。 |
檢查點處理
代理執行器支援長時間運行的工作流程檢查點。 當檢查點被取走時，執行者會序列化：
- 代理的會話狀態（透過
SerializeSessionAsync）。 - 當前回合的事件發射配置（僅在請求待處理且執行者尚未交付
TurnToken時才會顯示）。 - 任何待處理的使用者輸入請求和函式呼叫請求。
在還原時，執行器會將會話和待處理請求的狀態反序列化，讓工作流程能從中止的地方繼續。
運作方式
該 AgentExecutor 類別會包裝一個實作該 SupportsAgentRun 協定的代理程式。 當執行人收到訊息時：
-
訊息正規化 — 輸入會被正規化成物件清單
Message，並加入執行者的內部快取。 執行器接受多種輸入類型 —str、Messagelist[str | Message]AgentExecutorRequestAgentExecutorResponse和 —— 每個輸入類型都被導向專用處理器，該處理器在快取前對輸入進行正規化。 -
代理程式呼叫 — 執行者以快取訊息呼叫
agent.run()，根據工作流程執行模式自動選擇串流或非串流模式。 -
輸出發送 — 在串流模式下，每個
AgentResponseUpdate都會作為工作流程的輸出事件被發送。 在非串流模式下，會產生單個AgentResponse。 -
下游派遣 — 代理完成後，執行者會將 a
AgentExecutorResponse傳送給所有連接的下游執行器。 此回應包含完整的對話紀錄，確保順暢銜接。 - 快取重置 — 執行執行者的內部訊息快取會在代理被呼叫後清除，確保每次代理呼叫只處理自上次呼叫以來收到的新訊息。
小提示
某些情境可能需要更專業的代理執行器；例如，交接協調使用專用執行器並使用自訂的路由邏輯。
隱性創造與顯性創造
當你直接傳遞代理時，WorkflowBuilder 會自動將代理包裹在 AgentExecutor 實例中。 對大多數工作流程而言，隱含建立已足夠：
from agent_framework import WorkflowBuilder
writer_agent = client.as_agent(name="Writer", instructions="...")
reviewer_agent = client.as_agent(name="Reviewer", instructions="...")
# Agents are automatically wrapped — no manual AgentExecutor creation required
workflow = (
WorkflowBuilder(start_executor=writer_agent)
.add_edge(writer_agent, reviewer_agent)
.build()
)
明確創造
在需要時明確建立一個 AgentExecutor。
- 讓多名代理人員共享一個會話。
- 提供自訂執行器 ID 以進行路由和具針對性的執行時參數。
- 在多個邊中參考同一個執行者實例。
from agent_framework import AgentExecutor
writer_executor = AgentExecutor(writer_agent, id="my-writer")
reviewer_executor = AgentExecutor(reviewer_agent, id="my-reviewer")
workflow = (
WorkflowBuilder(start_executor=writer_executor)
.add_edge(writer_executor, reviewer_executor)
.build()
)
建造者參數：
| 參數 | 類型 | 說明 |
|---|---|---|
agent |
SupportsAgentRun |
要包裝的代理人。 |
session |
AgentSession \| None |
用於執行代理的 session。 若 None，則由代理程序建立一個新會話。 |
id |
str \| None |
唯一執行人ID。 默認為代理人的名字（如果有的話）。 |
小提示
執行者 ID 也是你鎖定 workflow.run(function_invocation_kwargs=...) 或 client_kwargs= 針對個別代理人時使用的鍵碼。 如果你省略 id，工作流程就會使用封裝代理的名稱。
輸入類型
定義 AgentExecutor 了多種處理方法，每個方法接受不同的輸入類型。 工作流程引擎會根據訊息類型自動派遣正確的處理器。 所有輸入類型都會立即觸發代理執行，唯獨旗標控制代理是否執行或僅快取訊息的情況除外AgentExecutorRequestshould_respond：
| 輸入類型 | 處理器 | 觸發代理程式 | 說明 |
|---|---|---|---|
AgentExecutorRequest |
run |
Conditional | 標準輸入型態。 包含訊息清單以及控制代理是否執行的should_respond 旗標。 |
str |
from_str |
永遠 | 接受原始字串提示。 |
Message |
from_message |
永遠 | 接受單一 Message 物件。 |
list[str \| Message] |
from_messages |
永遠 | 接受一串字串或 Message 物件作為對話上下文。 |
AgentExecutorResponse |
from_response |
永遠 | 接受之前代理執行者的回應，支持直接鏈接。 |
使用 AgentExecutorRequest
AgentExecutorRequest 是典型輸入類型，並提供最多的控制：
from agent_framework import AgentExecutorRequest, Message
# Create a request with messages
request = AgentExecutorRequest(
messages=[Message(role="user", contents=["Hello, world!"])],
should_respond=True,
)
# Run the workflow
result = await workflow.run(request)
該 should_respond 旗標控制代理是立即處理訊息，還是僅僅快取以備後用：
-
True（預設）— 代理人執行並產生回應。 -
False— 訊息被加入快取，但代理程式未執行。 這對於在觸發回應前預先載入對話背景很有用。
輸出與鏈結
代理人完成後，執行人會發送下游指令 AgentExecutorResponse 。 此資料類別包含：
| 領域 | 類型 | 說明 |
|---|---|---|
executor_id |
str |
提供回應的執行人的身分證。 |
agent_response |
AgentResponse |
底層代理人的回應（保持與客戶相同）。 |
full_conversation |
list[Message] \| None |
用於串聯的完整對話上下文（包括先前的輸入和代理的輸出）。 |
當串接代理執行者時，下游執行者會透過AgentExecutorResponse處理器接收from_response。 它利用該 full_conversation 欄位來保存完整的對話歷史，防止下游代理失去先前的上下文：
spam_detector = AgentExecutor(create_spam_detector_agent())
email_assistant = AgentExecutor(create_email_assistant_agent())
# The email_assistant receives the spam_detector's full conversation context
workflow = (
WorkflowBuilder(start_executor=spam_detector)
.add_edge(spam_detector, email_assistant)
.build()
)
串流行為
它 AgentExecutor 會自動適應工作流程執行模式：
-
stream=True— 呼叫agent.run(stream=True)並產生每個AgentResponseUpdate工作流輸出事件。 串流完成後，更新會被聚合成完整的AgentResponse，以供下游調度使用。 -
stream=False（預設） — 呼叫agent.run(stream=False)並產生一個作為工作流程輸出事件的單一AgentResponse。
# Streaming mode — receive incremental updates
events = workflow.run("Write a story about a cat.", stream=True)
async for event in events:
if event.type == "output" and isinstance(event.data, AgentResponseUpdate):
print(event.data.text, end="", flush=True)
# Non-streaming mode — receive complete response
result = await workflow.run("Write a story about a cat.")
# Retrieve AgentResponse objects from the result
outputs = result.get_outputs()
for output in outputs:
if isinstance(output, AgentResponse):
print(output.text)
共享會議
預設每個 AgentExecutor 都會建立自己的會話。 若要在多個代理間共享一個會話（例如，維持共用的對話執行緒），請明確建立一個會話並傳給每個執行者：
from agent_framework import AgentExecutor
# Create a shared session from one agent
shared_session = writer_agent.create_session()
# Both executors share the same session
writer_executor = AgentExecutor(writer_agent, session=shared_session)
reviewer_executor = AgentExecutor(reviewer_agent, session=shared_session)
備註
並非所有客服人員都支援共享會話。 通常，只有同一提供者類型的代理人員才能共用一個會話。
檢查點處理
它 AgentExecutor 支援檢查點功能，用於在長期執行的工作流程中儲存和還原狀態。 當檢查點被取走時，執行者會序列化：
- 內部訊息快取。
- 完整的對話紀錄。
- 代理程式會話狀態。
- 任何待處理的用戶輸入請求與回應。
在還原時，執行者會將此狀態反序列化，讓工作流程能從中止處恢復。
警告
使用伺服器端會話（例如 FoundryAgent）的代理程式設置檢查點有其限制。 伺服器端會話狀態不會被檢查點捕捉，且可透過後續執行進行修改。 如果需要在伺服器端會話中進行可靠的檢查點管理，可以考慮實作自訂的執行器。