Základní koncepty pracovních postupů Microsoft Agent – vykonavatelé

Tento dokument obsahuje podrobný pohled na komponentu Exekutory systému pracovních postupů rozhraní Microsoft Agent Framework.

Přehled

Exekutory jsou základní stavební bloky, které zpracovávají zprávy v pracovním postupu. Jedná se o autonomní jednotky zpracování, které přijímají typové zprávy, provádějí operace a můžou vytvářet výstupní zprávy nebo události.

Exekutory implementují rozhraní IMessageHandler<TInput> nebo IMessageHandler<TInput, TOutput> a dědí ze základní třídy ReflectingExecutor<T>. Každý exekutor má jedinečný identifikátor a může zpracovávat konkrétní typy zpráv.

Základní struktura vykonavatele

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

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

Zprávy je možné odesílat ručně bez vrácení hodnoty:

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

Je také možné zpracovat více vstupních typů implementací více rozhraní:

internal sealed class SampleExecutor() : ReflectingExecutor<SampleExecutor>("SampleExecutor"),
    IMessageHandler<string, string>, IMessageHandler<int, int>
{
    /// <summary>
    /// Converts input string to uppercase
    /// </summary>
    public async ValueTask<string> HandleAsync(string message, IWorkflowContext context)
    {
        string result = message.ToUpperInvariant();
        return result;
    }

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

Exekutoři dědí z Executor základní třídy. Každý vykonavatel má jedinečný identifikátor a může zpracovávat konkrétní typy zpráv pomocí metod, na které je aplikován dekorátor @handler. Obslužné rutiny musí mít správnou anotaci, aby bylo možné určit typ zpráv, které mohou zpracovávat.

Základní struktura vykonavatele

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

Exekutor je možné vytvořit 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.

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

Je také možné zpracovat více typů vstupu 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:
        """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)

Objekt WorkflowContext

Objekt WorkflowContext poskytuje metody pro handler interagující s pracovním postupem během provádění. Parametrizuje se WorkflowContext s typem zpráv, které obslužná rutina vysílá, a typem výstupů, které může poskytnout.

Nejčastěji používanou metodou je send_message, která obslužné rutině umožňuje odesílat zprávy připojeným exekutorům.

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!")

Obslužná rutina může použít yield_output k vytvoření výstupů, které se budou považovat za výstupy pracovního postupu a budou vráceny/streamovány volajícímu jako výstupní událost:

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!")

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

from agent_framework import WorkflowContext

class SomeHandler(Executor):

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

Další krok

• Zjistěte více o hranách, abyste pochopili, jak jsou vykonavatele propojeni v pracovním postupu.