Конструктор рабочих процессов и их выполнение

Рабочий процесс связывает исполнителей и ребер между собой в направленный граф и управляет выполнением. Он координирует вызов исполнителя, маршрутизацию сообщений и потоковую передачу событий.

Создание рабочих процессов

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

using Microsoft.Agents.AI.Workflows;

var processor = new DataProcessor();
var validator = new Validator();
var formatter = new Formatter();

// Build workflow
WorkflowBuilder builder = new(processor); // Set starting executor
builder.AddEdge(processor, validator);
builder.AddEdge(validator, formatter);
var workflow = builder.Build();

Рабочие процессы создаются с помощью WorkflowBuilder класса:

from agent_framework import WorkflowBuilder

processor = DataProcessor()
validator = Validator()
formatter = Formatter()

# Build workflow
builder = WorkflowBuilder(start_executor=processor)
builder.add_edge(processor, validator)
builder.add_edge(validator, formatter)
workflow = builder.build()

Во время проверки во время сборки обработчик может использовать аннотацию простого контейнера, например list, dict, Sequence или tuple. Он может принимать выходные данные, использующие соответствующий параметризованный контейнер, например list[str].

Пакет workflow предоставляет графовую модель выполнения, в которой исполнители соединены рёбрами.

Исполнитель — единица обработки, которая получает входные данные и создает выходные данные
Edge — подключает выходные данные одного исполнителя к входным данным другого
Builder — создает рабочие процессы, задавая исполнителей и рёбра
Выполнение — выполняет рабочий процесс с заданными входными данными

Эта функция создает рабочий процесс обработки текста, не требуя модели или учетных данных:

import (
    "slices"
    "strings"

    "github.com/microsoft/agent-framework-go/workflow"
)

func buildWorkflow() (*workflow.Workflow, error) {
    uppercase := workflow.NewExecutor(
        "UppercaseExecutor", strings.ToUpper,
    ).Bind()

    reverse := workflow.NewExecutor(
        "ReverseExecutor", func(input string) string {
            runes := []rune(input)
            slices.Reverse(runes)
            return string(runes)
        },
    ).Bind()

    return workflow.NewBuilder(uppercase).
        AddEdge(uppercase, reverse).
        WithOutputFrom(reverse).
        Build()
}

WithOutputFrom(reverse) предоставляет результат обратного исполнителя задач в виде выходных данных рабочего процесса. Ребро по-прежнему доставляет результат в верхнем регистре обратному исполнителю независимо от назначения выходных данных.

Передайте рабочий процесс, возвращенный buildWorkflow, в одну из двух указанных ниже функций выполнения. Для входных данных "Hello, World!", оба выводятся как Workflow output: !DLROW ,OLLEH.

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

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

using Microsoft.Agents.AI.Workflows;

// Streaming execution — get events as they happen
StreamingRun run = await InProcessExecution.RunStreamingAsync(workflow, inputMessage);
await foreach (WorkflowEvent evt in run.WatchStreamAsync())
{
    if (evt is ExecutorCompletedEvent executorComplete)
    {
        Console.WriteLine($"{executorComplete.ExecutorId}: {executorComplete.Data}");
    }

    if (evt is WorkflowOutputEvent outputEvt)
    {
        Console.WriteLine($"Workflow completed: {outputEvt.Data}");
    }
}

// Non-streaming execution — wait for completion
Run result = await InProcessExecution.RunAsync(workflow, inputMessage);
foreach (WorkflowEvent evt in result.NewEvents)
{
    if (evt is WorkflowOutputEvent outputEvt)
    {
        Console.WriteLine($"Final result: {outputEvt.Data}");
    }
}

# Streaming execution — get events as they happen
async for event in workflow.run(input_message, stream=True):
    if event.type == "output":
        print(f"Workflow completed: {event.data}")

# Non-streaming execution — wait for completion
events = await workflow.run(input_message)
print(f"Final result: {events.get_outputs()}")

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

import (
    "context"
    "errors"
    "fmt"

    "github.com/microsoft/agent-framework-go/workflow"
    "github.com/microsoft/agent-framework-go/workflow/inproc"
)

func runStreaming(
    ctx context.Context, wf *workflow.Workflow,
) (err error) {
    stream, err := inproc.Default.RunStreaming(ctx, wf, "Hello, World!")
    if err != nil {
        return err
    }
    defer func() {
        err = errors.Join(err, stream.Close(context.WithoutCancel(ctx)))
    }()

    for evt, streamErr := range stream.WatchStream(ctx) {
        if streamErr != nil {
            return streamErr
        }
        switch event := evt.(type) {
        case workflow.OutputEvent:
            fmt.Printf("Workflow output: %v\n", event.Output)
        case workflow.ErrorEvent:
            return event.Error
        case workflow.ExecutorFailedEvent:
            return fmt.Errorf("executor %q failed: %w",
                event.ExecutorID, event.Error)
        }
    }
    return ctx.Err()
}

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

