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

Während der Buildzeitüberprüfung kann ein Handler eine bloße Containerannotation verwenden, wie z. B. list, dict, tuple oder Sequence. Sie kann die Ausgabe akzeptieren, die den entsprechenden parametrisierten Container verwendet, wie z. B. list[str].

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.

Diese Funktion erstellt einen Textverarbeitungsworkflow, ohne dass ein Modell oder Anmeldeinformationen erforderlich sind:

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) macht das Ergebnis des Reverse-Executors als Workflowausgabe verfügbar. Der Edge übermittelt das in Großbuchstaben geschriebene Ergebnis unabhängig von der Ausgabebezeichnung weiterhin an den Reverse-Executor.

Übergeben Sie den von buildWorkflow zurückgegebenen Workflow an eine der folgenden Ausführungsfunktionen. Für die Eingabe "Hello, World!" geben beide Workflow output: !DLROW ,OLLEH aus.

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 bei deren Auftreten ein Ereignis wünschen. Übergeben Sie den Kontext des Aufrufers, damit der Abbruch den Workflow erreicht, überprüfen Sie den Fehler nach dem Ende der Iteration und schließen Sie die Ausführung, wenn Sie fertig sind:

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 wartet, während externe Anfragen ausstehen. Behandeln Sie für Workflows, die menschliche Eingaben erfordern, Anforderungsereignisse, oder verwenden Sie WatchUntilHalt, um bei einer Pause zurückzukehren. Siehe Workflow-Ereignisse.

Verwenden Sie Run, um zum nächsten Halt vorzurücken und dann die gesammelten Ereignisse zu überprüfen. Eine Ausführung kann anhalten, während auf externe Eingaben gewartet wird, nicht nur, wenn ihre Arbeit abgeschlossen ist:

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() rückt sein Lesezeichen bei jedem Aufruf vor. Die zweite Schleife verwendet OutgoingEvents(), um akkumulierte Executor-Ereignisse wiederzugeben, anstatt zu versuchen, sie erneut mit NewEvents() zu verarbeiten.

Diese Funktionen schließen ihre Ausführungen, nachdem sie das Ergebnis beobachtet haben. Lassen Sie den Run stattdessen geöffnet, wenn Ihre Anwendung mehr Nachrichten senden oder ihn nach einer Pause fortsetzen muss.

Informationen zu vollständigen Consumern, Ausgabeklassifizierung und Pausensemantik finden Sie unter Workflowereignisse.

Tip

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

Workflowüberprüfung

Workflow-Generatoren überprüfen den Graphen vor der Ausführung. Die genauen Überprüfungen hängen vom SDK und den verfügbaren Typmetadaten ab:

  • Typkompatibilität: Überprüft deklarierte Nachrichtenprotokolle zwischen verbundenen Executoren, sofern Metadaten verfügbar sind.
  • Graph-Konnektivität: Überprüft, ob registrierte Executors vom Start-Executor erreichbar sind.
  • Executor binding: Prüft, ob die erforderlichen Executor-Registrierungen vorhanden sind.
  • Edgeüberprüfung: Sucht nach ungültigen Verbindungen und doppelten Kanten gemäß den Regeln des Builders.

Go Build() lehnt nicht aufgelöste Bindungen, nicht erreichbare registrierte Executoren, doppelte bedingungslose Kanten und inkompatible deklarierte Nachrichtentypen ab. Bedingte Routen und dynamische Protokolle erfordern weiterhin Laufzeittests.

Die Typüberprüfung kann temporäre Executor-Instanzen erstellen, um deren Protokolle zu überprüfen. Wenn eine Factory oder ein Protokoll nicht geprüft werden kann, wird diese Kompatibilitätsprüfung übersprungen. Ein erfolgreicher Build garantiert nicht, dass die Erstellung eines Executors oder jede Nachrichtenübermittlung zur Laufzeit erfolgreich ist. Behandeln Sie Fehler bei der Erstellung von Runs, der Ereignisiteration und bei Workflow-Fehlerereignissen.

BuildWithOptions(workflow.BuildOptions{DisableOrphanValidation: true}) deaktiviert nur die Überprüfung auf verwaiste Executoren. Andere Überprüfungen werden weiterhin ausgeführt.

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. Ausgelöste Executoren können innerhalb eines Supersteps gleichzeitig ausgeführt werden, aber der Workflow rückt erst zum nächsten Superstep vor, wenn sie abgeschlossen sind.

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:

  • Definierte Schrittbegrenzungen: Nachrichten, die während eines Supersteps ausgegeben werden, werden zur späteren Verarbeitung in die Warteschlange gestellt.
  • Prüfpunktgrenzen: Prüfpunktfähige Workflows speichern den Ausführungszustand an den Grenzen der Supersteps.
  • Aktualisierungen des gepufferten Zustands: Änderungen am Workflowstatus in der Warteschlange werden zwischen Supersteps übernommen.

Diese Garantien beheben nicht die Ausführung oder die Ereignisreihenfolge zwischen gleichzeitigen Executoren innerhalb eines Supersteps. Synchronisieren Sie den Zugriff auf den gemeinsam genutzten Zustand der Anwendung, anstatt anzunehmen, dass die Superstep-Barriere alle Race Conditions verhindert.

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:

  • Zuletzt aktualisiert um