Исполнители

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

Overview

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

Пользовательские компоненты логики — обработка данных, вызов 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
    }
}

Tip

Исполнители могут хранить изменяемое состояние. Если исполнитель с отслеживанием состояния используется в нескольких запусках рабочего процесса, он должен реализовать 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...");
    }
}

Объявление типов протоколов

Протокол исполнителя объявляет типы сообщений, которые он может отправлять подключенным исполнителям, и выходные типы, которые он может дать. Рабочий процесс проверяет вызовы к YieldOutputAsync и SendMessageAsync на соответствие этим объявлениям и вызывает исключение InvalidOperationException, когда исполнитель использует необъявленный тип.

Используется [SendsMessage] для объявления типов отправленных сообщений и [YieldsOutput] объявления типов выходных данных. Эти атрибуты описывают возможности исполнителя; сами по себе они не отправляют и не возвращают значения. Примените каждый атрибут несколько раз, когда исполнитель использует несколько типов.

Для исполнителей с одним типизированным обработчиком унаследуйтесь от Executor<TInput> или Executor<TInput, TOutput> и переопределите HandleAsync:

internal sealed record ProcessRequest(string Text);
internal sealed record ProgressUpdate(string Status);

[SendsMessage(typeof(ProgressUpdate))]
[YieldsOutput(typeof(string))]
internal sealed partial class ProcessingExecutor()
    : Executor<ProcessRequest>("ProcessingExecutor")
{
    public override async ValueTask HandleAsync(
        ProcessRequest message,
        IWorkflowContext context,
        CancellationToken cancellationToken = default)
    {
        await context.SendMessageAsync(
            new ProgressUpdate("Processing started"),
            cancellationToken);

        await context.YieldOutputAsync(
            message.Text.ToUpperInvariant(),
            cancellationToken);
    }
}

Если используется генератор исходного кода workflows, класс с [SendsMessage] или [YieldsOutput] должен быть объявлен как partial, чтобы генератор мог добавить его конфигурацию протокола.

Для исполнителей, создаваемых генератором исходного кода, с методами [MessageHandler] объявляйте типы, используемые одним обработчиком, с помощью именованных аргументов Send и Yield, например [MessageHandler(Send = [typeof(ProgressUpdate)], Yield = [typeof(string)])]. Используйте [SendsMessage] и [YieldsOutput] на уровне класса, если объявления относятся ко всему исполнителю.

Типы возвращаемых значений обработчиков, отличные от void, автоматически добавляются к типам протокола sent и yielded, когда включены ExecutorOptions.AutoSendMessageHandlerResultObject и ExecutorOptions.AutoYieldOutputHandlerResultObject. Оба параметра включены по умолчанию. Поэтому явные объявления в основном необходимы для дополнительных типов, генерируемых непосредственно с помощью SendMessageAsync или YieldOutputAsync.

[YieldsOutput] позволяет исполнителю давать тип, но он не назначает исполнителя в качестве источника выходных данных терминала. Зарегистрируйте исполнитель с WorkflowBuilder.WithOutputFrom, чтобы его возвращаемые значения были доступны вызывающей стороне рабочего процесса.

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

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