Поделиться через

Создание простого последовательного рабочего процесса

В этом руководстве показано, как создать простой последовательный рабочий процесс с помощью рабочих процессов Agent Framework.

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

Обзор

В этом руководстве вы создадите рабочий процесс с двумя исполнителями:

  1. Исполнитель верхнего регистра — преобразует входной текст в верхний регистр
  2. Процедура переворота текста — переворачивает текст и выводит окончательный результат.

Рабочий процесс демонстрирует основные понятия, такие как:

  • Создание пользовательского исполнителя с одним обработчиком
  • Создание пользовательского исполнителя из функции
  • Использование WorkflowBuilder для подключения исполнителей к ребрам
  • Обработка данных с помощью последовательных шагов
  • Наблюдение за выполнением рабочего процесса с помощью событий

Основные понятия, описанные в статье

Предпосылки

Пошаговая реализация

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

Шаг 1. Установка пакетов NuGet

Сначала установите необходимые пакеты для проекта .NET:

dotnet add package Microsoft.Agents.AI.Workflows --prerelease

Шаг 2. Определение исполнителя верхнего регистра

Определите исполнителя, который преобразует текст в верхний регистр:

using System;
using System.Linq;
using System.Threading.Tasks;
using Microsoft.Agents.AI.Workflows;

/// <summary>
/// First executor: converts input text to uppercase.
/// </summary>
Func<string, string> uppercaseFunc = s => s.ToUpperInvariant();
var uppercase = uppercaseFunc.BindExecutor("UppercaseExecutor");

Ключевые моменты:

  • Создание функции, которая принимает строку и возвращает прописную версию
  • Используйте BindExecutor(), чтобы создать исполнителя из функции

Шаг 3. Определение исполнителя обратного текста

Определите исполнителя, который изменяет текст:

/// <summary>
/// Second executor: reverses the input text and completes the workflow.
/// </summary>
internal sealed class ReverseTextExecutor() : Executor<string, string>("ReverseTextExecutor")
{
    public override ValueTask<string> HandleAsync(string input, IWorkflowContext context, CancellationToken cancellationToken = default)
    {
        // Reverse the input text
        return ValueTask.FromResult(new string(input.Reverse().ToArray()));
    }
}

ReverseTextExecutor reverse = new();

Ключевые моменты:

  • Создание класса, наследуемого от Executor<TInput, TOutput>
  • Реализация HandleAsync() для обработки входных данных и возврата выходных данных

Шаг 4. Создание и подключение рабочего процесса

Подключите исполнители с помощью WorkflowBuilder:

// Build the workflow by connecting executors sequentially
WorkflowBuilder builder = new(uppercase);
builder.AddEdge(uppercase, reverse).WithOutputFrom(reverse);
var workflow = builder.Build();

Ключевые моменты:

  • WorkflowBuilder Конструктор принимает исходный исполнитель
  • AddEdge() создает перенаправленное соединение из верхнего регистра в обратный
  • WithOutputFrom() указывает, какие исполнители создают выходные данные рабочего процесса
  • Build() создает неизменяемый рабочий процесс

Шаг 5. Выполнение рабочего процесса

Запустите рабочий процесс и просмотрите результаты:

// Execute the workflow with input data
await using Run run = await InProcessExecution.RunAsync(workflow, "Hello, World!");
foreach (WorkflowEvent evt in run.NewEvents)
{
    switch (evt)
    {
        case ExecutorCompletedEvent executorComplete:
            Console.WriteLine($"{executorComplete.ExecutorId}: {executorComplete.Data}");
            break;
    }
}

Шаг 6. Общие сведения о выходных данных рабочего процесса

При запуске рабочего процесса вы увидите такие выходные данные:

UppercaseExecutor: HELLO, WORLD!
ReverseTextExecutor: !DLROW ,OLLEH

Сначала входные данные "Hello, World!" преобразуются в верхний регистр ("HELLO, WORLD!"), а затем их порядок изменяется на обратный ("!DLROW ,OLLEH").

Основные понятия, описанные

Интерфейс исполнителя

Исполнители задач из функций

  • Используйте BindExecutor(), чтобы создать исполнителя из функции

