執行人

執行程式是在工作流程中處理訊息的基本建置區塊。它們是自主處理單元，可接收類型化訊息、執行作業，並可產生輸出訊息或事件。

概觀

每個執行器都有一個唯一的標識符，可以處理特定的訊息類型。遺囑執行人可以是：

自訂邏輯元件 — 處理資料、呼叫 API 或轉換訊息
AI 代理 — 使用大型語言模型（LLMs）來產生回應（參見工作流程中的代理）

這很重要

在 C# 中定義執行器訊息處理器的建議方法是在一個由 [MessageHandler] 衍生的類別的方法上使用 partial 屬性。此系統使用編譯時的原始碼生成進行處理程序註冊，提供更佳的效能、編譯時驗證及 AOT 原生相容。

基本執行器結構

執行程序源自 Executor 基底類別，並使用屬性 [MessageHandler] 來宣告處理器方法。類別必須被標記 partial 以啟用原始碼產生。

using Microsoft.Agents.AI.Workflows;

internal sealed partial class UppercaseExecutor() : Executor("UppercaseExecutor")
{
    [MessageHandler]
    private ValueTask<string> HandleAsync(string message, IWorkflowContext context)
    {
        string result = message.ToUpperInvariant();
        return ValueTask.FromResult(result); // Return value is automatically sent to connected executors
    }
}

你也可以手動發送訊息而不回傳一個值：

internal sealed partial class UppercaseExecutor() : Executor("UppercaseExecutor")
{
    [MessageHandler]
    private async ValueTask HandleAsync(string message, IWorkflowContext context)
    {
        string result = message.ToUpperInvariant();
        await context.SendMessageAsync(result); // Manually send messages to connected executors
    }
}

小提示

執行者可以保持可變狀態。如果狀態化執行器在多個工作流程執行間被共享，必須實作 IResettableExecutor 以在執行間清除過期狀態。詳情請參見可重置執行者。

多重輸入類型

透過定義多種 [MessageHandler] 方法來處理多種輸入類型：

internal sealed partial class SampleExecutor() : Executor("SampleExecutor")
{
    [MessageHandler]
    private ValueTask<string> HandleStringAsync(string message, IWorkflowContext context)
    {
        return ValueTask.FromResult(message.ToUpperInvariant());
    }

    [MessageHandler]
    private ValueTask<int> HandleIntAsync(int message, IWorkflowContext context)
    {
        return ValueTask.FromResult(message * 2);
    }
}

Function-Based 執行器

使用 BindExecutor 擴充方法從函式建立執行器：

Func<string, string> uppercaseFunc = s => s.ToUpperInvariant();
var uppercase = uppercaseFunc.BindExecutor("UppercaseExecutor");

IWorkflowContext 物件

在執行期間，IWorkflowContext 提供與工作流程互動的方法：

SendMessageAsync — 向連接的執行者發送訊息
YieldOutputAsync — 產生回傳/串流給呼叫者的工作流程輸出

internal sealed partial class OutputExecutor() : Executor("OutputExecutor")
{
    [MessageHandler]
    private async ValueTask HandleAsync(string message, IWorkflowContext context)
    {
        await context.YieldOutputAsync("Hello, World!");
    }
}

如果處理器既不發送訊息也不輸出，它可以直接執行以下副作用：

internal sealed partial class LogExecutor() : Executor("LogExecutor")
{
    [MessageHandler]
    private void Handle(string message, IWorkflowContext context)
    {
        Console.WriteLine("Doing some work...");
    }
}

基本執行器結構

執行程式繼承自 Executor 基類。每個執行者都使用由 @handler 修飾器修飾的方法。處理器必須具備適當的型別註解，以指定其所處理的訊息類型。

from agent_framework import (
    Executor,
    WorkflowContext,
    handler,
)

class UpperCase(Executor):

    @handler
    async def to_upper_case(self, text: str, ctx: WorkflowContext[str]) -> None:
        """Convert the input to uppercase and forward it to the next node."""
        await ctx.send_message(text.upper())

Function-Based 執行器

使用 @executor 裝飾器從函式建立執行器：

from agent_framework import (
    WorkflowContext,
    executor,
)

@executor(id="upper_case_executor")
async def upper_case(text: str, ctx: WorkflowContext[str]) -> None:
    """Convert the input to uppercase and forward it to the next node."""
    await ctx.send_message(text.upper())

多重輸入類型

處理多種類型的輸入可透過定義多個處理常式來實現。

class SampleExecutor(Executor):

    @handler
    async def to_upper_case(self, text: str, ctx: WorkflowContext[str]) -> None:
        await ctx.send_message(text.upper())

    @handler
    async def double_integer(self, number: int, ctx: WorkflowContext[int]) -> None:
        await ctx.send_message(number * 2)

明確型別參數

作為類型註解的替代方案，你可以透過裝飾器參數明確指定類型：

這很重要

使用明確型別參數時，必須透過裝飾器指定所有型別——不能將明確參數與型別註解混用。 input該參數為必備;output且workflow_output為可選。

class ExplicitTypesExecutor(Executor):

    @handler(input=str, output=str)
    async def to_upper_case(self, text, ctx) -> None:
        await ctx.send_message(text.upper())

    @handler(input=str | int, output=str)
    async def handle_mixed(self, message, ctx) -> None:
        await ctx.send_message(str(message).upper())

    @handler(input=str, output=int, workflow_output=bool)
    async def process_with_workflow_output(self, message, ctx) -> None:
        await ctx.send_message(len(message))
        await ctx.yield_output(True)

WorkflowContext 物件

