Pelaksana

Pelaksana adalah blok penyusun mendasar yang memproses pesan dalam alur kerja. Mereka adalah unit pemrosesan otonom yang menerima pesan yang ditik, melakukan operasi, dan dapat menghasilkan pesan output atau peristiwa.

Overview

Setiap pelaksana memiliki pengidentifikasi unik dan dapat menangani jenis pesan tertentu. Pelaksana dapat berupa:

  • Komponen logika kustom — memproses data, memanggil API, atau mengubah pesan
  • Agen AI — gunakan LLM untuk menghasilkan respons (lihat Agen dalam Alur Kerja)

Important

Cara yang disarankan untuk menentukan handler pesan eksekutor di C# adalah dengan menggunakan [MessageHandler] atribut pada metode dalam partial kelas yang berasal dari Executor. Ini menggunakan pembuatan sumber waktu kompilasi untuk pendaftaran handler, memberikan performa yang lebih baik, validasi waktu kompilasi, dan kompatibilitas AOT Asli.

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

Tip

Eksekutor dapat menyimpan status yang dapat diubah. Jika pelaksana stateful digunakan bersama di seluruh eksekusi alur kerja, pelaksana harus mengimplementasikan IResettableExecutor untuk menghapus status kedaluarsa di antara eksekusi. Lihat Resettable Executors untuk detailnya.

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

Mendeklarasikan Jenis Protokol

Protokol pelaksana menyatakan jenis pesan yang mungkin dikirimnya ke pelaksana yang terhubung dan jenis output yang mungkin dihasilkannya. Alur kerja memvalidasi panggilan ke SendMessageAsync dan YieldOutputAsync terhadap deklarasi ini dan melempar InvalidOperationException ketika eksekutor menggunakan jenis yang tidak dinyatakan.

Gunakan [SendsMessage] untuk mendeklarasikan jenis pesan terkirim dan [YieldsOutput] untuk mendeklarasikan jenis output yang dihasilkan. Atribut ini menjelaskan kemampuan pelaksana; mereka tidak mengirim atau menghasilkan nilai itu sendiri. Terapkan setiap atribut beberapa kali saat pelaksana menggunakan beberapa jenis.

Untuk pelaksana dengan satu handler yang diketik, berasal dari Executor<TInput> atau Executor<TInput, TOutput> dan ambil alih HandleAsync:

internal sealed record ProcessRequest(string Text);
internal sealed record ProgressUpdate(string Status);

[SendsMessage(typeof(ProgressUpdate))]
[YieldsOutput(typeof(string))]
internal sealed partial class ProcessingExecutor()
    : Executor<ProcessRequest>("ProcessingExecutor")
{
    public override async ValueTask HandleAsync(
        ProcessRequest message,
        IWorkflowContext context,
        CancellationToken cancellationToken = default)
    {
        await context.SendMessageAsync(
            new ProgressUpdate("Processing started"),
            cancellationToken);

        await context.YieldOutputAsync(
            message.Text.ToUpperInvariant(),
            cancellationToken);
    }
}

Ketika generator sumber alur kerja direferensikan, kelas dengan [SendsMessage] atau [YieldsOutput] harus dideklarasikan partial sehingga generator dapat menambahkan konfigurasi protokolnya.

Untuk pelaksana yang dihasilkan sumber dengan [MessageHandler] metode, nyatakan jenis yang digunakan oleh satu handler dengan argumennya Send dan Yield bernama, seperti [MessageHandler(Send = [typeof(ProgressUpdate)], Yield = [typeof(string)])]. Gunakan tingkat [SendsMessage] kelas dan [YieldsOutput] saat deklarasi berlaku untuk seluruh pelaksana.

Jenis pengembalian handler non-void secara otomatis ditambahkan ke jenis protokol yang dikirim dan dihasilkan saat ExecutorOptions.AutoSendMessageHandlerResultObject dan ExecutorOptions.AutoYieldOutputHandlerResultObject diaktifkan. Kedua opsi diaktifkan secara default. Oleh karena itu, deklarasi eksplisit diperlukan untuk jenis tambahan yang dipancarkan langsung melalui SendMessageAsync atau YieldOutputAsync.

[YieldsOutput] mengizinkan pelaksana untuk menghasilkan jenis, tetapi tidak menunjuk pelaksana sebagai sumber output terminal. Daftarkan pelaksana dengan WorkflowBuilder.WithOutputFrom untuk nilai yang dihasilkan untuk muncul ke pemanggil alur kerja.

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:

Important

Saat menggunakan parameter jenis eksplisit, Anda harus menentukan semua jenis melalui dekorator — Anda tidak dapat mencampur parameter eksplisit dengan anotasi jenis. Parameter input diperlukan; output dan workflow_output bersifat opsional.

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

Important

ctx.yield_output(...) tidak memiliki bendera per emisi. Panggilan yang sama dilabeli "output" atau "intermediate" hanya berdasarkan sebutan dari pembuatnya. Tidak ada ctx.yield_intermediate(...) API — penamaan tidak bervariasi berdasarkan rendemen.

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:

Hook Kegunaan
ConfigureProtocol Menyiapkan perutean pesan dan jenis kirim/hasil yang dinyatakan
InitializeFunc Penyiapan saat instans eksekutor dibuat untuk suatu proses eksekusi
ResetFunc Reset status eksekutor-lokal sebelum digunakan kembali
OnCheckpointFunc Simpan status di titik pemeriksaan
OnCheckpointRestoredFunc Memulihkan status dari titik pemeriksaan
OnMessageDeliveryStartingFunc Jalankan sebelum superstep mengirimkan pesan
OnMessageDeliveryFinishedFunc Jalankan setelah superstep selesai mengirimkan pesan 
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()

Langkah berikutnya

Tepi

  • Last updated on