Exekutoři

Základní struktura vykonavatele

Exekutory jsou odvozeny ze Executor základní třídy a používají [MessageHandler] atribut k deklaraci metod obslužné rutiny. Aby bylo možné povolit generování zdroje, musí být třída označena partial .

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

Zprávy můžete odesílat také ručně bez vrácení hodnoty:

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

Více vstupních typů

Zpracování více vstupních typů definováním více [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);
    }
}

Vykonavatelé založené na funkcích

Vytvořte spouštěč z funkce pomocí metody rozšíření BindExecutor.

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

Objekt IWorkflowContext

Tento IWorkflowContext poskytuje metody, které umožňují interakci s pracovním postupem během jeho provádění:

• SendMessageAsync — odesílání zpráv připojeným exekutorům
• YieldOutputAsync — vytváří výstupy pracovního postupu vrácené nebo streamované volajícímu.

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

Pokud obslužná rutina neodesílá zprávy ani nevyvolá výstupy, může jednoduše provádět vedlejší účinky:

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

Základní struktura vykonavatele

Exekutoři dědí z Executor základní třídy. Každý executor používá metody, které jsou dekorovány dekorátorem @handler. Obslužné rutiny musí mít správné anotace typu, aby určily typy zpráv, které zpracovávají.

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

Vykonavatelé založené na funkcích

Vytvořte vykonavatele z funkce pomocí dekorátoru @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())

Více vstupních typů

Zpracování více vstupních typů definováním více obslužných rutin:

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)

Explicitní parametry typu

Jako alternativu k psaní poznámek můžete zadat typy explicitně prostřednictvím parametrů dekorátoru:

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)

Objekt WorkflowContext

Tento WorkflowContext poskytuje metody, které umožňují interakci s pracovním postupem během jeho provádění:

• send_message — odesílání zpráv připojeným exekutorům
• yield_output — vytváří výstupy pracovního postupu vrácené nebo streamované volajícímu.

class OutputExecutor(Executor):

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

Pokud obslužná rutina neodesílá zprávy ani nedává výstupy, není potřeba žádný parametr typu:

class LogExecutor(Executor):

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

Navrhování terminálových a průběžných výstupních exekutorů

Které executory přispívají ke konečné odpovědi workflow a které vysílají průběžné informace o stavu, se rozhoduje při sestavení a konfiguruje se na WorkflowBuilder, nikoli pomocí příznaku nastavovaného pro každé jednotlivé vyslání.

• output_from — executory, jejichž volání ctx.yield_output(...) generují události "output" a které vrací WorkflowRunResult.get_outputs().
• intermediate_output_from — executory, jejichž volání ctx.yield_output(...) generují události "intermediate" a které vrací 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()

Oba seznamy jsou volitelné. Pokud je k dispozici alespoň jeden ze seznamů výběru výstupu, exekutor, který se nenachází ani v jednom seznamu, může stále odesílat zprávy následným exekutorům prostřednictvím ctx.send_message(...), ale jeho volání yield_output jsou skrytá. Pokud nejsou uvedeny oba seznamy, každý yield_output stále vypisuje "output" kvůli kompatibilitě.

Základní struktura vykonavatele

Exekutory jsou jednotky zpracování v pracovním postupu. Přijímají vstup, provádějí práci a vytvářejí výstup.

Více vstupních typů

Zaregistrujte více obslužných rutin nakonfigurováním tras v executoru:

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

Vykonavatelé založené na funkcích

Nejjednodušší způsob, jak vytvořit executor, je pomocí workflow.NewExecutor(...).Bind():

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

Exekutory funkcí automaticky registrují vstupní typ a můžou automaticky odesílat a automaticky poskytovat vrácené hodnoty.

Pracovní postup. Kontextový objekt

Obslužné rutiny mohou přijímat *workflow.Context, aby mohly během vykonávání interagovat s workflow:

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

Kontext také zpřístupňuje rozhraní API, například SendMessage, AddEvent, PostRequest, ReadState a QueueStateUpdate.

Exekutory agentů

Agenty lze použít jako exekutory pracovního postupu prostřednictvím agentworkflow.New:

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

Životní cyklus exekutoru

Exekutory podporují háčky životního cyklu prostřednictvím polí:workflow.Executor

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