Microsoft Agent Framework 工作流核心概念 - 执行程序

执行程序实现 IMessageHandler<TInput> 或 IMessageHandler<TInput, TOutput> 接口并从基类继承 ReflectingExecutor<T> 。 每个执行程序都有一个唯一标识符，可以处理特定的消息类型。

基本执行器结构

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

无需返回值即可手动发送消息：

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

还可以通过实现多个接口来处理多个输入类型：

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

执行程序继承自 Executor 基类。 每个执行器都有一个唯一的标识符，并且可以通过使用 @handler 修饰器修饰的方法来处理特定的消息类型。 处理者必须具有正确的注解才能指定它们能够处理的消息类型。

基本执行器结构

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

可以使用修饰器从函数 @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())

还可以通过定义多个处理程序来处理多个输入类型：

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)

对象WorkflowContext

该 WorkflowContext 对象为处理程序提供在执行过程中与工作流交互的方法。 WorkflowContext 被参数化为处理程序将发出的消息类型和它可以产生的输出类型。

最常用的方法是 send_message，它允许处理程序将消息发送到连接的执行程序。

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

处理程序可用于 yield_output 生成将被视为工作流输出的输出，并将输出作为输出事件返回/流式传输到调用方：

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

如果处理程序既不发送消息也不生成输出，则不需要以下类型 WorkflowContext参数：

from agent_framework import WorkflowContext

class SomeHandler(Executor):

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