Исполнители

Исполнителями являются основные стандартные блоки, обрабатывающие сообщения в рабочем процессе. Они являются автономными единицами обработки, которые получают типизированные сообщения, выполняют операции и могут создавать выходные сообщения или события.

Обзор

Каждый исполнитель имеет уникальный идентификатор и может обрабатывать определенные типы сообщений. Исполнителями могут быть:

Пользовательские компоненты логики — обработка данных, вызов API или преобразование сообщений
Агенты ИИ — используйте LLM для создания ответов (см. раздел "Агенты в рабочих процессах")

Это важно

Рекомендуемый способ определения обработчиков сообщений исполнителя в C# — использовать [MessageHandler] атрибут для методов в классе, наследуемом partial от Executor. В этом случае используется генерация исходного кода во время компиляции для регистрации обработчиков, что обеспечивает лучшую производительность, проверку на этапе компиляции и совместимость с Native 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
    }
}

Подсказка

Исполнители могут хранить изменяемое состояние. Если исполнитель с отслеживанием состояния используется в нескольких запусках рабочего процесса, он должен реализовать IResettableExecutor для очистки устаревшего состояния между запусками. Дополнительные сведения см. в разделе "Сбрасываемые исполнители".

Несколько типов входных данных

Обработка нескольких типов входных данных путем определения нескольких [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");

Объект IWorkflowContext

Эта IWorkflowContext предоставляет методы для взаимодействия с рабочим процессом в ходе выполнения.

SendMessageAsync — отправка сообщений подключенным исполнителям
YieldOutputAsync — создание результатов рабочего процесса, возвращаемых или передаваемых потоково вызывающему

internal sealed partial class OutputExecutor() : Executor("OutputExecutor")
{
    [MessageHandler]
    private async ValueTask HandleAsync(string message, IWorkflowContext context)
    {
        await context.YieldOutputAsync("Hello, World!");
    }
}

Если обработчик не отправляет сообщения и не выдает выходные данные, он может просто выполнять побочные эффекты:

internal sealed partial class LogExecutor() : Executor("LogExecutor")
{
    [MessageHandler]
    private void Handle(string message, IWorkflowContext context)
    {
        Console.WriteLine("Doing some work...");
    }
}

Базовая структура исполнителя

Исполнители наследуются от 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."""
        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...")

Назначение исполнителей конечного и промежуточного вывода

Какие исполнители вносят вклад в итоговый ответ рабочего процесса, а какие выдают информацию о ходе выполнения, — это решение, принимаемое во время сборки, которое настраивается в WorkflowBuilder, а не флаг для каждой отдельной эмиссии.

output_from — исполнители, вызовы ctx.yield_output(...) которых генерируют события "output" и которые возвращаются WorkflowRunResult.get_outputs().
intermediate_output_from — исполнители, вызовы ctx.yield_output(...) которых генерируют события "intermediate" и которые возвращаются 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()

Это важно

ctx.yield_output(...) не имеет флага на выбросы. Один и тот же вызов помечается как "output" или "intermediate" исключительно в зависимости от обозначения построителя. Нет ctx.yield_intermediate(...) API — обозначение не меняется в зависимости от доходности.

Оба списка являются необязательными. Если указан любой из списков выбора выходных данных, исполнитель, который не входит ни в один из списков, всё равно может отправлять сообщения нижестоящим исполнителям через ctx.send_message(...), но его вызовы yield_output скрыты. Если оба списка опущены, каждый yield_output по-прежнему выводит "output" для совместимости.

Базовая структура исполнителя

Исполнителями являются единицы обработки в рабочем процессе. Они получают входные данные, выполняют работу и создают выходные данные.

Несколько типов входных данных

Зарегистрируйте несколько обработчиков, настроив маршруты для исполнителя:

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

Функционально-ориентированные исполнители

Самый простой способ создать исполнителя — с помощью workflow.NewExecutor(...).Bind():

uppercase := workflow.NewExecutor("UppercaseExecutor", func(input string) string {
    return strings.ToUpper(input)
}).Bind()

Исполнители функций автоматически регистрируют входной тип и могут автоматически отправлять и автоматически возвращать возвращаемые значения.

Рабочий процесс. Объект Context

Обработчики могут принимать *workflow.Context для взаимодействия с рабочим процессом во время выполнения:

output := workflow.NewExecutor("OutputExecutor", func(ctx *workflow.Context, message string) error {
    return ctx.YieldOutput("Hello, World!")
}).Bind()

Контекст также предоставляет такие API, как SendMessage, AddEvent, PostRequest, ReadStateи QueueStateUpdate.