Udostępnij za pośrednictwem

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.

Przegląd

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żne

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

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żne

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

Dalsze kroki

Krawędzie

  • Last updated on