Pelaksana

Struktur Pelaksana Dasar

Eksekutor berasal dari kelas dasar Executor dan menggunakan atribut [MessageHandler] untuk mendeklarasi metode handler. Kelas harus ditandai partial untuk mengaktifkan pembuatan sumber.

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
    }
}

Anda juga dapat mengirim pesan secara manual tanpa mengembalikan nilai:

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
    }
}

Beberapa Jenis Input

Tangani beberapa jenis input dengan mendefinisikan beberapa [MessageHandler] metode:

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);
    }
}

Eksekutor Berbasis Fungsi

Buat pelaksana dari fungsi menggunakan BindExecutor metode ekstensi:

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

Objek IWorkflowContext

IWorkflowContext menyediakan metode-metode untuk berinteraksi dengan alur kerja selama pelaksanaannya.

• SendMessageAsync — mengirim pesan ke pelaksana yang terhubung
• YieldOutputAsync — menghasilkan output alur kerja yang dikembalikan/dialirkan ke pemanggil

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

Jika handler tidak mengirim pesan atau menghasilkan output, itu hanya dapat melakukan efek samping:

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

Struktur Pelaksana Dasar

Eksekutor mewarisi dari kelas dasar Executor. Setiap pelaksana menggunakan metode yang dihiasi dengan @handler dekorator. Pengendali harus memiliki anotasi tipe yang tepat untuk menentukan jenis pesan yang diproses.

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())

Eksekutor Berbasis Fungsi

Buat eksekutor dari fungsi menggunakan dekorator @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())

Beberapa Jenis Input

Tangani beberapa jenis input dengan menentukan beberapa handler:

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)

Parameter Jenis Eksplisit

Sebagai alternatif untuk mengetik anotasi, Anda dapat menentukan jenis secara eksplisit melalui parameter dekorator:

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)

Objek WorkflowContext

WorkflowContext menyediakan metode-metode untuk berinteraksi dengan alur kerja selama pelaksanaannya.

• send_message — mengirim pesan ke pelaksana yang terhubung
• yield_output — menghasilkan output alur kerja yang dikembalikan/dialirkan ke pemanggil

class OutputExecutor(Executor):

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

Jika handler tidak mengirim pesan atau menghasilkan output, parameter jenis tidak diperlukan:

class LogExecutor(Executor):

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

Menunjuk Terminal dan Pelaksana Output Menengah

Pelaksana mana yang berkontribusi pada jawaban terminal alur kerja dan yang memancarkan kemajuan pengamatan adalah keputusan build-time yang dikonfigurasi pada WorkflowBuilder, bukan bendera per emisi.

• output_from — eksekutor yang pemanggilan ctx.yield_output(...)-nya menghasilkan event "output" dan dikembalikan oleh WorkflowRunResult.get_outputs().
• intermediate_output_from — eksekutor yang pemanggilan ctx.yield_output(...)-nya menghasilkan event "intermediate" dan dikembalikan oleh 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()

Kedua daftar bersifat opsional. Jika salah satu daftar pemilihan keluaran ditentukan, eksekutor yang tidak muncul di salah satu dari kedua daftar tersebut masih dapat mengirim pesan ke eksekutor downstream melalui ctx.send_message(...), tetapi pemanggilan yield_output-nya disembunyikan. Jika kedua daftar dihilangkan, setiap yield_output tetap menghasilkan "output" demi kompatibilitas.

Struktur Pelaksana Dasar

Pelaksana adalah unit pemrosesan dalam alur kerja. Mereka menerima input, melakukan pekerjaan, dan menghasilkan output.

Beberapa Jenis Input

Daftarkan beberapa handler dengan mengonfigurasi rute pada eksekutor:

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()

Eksekutor Berbasis Fungsi

Cara paling sederhana untuk membuat pelaksana adalah dengan workflow.NewExecutor(...).Bind():

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

Pelaksana fungsi secara otomatis mendaftarkan jenis input dan dapat mengirim dan menghasilkan nilai yang dikembalikan secara otomatis.

Alur kerja. Objek Konteks

Handler dapat menerima *workflow.Context untuk berinteraksi dengan alur kerja selama eksekusi:

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

Konteks ini juga mengekspos API seperti SendMessage, , AddEvent, PostRequestReadState, dan QueueStateUpdate.

Pelaksana Agen

Agen dapat digunakan sebagai pelaksana alur kerja melalui agentworkflow.New:

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

Siklus Hidup Pelaksana

Pelaksana mendukung kait siklus hidup melalui bidang pada workflow.Executor:

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()