Testamentsexekutorer

Utförare är de grundläggande byggstenarna som bearbetar meddelanden i ett arbetsflöde. De är autonoma bearbetningsenheter som tar emot inskrivna meddelanden, utför åtgärder och kan generera utdatameddelanden eller händelser.

Overview

Varje exekutor har en unik identifikator och kan hantera särskilda meddelandetyper. Utförare kan vara:

• Anpassade logikkomponenter – bearbeta data, anropa API:er eller transformera meddelanden
• AI-agenter – använd LLM:er för att generera svar (se Agenter i arbetsflöden)

Important

Det rekommenderade sättet att definiera körmeddelandehanterare i C# är att använda [MessageHandler] attributet på metoder i en partial klass som härleds från Executor. Detta använder generering av källkod vid kompileringstid för hanteringsregistrering, vilket ger bättre prestanda, kompileringstidsverifiering och native AOT-kompatibilitet.

Grundläggande körstruktur

Körarna härleds från basklassen Executor och använder [MessageHandler]-attributet för att deklarera hanteringsmetoder. Klassen måste markeras partial för att aktivera källgenerering.

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

Du kan också skicka meddelanden manuellt utan att returnera ett värde:

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

Tips/Råd

Körbara filer kan ha ett föränderligt tillstånd. Om en tillståndsorienterad exekutor delas mellan olika körningar av arbetsflöden, måste den implementera IResettableExecutor för att rensa inaktuellt tillstånd mellan körningarna. Se Återställningsbara exekverare för mer information.

Flera indatatyper

Hantera flera indatatyper genom att definiera flera [MessageHandler] metoder:

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

Funktionsbaserade exekverare

Skapa en exekverare från en funktion med hjälp av BindExecutor tilläggsmetoden.

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

IWorkflowContext-objektet

Tillhandahåller IWorkflowContext metoder för att interagera med arbetsflödet under körningen:

• SendMessageAsync — skicka meddelanden till anslutna köre
• YieldOutputAsync — skapa arbetsflödesutdata som returneras/strömmas till anroparen

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

Om en hanterare varken skickar meddelanden eller ger utdata kan den helt enkelt utföra biverkningar:

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

Deklarera protokolltyper

En exekutors protokoll deklarerar de meddelandetyper som den kan skicka till anslutna utförare och de utdatatyper som det kan ge. Arbetsflödet validerar anrop till SendMessageAsync och YieldOutputAsync mot dessa deklarationer och genererar en InvalidOperationException när en köre använder en odeklarerad typ.

Använd [SendsMessage] för att deklarera skickade meddelandetyper och [YieldsOutput] för att deklarera utdatatyper som returneras. Dessa attribut beskriver utförarens funktioner. de inte själva skickar eller ger värden. Använd varje attribut flera gånger när kören använder flera typer.

För exekutorer med en enda typhanterad hanterare härleder du från Executor<TInput> eller Executor<TInput, TOutput> och åsidosätter 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);
    }
}

När arbetsflödets källgenerator refereras måste en klass med [SendsMessage] eller [YieldsOutput] deklareras partial så att generatorn kan lägga till sin protokollkonfiguration.

För källgenererade utförare med [MessageHandler] metoder deklarerar du typer som används av en hanterare med dess Send och Yield namngivna argument, till exempel [MessageHandler(Send = [typeof(ProgressUpdate)], Yield = [typeof(string)])]. Använd klassnivå [SendsMessage] och [YieldsOutput] när deklarationerna gäller för hela kören.

Returtyper för icke-void-hanterare läggs automatiskt till i de protokolltyper som skickas och returneras när ExecutorOptions.AutoSendMessageHandlerResultObject och ExecutorOptions.AutoYieldOutputHandlerResultObject aktiveras. Båda alternativen är aktiverade som standard. Explicita deklarationer behövs därför främst för ytterligare typer som genereras direkt via SendMessageAsync eller YieldOutputAsync.

[YieldsOutput] tillåter att utföraren ger en typ, men den anger inte exekutorn som en terminalutdatakälla. Registrera kören med WorkflowBuilder.WithOutputFrom för att dess värden som returneras ska visas för arbetsflödesanroparen.

