Workflow Builder & Execution

A Workflow ties executors and edges together into a directed graph and manages execution. It coordinates executor invocation, message routing, and event streaming.

Building Workflows

Workflows are constructed using the WorkflowBuilder class, which provides a fluent API for defining the workflow structure:

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();

Workflows are constructed using the WorkflowBuilder class:

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()

During build-time validation, a handler can use a bare container annotation, such as list, dict, tuple, or Sequence. It can accept output that uses the corresponding parameterized container, such as list[str].

The workflow package provides a graph-based execution model where executors are connected by edges.

  • Executor - A processing unit that receives input and produces output
  • Edge - Connects the output of one executor to the input of another
  • Builder - Constructs workflows by defining executors and edges
  • Run - Executes a workflow with given input

This function builds a text-processing workflow without requiring a model or credentials:

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) exposes the reverse executor's result as workflow output. The edge still delivers the uppercase result to the reverse executor regardless of output designation.

Pass the workflow returned by buildWorkflow to either execution function below. For the input "Hello, World!", both print Workflow output: !DLROW ,OLLEH.

Workflow Execution

Workflows support both streaming and non-streaming execution modes:

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()}")

Use RunStreaming when you want events as they happen. Pass the caller's context so cancellation reaches the workflow, check its error after iteration ends, and close the run when you're finished:

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 waits while external requests are pending. For workflows that need human input, handle request events or use WatchUntilHalt to return at a pause. See Workflow events.

Use Run to advance to the next halt and then inspect the collected events. A run can halt while waiting for external input, not only when its work is complete:

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() advances its bookmark each time you call it. The second loop uses OutgoingEvents() to replay accumulated executor events instead of trying to consume them again with NewEvents().

These functions close their runs after observing the result. Keep the run open instead if your application needs to send more messages or resume it after a pause.

See Workflow events for complete consumers, output classification, and pause semantics.

Tip

See the workflow examples for complete runnable samples.

Workflow Validation

Workflow builders validate the graph before execution. The exact checks depend on the SDK and the available type metadata:

  • Type compatibility: Checks declared message protocols between connected executors where metadata is available.
  • Graph connectivity: Checks that registered executors are reachable from the start executor.
  • Executor binding: Checks that required executor registrations are present.
  • Edge validation: Checks for invalid connections and duplicate edges according to the builder's rules.

Go Build() rejects unresolved bindings, unreachable registered executors, duplicate unconditional edges, and incompatible declared message types. Conditional routes and dynamic protocols still require runtime testing.

Type validation can create temporary executor instances to inspect their protocols. If a factory or protocol can't be inspected, that compatibility check is skipped. A successful build doesn't guarantee that executor creation or every message delivery succeeds at run time. Handle errors from run creation, event iteration, and workflow failure events.

BuildWithOptions(workflow.BuildOptions{DisableOrphanValidation: true}) disables only the orphan-executor check. Other validation still runs.

Execution Model: Supersteps

The framework uses a modified Pregel execution model — a Bulk Synchronous Parallel (BSP) approach with superstep-based processing.

How Supersteps Work

Workflow execution is organized into discrete supersteps. Each superstep:

  1. Collects all pending messages from the previous superstep
  2. Routes messages to target executors based on edge definitions
  3. Runs all target executors concurrently within the superstep
  4. Waits for all executors to complete before advancing (synchronization barrier)
  5. Queues any new messages emitted by executors for the next superstep
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   │
└─────────────────┘    └─────────────────┘

Synchronization Barrier

The most important characteristic is the synchronization barrier between supersteps. Triggered executors can run concurrently within a superstep, but the workflow doesn't advance to the next superstep until they finish.

This affects fan-out patterns: if you fan out to multiple paths — one with a chain of executors and another with a single long-running executor — the chained path cannot advance until the long-running executor completes.

Why Supersteps?

The BSP model provides important guarantees:

  • Defined step boundaries: Messages emitted during a superstep are queued for later processing.
  • Checkpoint boundaries: Checkpoint-enabled workflows save execution state at superstep boundaries.
  • Buffered state updates: Queued workflow-state changes are committed between supersteps.

These guarantees don't fix execution or event ordering among concurrent executors within one superstep. Synchronize access to application-owned shared state rather than assuming the superstep barrier prevents all races.

Working with the Superstep Model

If you need truly independent parallel paths that don't block each other, consolidate sequential steps into a single executor. Instead of chaining step1 → step2 → step3, combine that logic into one executor. Both parallel paths then execute within a single superstep.

Next steps

Agents in Workflows

Related topics:

  • Viimeksi päivitetty