Egzekutorzy

Funkcje wykonawcze to podstawowe bloki konstrukcyjne, które przetwarzają komunikaty w przepływie pracy. Są to autonomiczne jednostki przetwarzania, które odbierają komunikaty typizowane, wykonują operacje i mogą generować komunikaty wyjściowe lub zdarzenia.

Overview

Każdy funkcja wykonawcza ma unikatowy identyfikator i może obsługiwać określone typy komunikatów. Wykonawcy mogą być:

• Niestandardowe składniki logiki — przetwarzanie danych, wywoływanie interfejsów API lub przekształcanie komunikatów
• Agenci sztucznej inteligencji — generowanie odpowiedzi przy użyciu modułów LLM (zobacz Agenci w przepływach pracy)

Ważna

Zalecanym sposobem definiowania procedur obsługi komunikatów funkcji wykonawczej w języku C# jest użycie atrybutu [MessageHandler] w metodach w partial klasie pochodzącej z Executorklasy . To używa generowania kodu źródłowego w czasie kompilacji na potrzeby rejestracji obsługi, zapewniając lepszą wydajność, walidację czasu kompilacji i zgodność z natywną funkcją AOT.

Podstawowa struktura funkcji wykonawczej

Funkcje wykonawcze pochodzą z klasy bazowej Executor i używają atrybutu [MessageHandler] do deklarowania metod obsługi. Klasa musi być oznaczona partial w celu włączenia generowania źródła.

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

Komunikaty można również wysyłać ręcznie bez zwracania wartości:

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

Wskazówka

Funkcje wykonawcze mogą przechowywać stan modyfikowalny. Jeśli stanowy wykonawca jest współdzielony w trakcie przebiegów przepływu pracy, musi zaimplementować IResettableExecutor, aby wyczyścić nieaktualny stan między przebiegami. Aby uzyskać szczegółowe informacje, zobacz Resettable Executors.

Wiele typów danych wejściowych

Obsługa wielu typów danych wejściowych przez zdefiniowanie wielu [MessageHandler] metod:

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

Wykonawcy oparci na funkcjach

Utwórz wykonawcę z funkcji przy użyciu metody rozszerzenia BindExecutor.

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

Obiekt IWorkflowContext

IWorkflowContext zapewnia metody interakcji z przepływem pracy w trakcie wykonania:

• SendMessageAsync — wysyłanie komunikatów do połączonych funkcji wykonawczych
• YieldOutputAsync — generowanie danych wyjściowych przepływu pracy zwracanych/przesyłanych strumieniowo do obiektu wywołującego

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

Jeśli program obsługi ani nie wysyła komunikatów, ani nie zwraca danych wyjściowych, może po prostu wykonywać skutki uboczne:

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

Deklarowanie typów protokołów

Protokół funkcji wykonawczej deklaruje typy komunikatów, które mogą być wysyłane do połączonych funkcji wykonawczych i typy danych wyjściowych, które mogą zwracać. Przepływ pracy weryfikuje InvalidOperationException wywołania i YieldOutputAsyncSendMessageAsync względem tych deklaracji oraz zgłasza błąd, gdy funkcja wykonawcza używa typu niezdecydowanego.

Służy [SendsMessage] do deklarowania typów wysłanych komunikatów i [YieldsOutput] deklarowania zwracanych typów danych wyjściowych. Te atrybuty opisują możliwości funkcji wykonawcy; nie wysyłają ani nie dają samych wartości. Zastosuj każdy atrybut wiele razy, gdy funkcja wykonawcza używa wielu typów.

W przypadku funkcji wykonawczych z pojedynczą procedurą obsługi typizowanej należy utworzyć metodę Executor<TInput> lub Executor<TInput, TOutput> i zastąpić HandleAsyncpolecenie :

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

Po odwoływaniu się do generatora źródła przepływów pracy należy zadeklarować partial klasę z elementem [SendsMessage] lub [YieldsOutput] , aby generator mógł dodać konfigurację protokołu.

W przypadku funkcji wykonawczych wygenerowanych przez źródło z metodami [MessageHandler] zadeklaruj typy używane przez jedną procedurę obsługi z argumentami o Send nazwach, Yield takich jak [MessageHandler(Send = [typeof(ProgressUpdate)], Yield = [typeof(string)])]. Użyj klasy poziom [SendsMessage] i [YieldsOutput] kiedy deklaracje mają zastosowanie do całej funkcji wykonawczej.

Typy zwracane przez program obsługi niepustej są automatycznie dodawane do typów wysyłanych i zwracanych protokołów, gdy ExecutorOptions.AutoSendMessageHandlerResultObject są włączone i ExecutorOptions.AutoYieldOutputHandlerResultObject włączone. Obie opcje są domyślnie włączone. W związku z tym deklaracje jawne są wymagane głównie w przypadku dodatkowych typów emitowanych bezpośrednio za pośrednictwem SendMessageAsync lub YieldOutputAsync.

[YieldsOutput] umożliwia wykonawcy uzyskanie typu, ale nie wyznacza funkcji wykonawczej jako źródła danych wyjściowych terminalu. Zarejestruj funkcję wykonawcza, WorkflowBuilder.WithOutputFrom aby uzyskać zwrócone wartości, aby wyświetlić obiekt wywołujący przepływ pracy.

