通过

执行器

执行程序是处理工作流中消息的基本构建基块。 它们是接收类型化消息、执行作并可以生成输出消息或事件的自治处理单元。

概述

每个执行程序都有一个唯一标识符，可以处理特定的消息类型。 执行者可以是：

  • 自定义逻辑组件 - 处理数据、调用 API 或转换消息
  • AI 代理 - 使用 LLM 生成响应（请参阅 工作流中的代理）

重要

在 C# 中，定义执行器消息处理程序的推荐方法是在派生自Executor的类中的方法上使用[MessageHandler]特性。 这使用编译时源生成进行处理程序注册，从而提供更好的性能、编译时验证和本机 AOT 兼容性。

基本执行器结构

执行程序派生自 Executor 基类，并使用 [MessageHandler] 特性声明处理程序方法。 必须标记 partial 该类才能启用源生成。

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

还可以在不返回值的情况下手动发送消息：

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

多个输入类型

通过定义多个方法来处理多个 [MessageHandler] 输入类型：

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

基于函数的执行器

使用 BindExecutor 扩展方法从函数创建执行程序：

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

基本执行器结构

执行程序继承自 Executor 基类。 每个执行程序都使用< c0 />装饰器修饰的方法。 处理程序必须具有正确的类型注解，以便明确它们所处理的消息类型。

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

基于函数的执行程序

使用 @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."""
    await ctx.send_message(text.upper())

多个输入类型

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

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)

显式类型参数

作为类型批注的替代方法，可以通过修饰器参数显式指定类型：

重要

使用显式类型参数时，必须通过修饰器指定 所有 类型 — 不能将显式参数与类型批注混合。 参数 input 是必需的; output 并且 workflow_output 是可选的。

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)

WorkflowContext 对象

WorkflowContext 提供在执行过程中与工作流交互的方法：

  • send_message — 将消息发送到连接的执行程序
  • yield_output — 生成返回/流式传输到调用方工作流输出
class OutputExecutor(Executor):

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

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

class LogExecutor(Executor):

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

后续步骤

边缘

  • Last updated on