Ejecutores

Los ejecutores son los componentes fundamentales que procesan mensajes en un flujo de trabajo. Son unidades de procesamiento autónomas que reciben mensajes tipados, realizan operaciones y pueden generar mensajes o eventos de salida.

Overview

Cada ejecutor tiene un identificador único y puede controlar tipos de mensajes específicos. Los ejecutores pueden ser:

• Componentes lógicos personalizados : procesar datos, llamar a las API o transformar mensajes
• Agentes de IA — use LLMs para generar respuestas (consulte Agentes en Flujos de Trabajo)

Importante

La manera recomendada de definir controladores de mensajes del ejecutor en C# es usar el [MessageHandler] atributo en métodos dentro de una partial clase que deriva de Executor. Esto usa la generación de código fuente en tiempo de compilación para el registro del manejador, lo que proporciona un mejor rendimiento, validación en tiempo de compilación y compatibilidad con AOT nativo.

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

Tip

Los ejecutores pueden mantener un estado mutable. Si un ejecutor con estado se comparte entre ejecuciones del flujo de trabajo, debe implementar IResettableExecutor para borrar el estado obsoleto entre ejecuciones. Consulte Resettable Executors (Ejecutores Reiniciables) para obtener más información.

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

Declaración de tipos de protocolo

El protocolo de un ejecutor declara los tipos de mensaje que puede enviar a los ejecutores conectados y los tipos de salida que puede producir. El flujo de trabajo valida las llamadas a SendMessageAsync estas declaraciones y YieldOutputAsync produce cuando InvalidOperationException un ejecutor usa un tipo no declarado.

Use [SendsMessage] para declarar tipos de mensajes enviados y [YieldsOutput] para declarar tipos de salida devueltos. Estos atributos describen las funcionalidades del ejecutor; no envían ni producen valores por sí mismos. Aplique cada atributo varias veces cuando el ejecutor use varios tipos.

Para ejecutores con un único controlador con tipo, derive de Executor<TInput> o Executor<TInput, TOutput> e invalide 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);
    }
}

Cuando se hace referencia al generador de origen de flujos de trabajo, se debe declarar partial una clase con [SendsMessage] o [YieldsOutput] para que el generador pueda agregar su configuración de protocolo.

Para los ejecutores generados por el origen con [MessageHandler] métodos, declare los tipos usados por un controlador con sus argumentos con nombre Send y Yield , como [MessageHandler(Send = [typeof(ProgressUpdate)], Yield = [typeof(string)])]. Use el nivel [SendsMessage] de clase y [YieldsOutput] cuando las declaraciones se apliquen a todo el ejecutor.

Los tipos de valor devueltos de controlador no void se agregan automáticamente a los tipos de protocolo enviados y devueltos cuando ExecutorOptions.AutoSendMessageHandlerResultObject y ExecutorOptions.AutoYieldOutputHandlerResultObject están habilitados. Ambas opciones están habilitadas de forma predeterminada. Por lo tanto, las declaraciones explícitas son necesarias principalmente para tipos adicionales emitidos directamente a través SendMessageAsync de o YieldOutputAsync.

[YieldsOutput] permite que el ejecutor produzca un tipo, pero no designa al ejecutor como origen de salida de terminal. Registre el ejecutor con WorkflowBuilder.WithOutputFrom para que sus valores devueltos se muestren al autor de la llamada del flujo de trabajo.

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:

Importante

Al usar parámetros de tipo explícitos, debe especificar todos los tipos a través del decorador; no se pueden mezclar parámetros explícitos con anotaciones de tipo. El input parámetro es obligatorio; output y workflow_output son opcionales.

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

Importante

ctx.yield_output(...) no tiene indicador por emisión. La misma llamada se etiqueta como "output" o "intermediate" únicamente en función de la designación del generador. No hay ninguna ctx.yield_intermediate(...) API: la designación no varía según el rendimiento.

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:

Enlace	Purpose
ConfigureProtocol	Configuración del enrutamiento de mensajes y tipos de envío/rendimiento declarados
InitializeFunc	Instalación cuando se crea una instancia del ejecutor para una ejecución
ResetFunc	Restablecer el estado local del ejecutor antes de volver a usar
OnCheckpointFunc	Guardar el estado en el punto de control
OnCheckpointRestoredFunc	Restaurar el estado desde el punto de control
OnMessageDeliveryStartingFunc	Ejecutar antes de que un superpaso envíe mensajes
OnMessageDeliveryFinishedFunc	Ejecutar después de que un superpaso finalice la entrega de mensajes

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

Pasos siguientes

Bordes