Workflow-Generator & Ausführung

Ein Workflow verknüpft Ausführende und Kanten in 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 Executoren vom Startausführer 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 Edgedefinitionen an Zielausführer weiter.
  3. Führt alle Zielausführungsinstanzen gleichzeitig innerhalb des Supersteps aus.
  4. Wartet auf die Fertigstellung aller Executoren, bevor ein Fortschritt erfolgt (Synchronisierungsbarriere)
  5. Stellt alle neuen Nachrichten, die von Ausführungsinstanzen 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 Fanout-Muster aus: Wenn Sie sich auf mehrere Pfade ausfächern – einem mit einer Folge von Executoren und einem anderen mit einem einzigen lang laufenden Executor – kann der verkettete Pfad erst nach Abschluss des lang laufenden 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 Superschrittgrenzen für Fehlertoleranz gespeichert werden.
  • Einfachere Begründung: Keine Rennbedingungen 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