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

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

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

Обзор

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

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

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

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

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

• Исполнитель
• Прямые края
• Построитель рабочих процессов
• Events

Предпосылки

• Пакет SDK для .NET 8.0 или более поздней версии
• Для этого базового примера не требуются внешние службы ИИ
• Новое консольное приложение

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

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

Шаг 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. Процедура переворота текста — переворачивает текст и выводит окончательный результат.

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

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

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

• Исполнитель
• Прямые края
• Построитель рабочих процессов
• Events

Предпосылки

• 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 регистрирует эту функцию как узел рабочего процесса
• WorkflowContext[str] указывает, что этот исполнитель отправляет строку вниз, указав параметр первого типа
• ctx.send_message() Передает данные на следующий шаг

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

Создайте исполнителя, который отменяет текст и выдает окончательные выходные данные метода, украшенного @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)

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

• WorkflowContext[Never, str] указывает, что это исполнитель терминала, который не отправляет сообщения, указывая Never в качестве параметра первого типа, но создает выходные данные рабочего процесса, указывая 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

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

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

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

• WorkflowContext[str] — отправляет строку следующему исполнителю.
• WorkflowContext[Never, str] — исполнитель терминала, который выдает выходные данные рабочего процесса типа строка

Типы событий

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

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

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

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

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

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

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

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

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

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

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