Исполнители реализуют Executor<TInput, TOutput>:

  • TInput: тип данных, которые этот исполнитель принимает
  • TOutput: тип данных, которые производит этот исполнитель
  • HandleAsync: метод, обрабатывающий входные данные и возвращающий выходные данные

Шаблон построителя рабочих процессов .NET

WorkflowBuilder предоставляет гибкий API для построения рабочих процессов:

  • Конструктор: принимает начальный исполнитель
  • AddEdge(): создает направленные соединения между исполнителями
  • WithOutputFrom(): указывает, какие исполнители создают выходные данные рабочего процесса
  • Build(): создает окончательный неизменяемый рабочий процесс

Типы событий .NET

Во время выполнения можно наблюдать следующие типы событий:

  • ExecutorCompletedEvent — Когда исполнитель завершит обработку

Полный пример .NET

Полную готовую к выполнению реализацию см. в примере 01_ExecutorsAndEdges в репозитории Agent Framework.

Этот пример включает в себя:

  • Полная реализация со всеми операторами using и структурой классов
  • Дополнительные комментарии, объясняющие основные понятия рабочего процесса
  • Завершение настройки и конфигурации проекта

Обзор

В этом руководстве вы создадите рабочий процесс с двумя исполнителями:

  1. Исполнитель верхнего регистра — преобразует входной текст в верхний регистр
  2. Процедура переворота текста — переворачивает текст и выводит окончательный результат.

Рабочий процесс демонстрирует основные понятия, такие как:

  • Два способа определения единицы работы (узла исполнителя):
    1. Пользовательский класс, который наследуется от Executor с асинхронным методом, обозначенным @handler
    2. Автономная асинхронная функция, украшенная @executor
  • Подключение исполнителей с помощью WorkflowBuilder
  • Передача данных между шагами с помощью ctx.send_message()
  • Получение окончательных выходных данных с помощью ctx.yield_output()
  • Потоковая передача событий для наблюдаемости в режиме реального времени

Основные понятия, описанные в статье

Предпосылки

  • Python 3.10 или более поздней версии
  • Установленный пакет Python для Agent Framework Core: pip install agent-framework-core --pre
  • Для этого базового примера не требуются внешние службы ИИ

Пошаговая реализация

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

Шаг 1. Импорт обязательных модулей

Сначала импортируйте необходимые модули из Agent Framework:

import asyncio
from typing_extensions import Never
from agent_framework import WorkflowBuilder, WorkflowContext, WorkflowOutputEvent, executor

Шаг 2. Создание первого исполнителя

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

