Végrehajtók

A végrehajtók a munkafolyamat üzeneteinek feldolgozásának alapvető építőelemei. Ezek önálló feldolgozási egységek, amelyek gépelt üzeneteket fogadnak, műveleteket hajtanak végre, és kimeneti üzeneteket vagy eseményeket képesek létrehozni.

Overview

Minden végrehajtó egyedi azonosítóval rendelkezik, és képes kezelni bizonyos üzenettípusokat. A végrehajtók lehetnek:

• Egyéni logikai összetevők – adatok feldolgozása, API-k meghívása vagy üzenetek átalakítása
• AI-ügynökök – a válaszok létrehozásához használjon LLM-eket (lásd : Ügynökök a munkafolyamatokban)

Important

A végrehajtó üzenetkezelők C#-ban való definiálásának ajánlott módja az, hogy az attribútumot olyan metódusokra alkalmazzuk, amelyek egy [MessageHandler]-ből származó partial osztályon belül találhatók. Ez fordítási idejű forráslétrehozást használ a kezelőregisztrációhoz, jobb teljesítményt, fordítási idő érvényesítést és natív AOT-kompatibilitást biztosít.

Egyszerű végrehajtói struktúra

A végrehajtók az Executor alaposztályból származnak, és az [MessageHandler] attribútum használatával deklarálják a kezelő metódusokat. Az osztályt meg kell jelölni partial a forráslétrehozás engedélyezéséhez.

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

Manuálisan is küldhet üzeneteket érték visszaadása nélkül:

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

A végrehajtók tárolhatják a módosítható állapotot. Ha egy állapotalapú végrehajtó meg van osztva a munkafolyamat-futtatások között, implementálnia IResettableExecutor kell a futtatások közötti elavult állapot törléséhez. A részletekért lásd a visszaállítható végrehajtók részt.

Több bemeneti típus

Több bemeneti típus kezelése több [MessageHandler] metódus definiálásával:

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

Funkcióalapú végrehajtók

Hozzon létre egy végrehajtót egy függvényből a bővítménymetódus BindExecutor használatával:

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

Az IWorkflowContext objektum

A IWorkflowContext módszereket kínál a munkafolyamat végrehajtása közbeni interakcióhoz.

• SendMessageAsync — üzenetek küldése csatlakoztatott végrehajtóknak
• YieldOutputAsync — a hívónak visszaadott/streamelt munkafolyamat-kimenetek létrehozása

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

Ha egy kezelő nem küld üzeneteket, és nem ad ki kimenetet, egyszerűen mellékhatásokat hajthat végre:

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

Protokolltípusok deklarálása

A végrehajtó protokollja deklarálja, hogy milyen üzenettípusokat küldhet a csatlakoztatott végrehajtóknak, és milyen kimeneti típusokat hozhat létre. A munkafolyamat ellenőrzi ezen deklarációk hívásait SendMessageAsync , YieldOutputAsync és értesítést ad InvalidOperationException , ha egy végrehajtó be nem jelentett típust használ.

Az elküldött üzenettípusok deklarálásához és [YieldsOutput] a kimeneti típusok deklarálásához használható[SendsMessage]. Ezek az attribútumok a végrehajtó képességeit írják le; nem küldenek és nem adnak értéket maguknak. Az egyes attribútumokat többször is alkalmazhatja, amikor a végrehajtó több típust használ.

Egyetlen típusú kezelővel rendelkező végrehajtók esetén a következőből Executor<TInput> származtathatja vagy Executor<TInput, TOutput> felülbírálhatja 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);
    }
}

A munkafolyamatok forrásgenerátorára való hivatkozáskor egy osztályt [SendsMessage] kell deklarálni vagy [YieldsOutput] deklarálni partial ahhoz, hogy a generátor hozzá tudja adni a protokollkonfigurációját.

A metódusokkal [MessageHandler] rendelkező, forrás által létrehozott végrehajtók esetében deklarálja az egyik kezelő által használt típusokat az SendYield argumentumaival, például [MessageHandler(Send = [typeof(ProgressUpdate)], Yield = [typeof(string)])]. Osztályszintű [SendsMessage] és [YieldsOutput] amikor a deklarációk a teljes végrehajtóra vonatkoznak.

A rendszer automatikusan hozzáadja a nem érvénytelen kezelő visszatérési típusait az elküldött és a visszaadott protokolltípusokhoz, amikor ExecutorOptions.AutoSendMessageHandlerResultObject engedélyezve vannak és ExecutorOptions.AutoYieldOutputHandlerResultObject engedélyezve vannak. Alapértelmezés szerint mindkét beállítás engedélyezve van. Ezért elsősorban explicit deklarációkra van szükség a közvetlenül vagy YieldOutputAsynca .-on keresztül SendMessageAsync kibocsátott további típusok esetében.

[YieldsOutput] lehetővé teszi a végrehajtó számára, hogy adjon meg egy típust, de nem jelöli meg a végrehajtót terminálkimeneti forrásként. Regisztrálja a végrehajtót WorkflowBuilder.WithOutputFrom , hogy a hozamértékek felszínre lépjenek a munkafolyamat-hívóban.