Используйте Run для перехода к следующей точке останова, а затем проверьте собранные события. Запуск может остановиться во время ожидания внешних входных данных, а не только после завершения работы:

import (
    "context"
    "errors"
    "fmt"

    "github.com/microsoft/agent-framework-go/workflow"
    "github.com/microsoft/agent-framework-go/workflow/inproc"
)

func runToHalt(
    ctx context.Context, wf *workflow.Workflow,
) (err error) {
    run, err := inproc.Default.Run(ctx, wf, "Hello, World!")
    if err != nil {
        return err
    }
    defer func() {
        err = errors.Join(err, run.Close(context.WithoutCancel(ctx)))
    }()

    for evt := range run.NewEvents() {
        switch event := evt.(type) {
        case workflow.OutputEvent:
            fmt.Printf("Workflow output: %v\n", event.Output)
        case workflow.ErrorEvent:
            return event.Error
        case workflow.ExecutorFailedEvent:
            return fmt.Errorf("executor %q failed: %w",
                event.ExecutorID, event.Error)
        }
    }

    for evt := range run.OutgoingEvents() {
        if evt, ok := evt.(workflow.ExecutorCompletedEvent); ok {
            fmt.Printf("%s: %v\n", evt.ExecutorID, evt.Result)
        }
    }
    return ctx.Err()
}

NewEvents() продвигает свою закладку при каждом вызове. Второй цикл использует OutgoingEvents() для воспроизведения накопленных событий executor вместо того, чтобы снова обработать их с помощью NewEvents().

Эти функции завершают выполнение после получения результата. Не завершайте запуск, если приложению нужно отправить больше сообщений или возобновить его после приостановки.

См. события рабочего процесса для получения полной информации о потребителях, классификации выходных данных и семантике приостановки.

Tip

См. примеры рабочих процессов, где приведены полные готовые к запуску примеры.

Проверка рабочего процесса

Конструкторы рабочих процессов проверяют граф перед выполнением. Точные проверки зависят от пакета SDK и доступных метаданных типа:

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

Go Build() отклоняет неразрешенные привязки, недостижимые зарегистрированные исполнители, дублирующиеся безусловные рёбра и несовместимые объявленные типы сообщений. Условные маршруты и динамические протоколы по-прежнему требуют тестирования во время выполнения.

Проверка типов может создавать временные экземпляры executor для проверки их протоколов. Если не удается проверить фабрику или протокол, проверка совместимости пропускается. Успешная сборка не гарантирует успешность создания Executor или каждой доставки сообщений во время выполнения. Обработка ошибок при создании запуска, итерации событий и сбоях рабочего процесса.

BuildWithOptions(workflow.BuildOptions{DisableOrphanValidation: true}) отключает только проверку исполнителя-сироты. Другая проверка по-прежнему выполняется.

Модель выполнения: supersteps

Фреймворк использует измененную модель выполнения Pregel — метод массового синхронного параллелизма (BSP) с обработкой, основанной на супершаге.

Как работают суперстепы

Выполнение рабочего процесса организовано в дискретные супершаги. Каждый супершаг:

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

Superstep N:
┌─────────────────┐    ┌─────────────────┐    ┌─────────────────┐
│  Collect All    │───▶│  Route Messages │───▶│  Execute All    │
│  Pending        │    │  Based on Type  │    │  Target         │
│  Messages       │    │  & Conditions   │    │  Executors      │
└─────────────────┘    └─────────────────┘    └─────────────────┘
                                                       │
                                                       │ (barrier: wait for all)
┌─────────────────┐    ┌─────────────────┐             │
│  Start Next     │◀───│  Emit Events &  │◀────────────┘
│  Superstep      │    │  New Messages   │
└─────────────────┘    └─────────────────┘

Барьер синхронизации

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

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

Почему Суперстеп?

Модель BSP обеспечивает важные гарантии:

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

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

Работа с моделью Superstep

Если вам нужны действительно независимые параллельные пути, которые не блокируют друг друга, консолидируйте последовательные шаги в один исполнитель. Вместо чейниннирования step1 → step2 → step3, объедините эту логику в одного исполнителя. Оба параллельных пути затем выполняются в одном суперстепе.

Дальнейшие действия

Агенты в рабочих процессах

См. также:

Исполнители — единицы обработки в рабочем процессе
Ребра — связи между исполнителями
События — наблюдаемость рабочего процесса
Управление состоянием

Обратная связь

Были ли сведения на этой странице полезными?

Последнее обновление: 2026-10-29