Workflow-Generator & Ausführung

Ein Workflow verknüpft Executors und Edges zu einem gerichteten Graphen und verwaltet die Ausführung. Er koordiniert den Executoraufruf, das Nachrichtenrouting und das Ereignisstreaming.

Erstellen von Workflows

Workflows werden mithilfe der WorkflowBuilder Klasse erstellt, die eine Fluent-API zum Definieren der Workflowstruktur bereitstellt:

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 werden mithilfe der WorkflowBuilder Klasse erstellt:

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

Das workflow Paket stellt ein graphbasiertes Ausführungsmodell bereit, bei dem Executoren von Kanten verbunden sind.

  • Executor – Eine Verarbeitungseinheit, die Eingabe empfängt und Ausgabe erzeugt
  • Edge – Verbindet die Ausgabe eines Executors mit der Eingabe eines anderen.
  • Builder - Erstellt Workflows durch das Definieren von Executoren und Kanten
  • Ausführen – Führt einen Workflow mit gegebener Eingabe aus.
import (
    "github.com/microsoft/agent-framework-go/workflow"
    "github.com/microsoft/agent-framework-go/workflow/inproc"
)

uppercase := workflow.NewExecutor("UppercaseExecutor", func(input string) string {
    return strings.ToUpper(input)
}).Bind()

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

wf, err := workflow.NewBuilder(uppercase).
    AddEdge(uppercase, reverse).
    WithOutputFrom(reverse).
    Build()
if err != nil {
    return err
}

Workflowausführung

Workflows unterstützen Streaming- und Nicht-Streaming-Ausführungsmodi:

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

Verwenden Sie RunStreaming, wenn Sie Ereignisse in Echtzeit erhalten möchten:

stream, err := inproc.Default.RunStreaming(context.Background(), wf, "Hello, World!")
if err != nil {
    return err
}
defer stream.Close(context.Background())

for evt, err := range stream.WatchStream(context.Background()) {
    if err != nil {
        return err
    }
    if output, ok := evt.(workflow.OutputEvent); ok {
        fmt.Printf("Workflow completed: %v\n", output.Output)
    }
}

Verwenden Sie Run, wenn Sie auf den Abschluss des Workflows warten möchten und anschließend die gesammelten Ereignisse prüfen:

run, err := inproc.Default.Run(context.Background(), wf, "Hello, World!")
if err != nil {
    return err
}

for evt := range run.NewEvents() {
    if output, ok := evt.(workflow.OutputEvent); ok {
        fmt.Printf("Final result: %v\n", output.Output)
    }
}

Sie können auch die Executor-Ereignisse überprüfen, die von einer nicht streamenden Ausführung erfasst wurden:

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

Tip

In den Workflowbeispielen finden Sie vollständige Runnable-Beispiele.

Workflowüberprüfung

Das Framework führt beim Erstellen von Workflows eine umfassende Validierung durch:

  • Typkompatibilität: Stellt sicher, dass Nachrichtentypen zwischen verbundenen Executoren kompatibel sind
  • Graph-Konnektivität: Überprüft, ob alle Executors vom Start-Executor erreichbar sind
  • Executor Binding: Bestätigt, dass alle Executor-Objekte ordnungsgemäß gebunden und instanziiert sind
  • Kantenvalidierung: Überprüft auf doppelte Kanten und ungültige Verbindungen

Ausführungsmodell: Supersteps

Das Framework verwendet ein modifiziertes Pregel-Ausführungsmodell – einen Bulk Synchronous Parallel (BSP)-Ansatz mit superstepbasierter Verarbeitung.

Funktionsweise von Supersteps

Die Workflowausführung ist in diskrete Supersteps unterteilt. Jeder Superstep:

  1. Sammelt alle ausstehenden Nachrichten aus dem vorherigen Superstep
  2. Leitet Nachrichten basierend auf Edge-Definitionen an Ziel-Executors weiter.
  3. Führt alle Zielausführungsinstanzen gleichzeitig innerhalb des Supersteps aus
  4. Wartet, bis alle Executors abgeschlossen sind, bevor ein Fortschritt erfolgt (Synchronisierungsbarriere)
  5. Stellt alle neuen Nachrichten, die von Executors für den nächsten Superstep ausgegeben werden, in die Warteschlange.
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   │
└─────────────────┘    └─────────────────┘

Synchronisierungsbarriere

Das wichtigste Merkmal ist die Synchronisierungsbarriere zwischen Supersteps. Innerhalb eines einzelnen Supersteps werden alle ausgelösten Executoren parallel ausgeführt, der Workflow wechselt jedoch erst zum nächsten Superstep, wenn jeder Executor abgeschlossen ist.

Dies wirkt sich auf Ausfächerungsmuster aus: Wenn Sie auf mehrere Pfade ausfächern – einen mit einer Folge von Executors und einen anderen mit einem einzigen zeitintensiven Executor – kann der verkettete Pfad erst nach Abschluss des zeitintensiven Executors voranschreiten.

Warum Supersteps?

Das BSP-Modell bietet wichtige Garantien:

  • Deterministische Ausführung: Aufgrund derselben Eingabe wird der Workflow immer in derselben Reihenfolge ausgeführt.
  • Zuverlässige Prüfpunkte: Der Zustand kann an Superstepgrenzen für Fehlertoleranz gespeichert werden.
  • Einfachere Schlussfolgerung: Keine Race-Bedingungen zwischen Supersteps, jeder sieht eine einheitliche Ansicht von Nachrichten

Arbeiten mit dem Superstep-Modell

Wenn Sie wirklich unabhängige parallele Pfade benötigen, die sich nicht gegenseitig blockieren, konsolidieren Sie sequenzielle Schritte in einem einzelnen Executor. Statt die Verkettung step1 → step2 → step3 zu nutzen, fassen Sie diese Logik in einem Executor zusammen. Beide parallelen Pfade werden dann innerhalb eines einzigen Supersteps ausgeführt.

Nächste Schritte

Agents in Workflows

Verwandte Themen:

  • Last updated on