Grundläggande körstruktur

Utförare ärver från basklassen Executor . Varje exekutor använder metoder som är dekorerade med dekoratören @handler . Hanterarna måste ha rätt typanteckningar för att kunna ange de meddelandetyper som de bearbetar.

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

Funktionsbaserade exekverare

Skapa en exekutor från en funktion med hjälp av dekoratören @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())

Flera indatatyper

Hantera flera indatatyper genom att definiera flera hanterare:

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)

Explicita typparametrar

Som ett alternativ till att skriva anteckningar kan du uttryckligen ange typer via dekoratörsparametrar:

Important

När du använder explicita typparametrar måste du ange alla typer via dekoratören – du kan inte blanda explicita parametrar med typanteckningar. Parametern input krävs output och workflow_output är valfri.

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)

WorkflowContext-objektet

Tillhandahåller WorkflowContext metoder för att interagera med arbetsflödet under körningen:

• send_message — skicka meddelanden till anslutna köre
• yield_output — skapa arbetsflödesutdata som returneras/strömmas till anroparen

class OutputExecutor(Executor):

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

Om en hanterare varken skickar meddelanden eller ger utdata behövs ingen typparameter:

class LogExecutor(Executor):

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

Utse terminal- och mellanliggande utdataexekutorer

Vilka exekverare som bidrar till arbetsflödets slutliga svar och vilka som avger observationsrelaterad förloppsinformation är ett beslut som fattas vid byggtid och konfigureras i , inte en flagga för varje utsändning.

• output_from — utförare vars ctx.yield_output(...) anrop genererar "output" händelser och returneras av WorkflowRunResult.get_outputs().
• intermediate_output_from — utförare vars ctx.yield_output(...) anrop genererar "intermediate" händelser och returneras av 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(...) har ingen flagga för varje utsläpp. Samma anrop är märkt "output" eller "intermediate" enbart baserat på byggarens beteckning. Det finns inget ctx.yield_intermediate(...) API – beteckningen varierar inte per avkastning.

Båda listorna är valfria. Om någon av de två utdataurvalslistorna anges kan en exekverare som inte förekommer i någon av listorna fortfarande skicka meddelanden till efterföljande exekverare via ctx.send_message(...), men dess anrop till yield_output döljs. Om båda listorna utelämnas genererar varje yield_output fortfarande "output" för kompatibilitet.

Grundläggande körstruktur

Utförare är bearbetningsenheterna i ett arbetsflöde. De tar emot indata, utför arbete och producerar utdata.

Flera indatatyper

Registrera flera hanterare genom att konfigurera rutter på en exekverare:

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

Funktionsbaserade exekverare

Det enklaste sättet att skapa en executor är med hjälp av workflow.NewExecutor(...).Bind():

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

Funktionsexekutorer registrerar automatiskt indatatypen och kan automatiskt skicka och returnera returnerade värden automatiskt.

Arbetsflödet. Kontextobjekt

Hanterare kan ta emot *workflow.Context för att interagera med arbetsflödet under körning:

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

Kontexten exponerar även API:er som SendMessage, AddEvent, PostRequest, ReadStateoch QueueStateUpdate.

Agentexekutorer

Agenter kan användas som arbetsflödesexekutorer via agentworkflow.New:

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

Exekverarens livscykel

Executorer har stöd för livscykelkrokar via fält i workflow.Executor:

Krok	Purpose
ConfigureProtocol	Konfigurera meddelanderoutning och deklarerade typer av send/yield
InitializeFunc	Konfigurera när en körinstans skapas för en körning
ResetFunc	Återställ exekverarens lokala tillstånd före återanvändning
OnCheckpointFunc	Spara tillstånd vid kontrollpunkt
OnCheckpointRestoredFunc	Återställa tillstånd från kontrollpunkt
OnMessageDeliveryStartingFunc	Körs innan ett supersteg levererar meddelanden
OnMessageDeliveryFinishedFunc	Kör när ett supersteg har slutfört meddelandeleveransen

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

Nästa steg

kanter