Podstawowa struktura funkcji wykonawczej

Funkcje wykonawcze dziedziczą z klasy bazowej Executor . Każdy wykonawca używa metod ozdobionych dekoratorem @handler . Programy obsługi muszą mieć właściwe adnotacje typu, aby określić rodzaje komunikatów, które przetwarzają.

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

Wykonawcy oparci na funkcjach

Tworzenie funkcji wykonawczej na podstawie funkcji przy użyciu dekoratora @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())

Wiele typów danych wejściowych

Obsługa wielu typów danych wejściowych przez zdefiniowanie wielu procedur obsługi:

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)

Jawne parametry typu

Alternatywą dla adnotacji typów jest jawne określenie typów za pomocą parametrów dekoratora:

Ważna

W przypadku używania jawnych parametrów typu należy określić wszystkie typy za pośrednictwem dekoratora — nie można mieszać jawnych parametrów z adnotacjami typów. Parametr input jest wymagany i outputworkflow_output jest opcjonalny.

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)

Obiekt WorkflowContext

WorkflowContext zapewnia metody interakcji z przepływem pracy w trakcie wykonania:

• send_message — wysyłanie komunikatów do połączonych funkcji wykonawczych
• yield_output — generowanie danych wyjściowych przepływu pracy zwracanych/przesyłanych strumieniowo do obiektu wywołującego

class OutputExecutor(Executor):

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

Jeśli program obsługi nie wysyła komunikatów ani nie zwraca danych wyjściowych, żaden parametr typu nie jest wymagany:

class LogExecutor(Executor):

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

Wyznaczanie funkcji wykonawczych terminalu i danych wyjściowych pośrednich

To, które egzekutory przyczyniają się do końcowej odpowiedzi przepływu pracy, a które emitują obserwowany postęp, jest decyzją podejmowaną na etapie kompilacji, konfigurowaną w WorkflowBuilder, a nie flagą ustawianą osobno dla każdej emisji.

• output_from — egzekutory, których wywołania ctx.yield_output(...) powodują wystąpienie zdarzeń "output" i są zwracane przez WorkflowRunResult.get_outputs().
• intermediate_output_from — egzekutory, których wywołania ctx.yield_output(...) powodują wystąpienie zdarzeń "intermediate" i są zwracane przez 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()

Ważna

ctx.yield_output(...) nie ma flagi dla każdej emisji. To samo wywołanie jest oznaczone jako "output" lub "intermediate" wyłącznie ze względu na nazwę kreatora. Nie istnieje interfejs API ctx.yield_intermediate(...) — oznaczenie nie zmienia się w zależności od uzysku.

Obie listy są opcjonalne. Jeśli podano którąkolwiek z list wyboru wyjść, egzekutor, który nie pojawia się na żadnej z nich, nadal może wysyłać komunikaty do egzekutorów podrzędnych za pośrednictwem ctx.send_message(...), ale jego wywołania yield_output są ukryte. Jeśli obie listy zostaną pominięte, każdy yield_output nadal generuje "output" dla zachowania zgodności.

Podstawowa struktura funkcji wykonawczej

Funkcje wykonawcze to jednostki przetwarzania w przepływie pracy. Otrzymują dane wejściowe, wykonują pracę i generują dane wyjściowe.

Wiele typów danych wejściowych

Zarejestruj wiele programów obsługi, konfigurując ścieżki w executorze:

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

Wykonawcy oparci na funkcjach

Najprostszym sposobem utworzenia funkcji wykonawczej jest użycie polecenia workflow.NewExecutor(...).Bind():

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

Funkcje wykonawcze automatycznie rejestrują typ danych wejściowych i mogą automatycznie wysyłać i automatycznie zwracać zwracane wartości.

Przepływ pracy. Obiekt kontekstu

Programy obsługi mogą zaakceptować *workflow.Context interakcję z przepływem pracy podczas wykonywania:

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

Kontekst uwidacznia również interfejsy API, takie jak SendMessage, AddEvent, PostRequest, ReadStatei QueueStateUpdate.

Egzekutory agenta

Agenci mogą służyć jako funkcje wykonawcze przepływu pracy za pomocą polecenia agentworkflow.New:

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

Cykl życia funkcji wykonawczej

Egzekutory obsługują hooki cyklu życia za pomocą pól w workflow.Executor:

Haczyk	Purpose
ConfigureProtocol	Konfigurowanie routingu komunikatów i zadeklarowanych typów wysyłania/wydajności
InitializeFunc	Konfiguracja podczas tworzenia instancji egzekutora dla uruchomienia
ResetFunc	Resetuj stan funkcji wykonawczej lokalnej przed ponownym użyciem
OnCheckpointFunc	Zapisz stan w punkcie kontrolnym
OnCheckpointRestoredFunc	Przywróć stan z punktu kontrolnego
OnMessageDeliveryStartingFunc	Uruchamianie przed superkrokiem dostarcza komunikaty
OnMessageDeliveryFinishedFunc	Uruchamianie po zakończeniu dostarczania komunikatów przez superkrok

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

Następne kroki

Krawędzie