在執行期間，WorkflowContext 提供與工作流程互動的方法：

send_message — 向連接的執行者發送訊息
yield_output — 產生回傳/串流給呼叫者的工作流程輸出

class OutputExecutor(Executor):

    @handler
    async def handle(self, message: str, ctx: WorkflowContext[Never, str]) -> None:
        await ctx.yield_output("Hello, World!")

如果處理器既不發送訊息也不輸出，則不需要類型參數：

class LogExecutor(Executor):

    @handler
    async def handle(self, message: str, ctx: WorkflowContext) -> None:
        print("Doing some work...")

指定終端與中介輸出執行器

哪些執行器會影響工作流程的最終答案、哪些會輸出觀察性進度資訊，是在上配置的WorkflowBuilder決策，而不是每次輸出時的旗標。

output_from — 其 ctx.yield_output(...) 呼叫會產生 "output" 事件，且由 WorkflowRunResult.get_outputs() 傳回的執行器。
intermediate_output_from — 其 ctx.yield_output(...) 呼叫會產生 "intermediate" 事件，且由 WorkflowRunResult.get_intermediate_outputs() 傳回的執行器。

from agent_framework import WorkflowBuilder

workflow = WorkflowBuilder(
    start_executor=analysis_executor,
    output_from=[summary_executor],
    intermediate_output_from=[analysis_executor],
).build()

這很重要

ctx.yield_output(...)沒有任何每次發送的旗標。同一個呼叫會被標示為 "output" 或 "intermediate"，完全取決於建置者的命名。目前沒有 ctx.yield_intermediate(...) API規定——指定不會因產量而異。

這兩個名單都是可選的。若提供任一輸出選擇清單，未出現在兩個清單中的執行器仍可透過 ctx.send_message(...)向下游執行器發送訊息，但其 yield_output 呼叫會被隱藏。若省略這兩個清單，為了保持相容性，每個 yield_output 仍會發出 "output"。

基本執行器結構

執行器是工作流程中的處理單元。他們接收輸入、執行工作並產生產出。

多重輸入類型

透過在執行器上設定路由來註冊多個處理常式：

sample := (&workflow.Executor{
    ID: "SampleExecutor",
    ConfigureProtocol: func(pb *workflow.ProtocolBuilder) (*workflow.ProtocolBuilder, error) {
        pb.RouteBuilder.
            AddHandlerRaw(reflect.TypeFor[string](), reflect.TypeFor[string](), func(_ *workflow.Context, msg any) (any, error) {
                return strings.ToUpper(msg.(string)), nil
            }).
            AddHandlerRaw(reflect.TypeFor[int](), reflect.TypeFor[int](), func(_ *workflow.Context, msg any) (any, error) {
                return msg.(int) * 2, nil
            })
        return pb, nil
    },
}).Bind()

Function-Based 執行器

建立執行人最簡單的方法是：workflow.NewExecutor(...).Bind()

uppercase := workflow.NewExecutor("UppercaseExecutor", func(input string) string {
    return strings.ToUpper(input)
}).Bind()

函式執行器會自動註冊輸入類型，並能自動傳送及自動回傳回傳的值。

workflow.Context 物件

處理常式可接受 *workflow.Context 作為參數，以便在執行期間與工作流程互動：

output := workflow.NewExecutor("OutputExecutor", func(ctx *workflow.Context, message string) error {
    return ctx.YieldOutput("Hello, World!")
}).Bind()

上下文也會公開 API，例如 SendMessage、AddEvent、PostRequest、ReadState 和 QueueStateUpdate 等。

代理執行器

代理可透過以下 agentworkflow.New方式作為工作流程執行者使用：

agentExecutor := agentworkflow.New(myAgent, agentworkflow.Config{
    EmitUpdateEvents: true,
})

執行者生命週期

執行者透過以下 workflow.Executor欄位支援生命週期鉤子：

勾點	Purpose
`ConfigureProtocol`	設定訊息路由並宣告發送/收益類型
`InitializeFunc`	為某次執行建立執行器執行個體時的設定
`ResetFunc`	在重用前重置執行者本地狀態
`OnCheckpointFunc`	在檢查點儲存狀態
`OnCheckpointRestoredFunc`	從檢查點還原狀態
`OnMessageDeliveryStartingFunc`	在超步驟傳遞訊息之前執行
`OnMessageDeliveryFinishedFunc`	在超步驟完成訊息傳遞後執行

stateful := workflow.NewExecutor("StatefulExecutor", handleMessage).Extend(&workflow.Executor{
    InitializeFunc: func(ctx *workflow.Context) error {
        return nil
    },
    ResetFunc: func() error {
        return nil
    },
    OnCheckpointFunc: func(ctx *workflow.Context) error {
        return ctx.QueueStateUpdate("StatefulExecutorState", "", currentState)
    },
    OnCheckpointRestoredFunc: func(ctx *workflow.Context) error {
        restored, err := ctx.ReadState("StatefulExecutorState", "")
        if err != nil {
            return err
        }
        currentState = restored
        return nil
    },
}).Bind()

後續步驟

邊緣

意見反應

此頁面對您有幫助嗎？

Last updated on 2026-07-10

執行人

概觀

基本執行器結構

多重輸入類型

Function-Based 執行器

IWorkflowContext 物件

基本執行器結構

Function-Based 執行器

多重輸入類型

明確型別參數

WorkflowContext 物件

指定終端與中介輸出執行器

基本執行器結構

多重輸入類型

Function-Based 執行器

workflow.Context 物件

代理執行器

執行者生命週期

後續步驟

意見反應

其他資源