😎 AI-агенты в отказоустойчивом пайплайне на .NET

Microsoft Agent Framework получил поддержку устойчивых воркфлоу. Теперь можно описать пайплайн из нескольких AI-агентов один раз и запустить его как в памяти процесса, так и на Azure Functions с автомасштабированием и отказоустойчивостью.

Что было до этого

Если в многошаговом пайплайне процесс падал или перезапускался, всё состояние терялось. Для долгоживущих задач это критично: агент может работать минуты или часы, и потерять прогресс на середине — дорого.

Что делает MAF Workflows

Фреймворк позволяет описать шаги как Executor<TInput, TOutput>, соединить их в граф через WorkflowBuilder и выбрать способ запуска, не меняя определение воркфлоу.

Поддерживаются паттерны:

- последовательные цепочки
- параллельный fan-out / fan-in
- условный роутинг (AddSwitch)
- человек в петле (human-in-the-loop через RequestPort)
- вложенные воркфлоу как sub-orchestrations

Как это работает

Описываем шаги:

internal sealed class OrderLookup()
    : Executor<OrderCancelRequest, Order>("OrderLookup")
{
    public override async ValueTask<Order> HandleAsync(
        OrderCancelRequest message,
        IWorkflowContext context,
        CancellationToken cancellationToken = default)
    {
        await Task.Delay(100, cancellationToken);
        return new Order(Id: message.OrderId, ...);
    }
}

Собираем граф:

Workflow cancelOrder = new WorkflowBuilder(orderLookup)
    .WithName("CancelOrder")
    .AddEdge(orderLookup, orderCancel)
    .AddEdge(orderCancel, sendEmail)
    .Build();

Запускаем в памяти (для локальной разработки):

await using StreamingRun run =
    await InProcessExecution.RunStreamingAsync(cancelOrder, input: cancelRequest);

await foreach (WorkflowEvent evt in run.WatchStreamAsync())
{
    if (evt is ExecutorCompletedEvent completed)
        Console.WriteLine($"{completed.ExecutorId}: {completed.Data}");
}

Добавляем устойчивость

Для production подключаем Microsoft.Agents.AI.DurableTask. Определение воркфлоу остаётся прежним, меняется только хостинг:

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

Запускаем локальный эмулятор DTS:

docker run -d --name dts-emulator \
  -p 8080:8080 -p 8082:8082 \
  mcr.microsoft.com/dts/dts-emulator:latest

Дашборд доступен на http://localhost:8082. Там видно хронологию шагов, входные и выходные данные каждого executor'а.

Регистрируем воркфлоу с durable runtime:

IHost host = Host.CreateDefaultBuilder(args)
    .ConfigureServices(services =>
    {
        services.ConfigureDurableWorkflows(
            workflowOptions => workflowOptions.AddWorkflow(cancelOrder),
            workerBuilder: b => b.UseDurableTaskScheduler(dtsConnectionString),
            clientBuilder: b => b.UseDurableTaskScheduler(dtsConnectionString));
    })
    .Build();

Параллельные AI-агенты

Агенты подключаются напрямую через AsAIAgent и запускаются параллельно. Если процесс упал в середине, завершённые агенты не перезапускаются:

AIAgent physicist = chatClient.AsAIAgent(
    "You are a physics expert. Be concise.", "Physicist");
AIAgent chemist = chatClient.AsAIAgent(
    "You are a chemistry expert. Be concise.", "Chemist");

Workflow workflow = new WorkflowBuilder(parseQuestion)
    .AddFanOutEdge(parseQuestion, [physicist, chemist])
    .AddFanInBarrierEdge([physicist, chemist], aggregator)
    .Build();

Одно определение воркфлоу работает везде: в памяти для разработки, с Durable Task для отказоустойчивости, на Azure Functions для serverless. Переключение между режимами не затрагивает код executor'ов.

➡️ Блог разработчиков

📍 Навигация: Вакансии • Задачи • Собесы

🐸 Библиотека шарписта

#il_люминатор