Ejecutores

Estructura básica del ejecutor

Los ejecutores derivan de la Executor clase base y usan el [MessageHandler] atributo para declarar métodos de controlador. La clase debe marcarse partial para habilitar la generación de código fuente.

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

También puede enviar mensajes manualmente sin devolver un valor:

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

Varios tipos de entrada

Controle varios tipos de entrada mediante la definición de varios [MessageHandler] métodos:

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

Ejecutores basados en funciones

Cree un ejecutor a partir de una función mediante el método de extensión BindExecutor.

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

El objeto IWorkflowContext

IWorkflowContext proporciona métodos para interactuar con el flujo de trabajo durante la ejecución:

• SendMessageAsync : enviar mensajes a ejecutores conectados
• YieldOutputAsync : genera salidas de flujo de trabajo devueltas o transmitidas al autor de la llamada.

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

Si un controlador no envía mensajes ni produce salidas, simplemente puede realizar efectos secundarios:

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

Estructura básica del ejecutor

Los ejecutores heredan de la Executor clase base. Cada ejecutor usa métodos decorados con el decorador @handler. Los controladores deben contar con anotaciones tipográficas correctas para especificar los tipos de mensaje que procesan.

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

Ejecutores basados en funciones

Cree un ejecutor a partir de una función mediante el @executor decorador:

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

Varios tipos de entrada

Controle varios tipos de entrada mediante la definición de varios controladores:

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)

Parámetros de tipo explícitos

Como alternativa a las anotaciones de tipo, puede especificar tipos explícitamente a través de parámetros de decorador:

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)

El objeto WorkflowContext

WorkflowContext proporciona métodos para interactuar con el flujo de trabajo durante la ejecución:

• send_message : enviar mensajes a ejecutores conectados
• yield_output : genera salidas de flujo de trabajo devueltas o transmitidas al autor de la llamada.

class OutputExecutor(Executor):

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

Si un controlador no envía mensajes ni genera salidas, no se necesita ningún parámetro de tipo:

class LogExecutor(Executor):

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

Designación de ejecutores de salida terminales e intermedios

Qué ejecutores contribuyen a la respuesta del terminal del flujo de trabajo y cuáles emiten progreso observacional es una decisión de tiempo de compilación configurada en WorkflowBuilder, no una marca por emisión.

• output_from: ejecutores cuyas llamadas ctx.yield_output(...) producen eventos "output" y los devuelve WorkflowRunResult.get_outputs().
• intermediate_output_from: ejecutores cuyas llamadas ctx.yield_output(...) producen eventos "intermediate" y los devuelve 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()

Ambas listas son opcionales. Si se proporciona cualquiera de las dos listas de selección de salida, un ejecutor que no aparece en ninguna de ellas aún puede enviar mensajes a ejecutores posteriores a través de ctx.send_message(...), pero sus llamadas a yield_output quedan ocultas. Si se omiten ambas listas, todas yield_output siguen emitiendo "output" por compatibilidad.

Estructura básica del ejecutor

Los ejecutores son las unidades de procesamiento de un flujo de trabajo. Reciben entradas, realizan trabajo y generan resultados.

Varios tipos de entrada

Registre varios controladores mediante la configuración de rutas en un ejecutor:

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

Ejecutores basados en funciones

La manera más sencilla de crear un ejecutor es con workflow.NewExecutor(...).Bind():

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

Los ejecutores de función registran automáticamente el tipo de entrada y pueden enviar automáticamente y devolver valores devueltos automáticamente.

El objeto workflow.Context

Los controladores pueden aceptar *workflow.Context para interactuar con el flujo de trabajo durante la ejecución:

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

El contexto también expone las API como SendMessage, AddEvent, PostRequest, ReadStatey QueueStateUpdate.

Ejecutores de agentes

Los agentes se pueden usar como ejecutores de flujo de trabajo a través de agentworkflow.New:

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

Ciclo de vida del ejecutor

Los ejecutores admiten enlaces de ciclo de vida a través de campos en 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()