Concepts fondamentaux des flux de travail Microsoft Agent Framework - Exécuteurs

Ce document fournit un aperçu approfondi du composant Exécuteurs du système de flux de travail Microsoft Agent Framework.

Aperçu

Les exécuteurs sont les blocs de construction fondamentaux qui traitent les messages dans un flux de travail. Ils sont des unités de traitement autonomes qui reçoivent des messages typés, effectuent des opérations et peuvent produire des messages de sortie ou des événements.

Les exécuteurs héritent de la Executor<TInput, TOutput> classe de base. Chaque exécuteur a un identificateur unique et peut gérer des types de messages spécifiques.

Structure d’exécuteur de base

using Microsoft.Agents.AI.Workflows;
using Microsoft.Agents.AI.Workflows.Reflection;

internal sealed class UppercaseExecutor() : Executor<string, string>("UppercaseExecutor")
{
    public async ValueTask<string> HandleAsync(string message, IWorkflowContext context)
    {
        string result = message.ToUpperInvariant();
        return result; // Return value is automatically sent to connected executors
    }
}

Il est possible d’envoyer manuellement des messages sans retourner une valeur :

internal sealed class UppercaseExecutor() : Executor<string>("UppercaseExecutor")
{
    public async ValueTask HandleAsync(string message, IWorkflowContext context)
    {
        string result = message.ToUpperInvariant();
        await context.SendMessageAsync(result); // Manually send messages to connected executors
    }
}

Il est également possible de gérer plusieurs types d’entrée en remplaçant la ConfigureRoutes méthode :

internal sealed class SampleExecutor() : Executor("SampleExecutor")
{
    protected override RouteBuilder ConfigureRoutes(RouteBuilder routeBuilder)
    {
        return routeBuilder
            .AddHandler<string>(this.HandleStringAsync)
            .AddHandler<int>(this.HandleIntAsync);
    }

    /// <summary>
    /// Converts input string to uppercase
    /// </summary>
    public async ValueTask<string> HandleStringAsync(string message, IWorkflowContext context)
    {
        string result = message.ToUpperInvariant();
        return result;
    }

    /// <summary>
    /// Doubles the input integer
    /// </summary>
    public async ValueTask<int> HandleIntAsync(int message, IWorkflowContext context)
    {
        int result = message * 2;
        return result;
    }
}

Il est également possible de créer un exécuteur à partir d’une fonction à l’aide de la BindExecutor méthode d’extension :

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

Les exécuteurs héritent de la Executor classe de base. Chaque exécuteur a un identificateur unique et peut gérer des types de messages spécifiques à l’aide de méthodes décorées avec le @handler décorateur. Les gestionnaires doivent avoir l’annotation appropriée pour spécifier le type de messages qu’ils peuvent traiter.

Structure d’exécuteur de base

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.

        Note: The WorkflowContext is parameterized with the type this handler will
        emit. Here WorkflowContext[str] means downstream nodes should expect str.
        """
        await ctx.send_message(text.upper())

Il est possible de créer un exécuteur à partir d’une fonction à l’aide de l’élément @executor décoratif :

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.

    Note: The WorkflowContext is parameterized with the type this handler will
    emit. Here WorkflowContext[str] means downstream nodes should expect str.
    """
    await ctx.send_message(text.upper())

Il est également possible de gérer plusieurs types d’entrée en définissant plusieurs gestionnaires :

class SampleExecutor(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.

        Note: The WorkflowContext is parameterized with the type this handler will
        emit. Here WorkflowContext[str] means downstream nodes should expect str.
        """
        await ctx.send_message(text.upper())

    @handler
    async def double_integer(self, number: int, ctx: WorkflowContext[int]) -> None:
        """Double the input integer and forward it to the next node.

        Note: The WorkflowContext is parameterized with the type this handler will
        emit. Here WorkflowContext[int] means downstream nodes should expect int.
        """
        await ctx.send_message(number * 2)

Objet WorkflowContext

L’objet WorkflowContext fournit des méthodes pour que le gestionnaire interagisse avec le flux de travail pendant l’exécution. Le WorkflowContext paramètre est paramétré avec le type de messages que le gestionnaire émet et le type de sortie qu’il peut générer.

La méthode la plus couramment utilisée est send_message, qui permet au gestionnaire d’envoyer des messages à des exécuteurs connectés.

from agent_framework import WorkflowContext

class SomeHandler(Executor):

    @handler
    async def some_handler(message: str, ctx: WorkflowContext[str]) -> None:
        await ctx.send_message("Hello, World!")

Un gestionnaire peut utiliser yield_output pour produire des sorties qui seront considérées comme des sorties de flux de travail et renvoyées/diffusées en continu à l’appelant en tant qu’événement de sortie :

from agent_framework import WorkflowContext

class SomeHandler(Executor):

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

Si un gestionnaire n’envoie ni messages ni génère de sorties, aucun paramètre de type n’est nécessaire pour WorkflowContext:

from agent_framework import WorkflowContext

class SomeHandler(Executor):

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

Étape suivante

• Découvrez les edges pour comprendre comment les exécuteurs sont connectés dans un flux de travail.