Шаг 5. Рабочие процессы

Рабочие процессы позволяют объединить несколько шагов — каждый шаг обрабатывает данные и передает их в следующий.

Определение шагов рабочего процесса (исполнителей):

using Microsoft.Agents.AI.Workflows;

// Step 1: Convert text to uppercase
Func<string, string> uppercaseFunc = s => s.ToUpperInvariant();
var uppercase = uppercaseFunc.BindAsExecutor("UppercaseExecutor");

// Step 2: Reverse the string and yield output
class ReverseTextExecutor() : Executor<string, string>("ReverseTextExecutor")
{
    public override ValueTask<string> HandleAsync(string message, IWorkflowContext context, CancellationToken cancellationToken = default)
    {
        return ValueTask.FromResult(string.Concat(message.Reverse()));
    }
}
ReverseTextExecutor reverse = new();

Создайте и запустите рабочий процесс:

WorkflowBuilder builder = new(uppercase);
builder.AddEdge(uppercase, reverse).WithOutputFrom(reverse);
var workflow = builder.Build();

await using Run run = await InProcessExecution.RunAsync(workflow, "Hello, World!");
foreach (WorkflowEvent evt in run.NewEvents)
{
    if (evt is ExecutorCompletedEvent executorComplete)
    {
        Console.WriteLine($"{executorComplete.ExecutorId}: {executorComplete.Data}");
    }
}

Подсказка

См. здесь полный пример запускаемого приложения.

Определите шаги рабочего процесса (исполнители) и соедините их с ребрами:

# Step 1: A class-based executor that converts text to uppercase
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 input to uppercase and forward to the next node."""
        await ctx.send_message(text.upper())


# Step 2: A function-based executor that reverses the string and yields output
@executor(id="reverse_text")
async def reverse_text(text: str, ctx: WorkflowContext[Never, str]) -> None:
    """Reverse the string and yield the final workflow output."""
    await ctx.yield_output(text[::-1])


def create_workflow():
    """Build the workflow: UpperCase → reverse_text."""
    upper = UpperCase(id="upper_case")
    return WorkflowBuilder(start_executor=upper).add_edge(upper, reverse_text).build()

Создайте и запустите рабочий процесс:

workflow = create_workflow()

events = await workflow.run("hello world")
print(f"Output: {events.get_outputs()}")
print(f"Final state: {events.get_final_state()}")

Подсказка

Подробный пример можно найти в файле для полного выполнения.

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

Шаг 6. Размещение агента

Вернитесь глубже:

• Общие сведения о рабочих процессах— общие сведения об архитектуре рабочих процессов
• Последовательные рабочие процессы — линейные пошаговые шаблоны
• Агенты в рабочих процессах — использование агентов в качестве шагов рабочего процесса