class UpperCase(Executor):
    def __init__(self, id: str):
        super().__init__(id=id)

    @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.
        """
        result = text.upper()

        # Send the result to the next executor in the workflow.
        await ctx.send_message(result)

Ключевые моменты:

  • Подклассы Executor позволяют определить именованный узел с хуками жизненного цикла при необходимости
  • Декоратор @handler помечает асинхронный метод, который выполняет работу
  • Подпись обработчика следует контракту:
    • Первый параметр — это типизированные входные данные для этого узла (здесь: text: str)
    • Второй параметр — это WorkflowContext[T_Out], где T_Out является типом данных, который этот узел будет передавать через ctx.send_message() (здесь: str)
  • В обработчике вы вычисляете результат и перенаправляете его на подчиненные узлы с помощью ctx.send_message(result)

Шаг 3. Создание второго исполнителя

Для простых шагов можно пропустить подклассы и определить асинхронную функцию с тем же шаблоном подписи (типизированный вход + WorkflowContext) и декорировать его @executor. При этом создается полностью функциональный узел, который можно подключить к потоку:

@executor(id="reverse_text_executor")
async def reverse_text(text: str, ctx: WorkflowContext[Never, str]) -> None:
    """Reverse the input and yield the workflow output."""
    result = text[::-1]

    # Yield the final output for this workflow run
    await ctx.yield_output(result)

Ключевые моменты:

  • Декоратор @executor преобразует автономную асинхронную функцию в узел рабочего процесса
  • Параметризуется WorkflowContext двумя типами.
    • T_Out = Never: этот узел не отправляет сообщения на подчиненные узлы
    • T_W_Out = str: этот узел выдает выходные данные рабочего процесса типа str
  • Терминальные узлы выдают результаты, используя ctx.yield_output() для предоставления результатов рабочего процесса.
  • Рабочий процесс завершается, когда он становится неактивным (больше не требуется выполнять)

Шаг 4. Создание рабочего процесса

Подключите исполнители с помощью WorkflowBuilder:

upper_case = UpperCase(id="upper_case_executor")

workflow = (
    WorkflowBuilder()
    .add_edge(upper_case, reverse_text)
    .set_start_executor(upper_case)
    .build()
)

Ключевые моменты:

  • add_edge() создает направленные соединения между исполнителями
  • set_start_executor() определяет точку входа
  • build() завершает рабочий процесс

Шаг 5. Запуск рабочего процесса с потоковой передачей

Выполните рабочий процесс и просмотрите события в режиме реального времени:

async def main():
    # Run the workflow and stream events
    async for event in workflow.run_stream("hello world"):
        print(f"Event: {event}")
        if isinstance(event, WorkflowOutputEvent):
            print(f"Workflow completed with result: {event.data}")

if __name__ == "__main__":
    asyncio.run(main())

Шаг 6. Общие сведения о выходных данных

При запуске рабочего процесса вы увидите такие события:

Event: ExecutorInvokedEvent(executor_id=upper_case_executor)
Event: ExecutorCompletedEvent(executor_id=upper_case_executor)
Event: ExecutorInvokedEvent(executor_id=reverse_text_executor)
Event: ExecutorCompletedEvent(executor_id=reverse_text_executor)
Event: WorkflowOutputEvent(data='DLROW OLLEH', source_executor_id=reverse_text_executor)
Workflow completed with result: DLROW OLLEH

Основные понятия, описанные

Два способа определения исполнителей

  1. Пользовательский класс (подкласс Executor): лучше всего подходит, если требуются хуки жизненного цикла или сложное состояние. Определите асинхронный метод с декоратором @handler .
  2. На основе функций (@executor декоратор): лучше всего выполнять простые действия. Определите автономную асинхронную функцию с одинаковым шаблоном подписи.

Оба подхода используют одну и ту же подпись обработчика:

  • Первый параметр: типизированные входные данные для этого узла
  • Второй параметр: a WorkflowContext[T_Out, T_W_Out]

Типы контекста рабочего процесса

Универсальный WorkflowContext тип данных определяет, какие данные передаются между исполнителями.

  • WorkflowContext[T_Out] — используется для узлов, отправляющих сообщения типа T_Out на подчиненные узлы через ctx.send_message()
  • WorkflowContext[T_Out, T_W_Out] — Используется для узлов, которые также дают выходные данные рабочего процесса типа T_W_Out через ctx.yield_output()
  • WorkflowContext без параметров типа эквивалентен WorkflowContext[Never, Never], то есть этот узел не отправляет сообщения на нижестоящие узлы и не выдает выходные данные рабочего процесса.

Типы событий

Во время выполнения потоковой передачи вы увидите следующие типы событий:

  • ExecutorInvokedEvent — При запуске исполнения
  • ExecutorCompletedEvent — Когда исполнитель завершит обработку
  • WorkflowOutputEvent — содержит окончательный результат рабочего процесса

Шаблон построителя рабочих процессов Python

WorkflowBuilder предоставляет гибкий API для построения рабочих процессов:

  • add_edge(): создает направленные соединения между исполнителями
  • set_start_executor(): определяет точку входа рабочего процесса
  • build(): завершает работу и возвращает неизменяемый объект рабочего процесса.

Полный пример

Полную готовую к выполнению реализацию см. в примере в репозитории Agent Framework.

Этот пример включает в себя:

  • Полная реализация со всеми импортами и документацией
  • Дополнительные комментарии, объясняющие основные понятия рабочего процесса
  • Пример выходных данных с ожидаемыми результатами

Дальнейшие шаги

Сведения о создании простого параллельного рабочего процесса

  • Last updated on