Egyszerű végrehajtói struktúra

A végrehajtók az Executor alaposztálytól öröklődnek. Minden végrehajtó a @handler dekoratőrrel ellátott metódusokat használja. A kezelőknek megfelelő típusú széljegyzetekkel kell rendelkezniük a feldolgozott üzenettípusok megadásához.

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

Funkcióalapú végrehajtók

Hozzon létre egy végrehajtót egy függvényből a @executor dekorátor használatával:

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

Több bemeneti típus

Több bemeneti típus kezelése több kezelő definiálásával:

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)

Explicit típusparaméterek

A széljegyzetek típusának alternatívaként a dekorátorparamétereken keresztül explicit módon adhatja meg a típusokat:

Important

Explicit típusparaméterek használatakor minden típust meg kell adnia a dekorátoron keresztül – az explicit paraméterek nem keverhetők össze típusjegyzetekkel. A input paraméter megadása kötelező, output és workflow_output nem kötelező.

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)

A WorkflowContext objektum

A WorkflowContext módszereket kínál a munkafolyamat végrehajtása közbeni interakcióhoz.

• send_message — üzenetek küldése csatlakoztatott végrehajtóknak
• yield_output — a hívónak visszaadott/streamelt munkafolyamat-kimenetek létrehozása

class OutputExecutor(Executor):

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

Ha egy kezelő nem küld üzeneteket, és nem ad kimenetet, nincs szükség típusparaméterre:

class LogExecutor(Executor):

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

Terminál- és köztes kimeneti végrehajtók tervezése

Az, hogy mely végrehajtók járulnak hozzá a munkafolyamat végső válaszához, és melyek küldenek megfigyelési állapotfrissítéseket, egy build időben meghozott, a(z) WorkflowBuilder elemen konfigurált döntés, nem pedig kibocsátásonkénti jelző.

• output_from — azok a végrehajtók, amelyek ctx.yield_output(...) hívásai "output" eseményeket hoznak létre, és amelyeket a WorkflowRunResult.get_outputs() ad vissza.
• intermediate_output_from — azok a végrehajtók, amelyek ctx.yield_output(...) hívásai "intermediate" eseményeket hoznak létre, és amelyeket a WorkflowRunResult.get_intermediate_outputs() ad vissza.

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(...) nem rendelkezik kibocsátásonkénti jelölővel. Ugyanaz a hívás "output" vagy "intermediate" címkével van megjelölve, kizárólag az építő megnevezése alapján. Nincs API – a ctx.yield_intermediate(...) kijelölés nem változik hozamonként.

Mindkét lista nem kötelező. Ha bármelyik kimeneti kijelölési lista meg van adva, a egyik listában sem szereplő végrehajtó továbbra is küldhet üzeneteket az alsóbb rétegbeli végrehajtóknak ctx.send_message(...), de a yield_output hívások rejtettek. Ha mindkét lista el van hagyva, minden yield_output továbbra is kibocsátja a "output" elemet a kompatibilitás érdekében.

Egyszerű végrehajtói struktúra

A végrehajtók a munkafolyamat feldolgozási egységei. Bemenetet kapnak, munkát végeznek és kimenetet hoznak létre.

Több bemeneti típus

Több kezelő regisztrálása egy végrehajtó útvonalainak konfigurálásával:

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

Funkcióalapú végrehajtók

A végrehajtó objektum létrehozásának legegyszerűbb módja a workflow.NewExecutor(...).Bind() használatával történik:

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

A függvény-végrehajtók automatikusan regisztrálják a bemeneti típust, és automatikusan elküldhetik és automatikusan visszaadhatják a visszaadott értékeket.

A munkafolyamat. Környezeti objektum

A kezelők a végrehajtás során elfogadhatják *workflow.Context a munkafolyamattal való interakciót:

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

A környezet olyan API-kat is elérhetővé tesz, mint a SendMessage, AddEvent, PostRequest, ReadStateés QueueStateUpdate.

Ügynök-végrehajtók

Az ügynökök a következő módon agentworkflow.Newhasználhatók munkafolyamat-végrehajtóként:

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

Végrehajtó életciklusa

A végrehajtók a következő mezőkön keresztül támogatják az workflow.Executoréletciklus-horgokat:

Horog	Alkalmazás célja
ConfigureProtocol	Üzenet-útválasztás és deklarált küldési/hozamtípusok beállítása
InitializeFunc	Beállítás, amikor egy végrehajtó példány létrejön egy futtatáshoz
ResetFunc	A végrehajtó helyi állapotának visszaállítása újbóli felhasználás előtt
OnCheckpointFunc	Állapot mentése az ellenőrzőponton
OnCheckpointRestoredFunc	Állapot visszaállítása ellenőrzőpontról
OnMessageDeliveryStartingFunc	Fusson le, mielőtt egy szuperlépés kézbesíti az üzeneteket
OnMessageDeliveryFinishedFunc	Futtatás egy szuperstep üzenetkézbesítésének befejeződése után

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

Következő lépések

edges