Events

Das Workflow-Ereignissystem ermöglicht die Beobachtbarkeit der Workflow-Ausführung. Ereignisse werden während der Ausführung an wichtigen Punkten ausgegeben und können per Streaming in Echtzeit genutzt werden.

Integrierte Ereignistypen

// Workflow lifecycle events
WorkflowStartedEvent     // Workflow execution begins
WorkflowOutputEvent      // Workflow outputs data
WorkflowErrorEvent       // Workflow encounters an error
WorkflowWarningEvent     // Workflow encountered a warning

// Executor events
ExecutorInvokedEvent     // Executor starts processing
ExecutorCompletedEvent   // Executor finishes processing
ExecutorFailedEvent      // Executor encounters an error
AgentResponseEvent       // An agent run produces output
AgentResponseUpdateEvent // An agent run produces a streaming update

// Superstep events
SuperStepStartedEvent    // Superstep begins
SuperStepCompletedEvent  // Superstep completes

// Request events
RequestInfoEvent         // A request is issued

Note

Wenn Agents genehmigungsrelevante Tools verwenden, trägt RequestInfoEvent typischerweise eine ToolApprovalRequestContent-Nutzlast für Toolaufrufe, die eine menschliche Genehmigung erfordern. Details zur Behandlung dieser Ereignisse finden Sie unter "Human-in-the-Loop ".

# All events use the unified WorkflowEvent class with a type discriminator:

# Workflow lifecycle events
WorkflowEvent.type == "started"             # Workflow execution begins
WorkflowEvent.type == "status"              # Workflow state changed (use .state)
WorkflowEvent.type == "output"              # Workflow produces a terminal (final) output
WorkflowEvent.type == "intermediate"        # Workflow produces an intermediate (observational) output
WorkflowEvent.type == "failed"              # Workflow terminated with error (use .details)
WorkflowEvent.type == "error"               # Non-fatal error from user code
WorkflowEvent.type == "warning"             # Workflow encountered a warning

# Executor events
WorkflowEvent.type == "executor_invoked"    # Executor starts processing
WorkflowEvent.type == "executor_completed"  # Executor finishes processing
WorkflowEvent.type == "executor_failed"     # Executor encounters an error
WorkflowEvent.type == "data"                # Deprecated alias for "intermediate"

# Superstep events
WorkflowEvent.type == "superstep_started"   # Superstep begins
WorkflowEvent.type == "superstep_completed" # Superstep completes

# Request events
WorkflowEvent.type == "request_info"        # A request is issued

Note

Wenn Agents genehmigungsrelevante Tools verwenden, tragen request_info-Ereignisse in der Regel eine Content-Nutzlast mit type == "function_approval_request" für Toolaufrufe, die eine menschliche Genehmigung erfordern. Details zur Behandlung dieser Ereignisse finden Sie unter "Human-in-the-Loop ".

Note

"output" und "intermediate" sind die beiden Ausgabediskriminatoren. Ein als Quelle für die Terminalausgabe festgelegter Executor gibt "output"-Ereignisse aus (genutzt von WorkflowRunResult.get_outputs()). Eine, die als Zwischenausgabequelle festgelegt ist, gibt "intermediate" Ereignisse aus (genutzt von WorkflowRunResult.get_intermediate_outputs()). Der Typ "data" ist ein veralteter Alias für "intermediate" und wird in einer künftigen Version entfernt; in neuem Code sollte stattdessen bevorzugt nach "intermediate" gefiltert werden.

Go-Ereignisse implementieren workflow.Event und stellen ihre Nutzlast über Data() bereit. Verwenden Sie einen Typ-Switch, um die typisierten Felder zu überprüfen:

Ereignistyp	Nutzlast und Zweck
workflow.StartedEvent	Startet einen Eingabeverarbeitungszyklus. Message enthält die Eingabe.
workflow.OutputEvent	Ein festgelegter Executor liefert einen Wert in Output. Tags identifiziert die Zwischenausgabe.
workflow.ErrorEvent	Meldet einen Fehler in Error. SubWorkflowID kennzeichnet einen verschachtelten Workflow, wenn diese Option gesetzt ist.
workflow.WorkflowWarningEvent	Gibt eine Warnung in Message aus.
workflow.ExecutorInvokedEvent	Ein Executor beginnt mit der Verarbeitung. ExecutorID identifiziert ihn, und Message enthält seine Eingabe.
workflow.ExecutorCompletedEvent	Ein Executor schließt die Verarbeitung ab. Result enthält sein Ergebnis.
workflow.ExecutorFailedEvent	Ein Executor schlägt fehl. ExecutorID und Error identifizieren den Fehler.
workflow.SuperStepStartedEvent	Ein Superstep beginnt. StepNumber und StartInfo beschreiben es.
workflow.SuperStepCompletedEvent	Ein Superstep ist abgeschlossen. CompletionInfo beschreibt ausstehende Arbeit, Anfragen und jeden Prüfpunkt.
workflow.RequestInfoEvent	Request enthält eine externe Anfrage, die der Host beantworten kann.
workflow.RequestHaltEvent	Ein Executor fordert einen Halt an. Result enthält die Nutzlast, falls vorhanden.

Go verwendet OutputEvent sowohl für Terminal- als auch für Zwischenausgabe. Rufen Sie sie mit IsIntermediate() auf, um sie zu unterscheiden; dabei werden Pythons separate "output"- und "intermediate"-Diskriminatoren nicht verwendet.

Gehostete Agents verwenden ebenfalls OutputEvent, mit einer *agent.Response- oder *agent.ResponseUpdate-Nutzlast, wenn die entsprechenden Ereignisse aktiviert sind. Go definiert keine separaten AgentResponseEvent Typen und AgentResponseUpdateEvent Typen. Siehe Agent-Executors für die Konfiguration.

Verbrauchsereignisse

using Microsoft.Agents.AI.Workflows;

await foreach (WorkflowEvent evt in run.WatchStreamAsync())
{
    switch (evt)
    {
        case ExecutorInvokedEvent invoke:
            Console.WriteLine($"Starting {invoke.ExecutorId}");
            break;

        case ExecutorCompletedEvent complete:
            Console.WriteLine($"Completed {complete.ExecutorId}: {complete.Data}");
            break;

        case WorkflowOutputEvent output:
            Console.WriteLine($"Workflow output: {output.Data}");
            return;

        case WorkflowErrorEvent error:
            Console.WriteLine($"Workflow error: {error.Exception}");
            return;
    }
}

from agent_framework import WorkflowEvent

async for event in workflow.run(input_message, stream=True):
    if event.type == "executor_invoked":
        print(f"Starting {event.executor_id}")
    elif event.type == "executor_completed":
        print(f"Completed {event.executor_id}: {event.data}")
    elif event.type == "intermediate":
        print(f"Intermediate output from {event.executor_id}: {event.data}")
    elif event.type == "output":
        print(f"Terminal output: {event.data}")
        return
    elif event.type == "error":
        print(f"Workflow error: {event.data}")
        return

Verwenden Sie einen Typschalter für Go-Workflowereignisse. Behandeln Sie Iteratorfehler und Fehlerereignisse separat: Der Empfang einer Ausgabe garantiert nicht, dass der Rest des Workflows erfolgreich ist.

Beobachten von Ereignissen

inproc.Default.Run rückt zum nächsten Halt vor und akkumuliert Ereignisse. Die Ausführung kann für externe Eingaben pausieren. Ein Halt ist nicht zwangsläufig ein Abschluss.

Die folgende Funktion gibt Executor-Aktivität und -Ausgabe aus, gibt Fehler zurück und schließt die Ausführung:

import (
    "context"
    "errors"
    "fmt"

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

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

    for event := range run.NewEvents() {
        switch event := event.(type) {
        case workflow.ExecutorInvokedEvent:
            fmt.Printf("Starting %s\n", event.ExecutorID)
        case workflow.ExecutorCompletedEvent:
            fmt.Printf("Completed %s: %v\n",
                event.ExecutorID, event.Result)
        case workflow.OutputEvent:
            kind := "Terminal"
            if event.IsIntermediate() {
                kind = "Intermediate"
            }
            fmt.Printf("%s output from %s: %v\n",
                kind, event.ExecutorID, event.Output)
        case workflow.ErrorEvent:
            return event.Error
        case workflow.ExecutorFailedEvent:
            return fmt.Errorf("executor %q failed: %w",
                event.ExecutorID, event.Error)
        }
    }
    return nil
}

NewEvents() gibt die Ereignisse zurück, die seit dem vorherigen Aufruf gesammelt wurden. Sein Lesezeichen rückt vor, wenn Sie es aufrufen, auch wenn Sie die Iteration vorzeitig beenden. Verwenden Sie OutgoingEvents(), um alle angesammelten Ereignisse wiederzugeben, statt NewEvents() erneut aufzurufen.

Streamingereignisse

Verwenden Sie RunStreaming, um Ereignisse beim Eintreffen zu beobachten. Diese Funktion beobachtet einen Workflow, der keine externen Eingaben erfordert:

import (
    "context"
    "errors"
    "fmt"

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

func streamEvents(
    ctx context.Context, wf *workflow.Workflow, input any,
) (err error) {
    run, err := inproc.Default.RunStreaming(ctx, wf, input)
    if err != nil {
        return err
    }
    defer func() {
        err = errors.Join(err, run.Close(context.WithoutCancel(ctx)))
    }()

    for event, streamErr := range run.WatchStream(ctx) {
        if streamErr != nil {
            return streamErr
        }
        switch event := event.(type) {
        case workflow.OutputEvent:
            kind := "Terminal"
            if event.IsIntermediate() {
                kind = "Intermediate"
            }
            fmt.Printf("%s output from %s: %v\n",
                kind, event.ExecutorID, event.Output)
        case workflow.ErrorEvent:
            return event.Error
        case workflow.ExecutorFailedEvent:
            return fmt.Errorf("executor %q failed: %w",
                event.ExecutorID, event.Error)
        }
    }
    return nil
}

WatchStream wartet weiter, wenn ein Workflow ausstehende externe Anforderungen aufweist. Behandeln Sie RequestInfoEvent, und senden Sie übereinstimmende Antworten über SendResponse, wenn Sie es für anforderungsgesteuerte Workflows verwenden.

Verwenden Sie stattdessen WatchUntilHalt, wenn der Consumer bei einer Pause zurückkehren muss, ohne auf eine Antwort zu warten. Nur ein Ereignisiterator kann eine Streamingausführung gleichzeitig verarbeiten. Siehe "Human-in-the-Loop"-Workflows für die Anfragebearbeitung und Wiederaufnahme.

Benutzerdefinierte Ereignisse

Mit benutzerdefinierten Ereignissen können Executoren domänenspezifische Signale während der Workflowausführung ausgeben, die auf die Anforderungen Ihrer Anwendung zugeschnitten sind. Im Folgenden finden Sie einige Beispiele für Anwendungsfälle:

• Nachverfolgen des Fortschritts – Melden Sie Zwischenschritte, damit Anrufer Statusaktualisierungen anzeigen können.
• Ausgeben der Diagnose – Anzeigen von Warnungen, Metriken oder Debuginformationen, ohne die Workflowausgabe zu ändern.
• Domänendaten weiterleiten – Strukturierte Nutzlasten (z. B. Datenbankschreibvorgänge, Toolaufrufe) in Echtzeit an Empfänger senden.

Definieren benutzerdefinierter Ereignisse

Definieren Sie ein benutzerdefiniertes Ereignis, indem Sie eine Unterklasse von WorkflowEvent erstellen. Der Basiskonstruktor akzeptiert eine optionale object? data Nutzlast, die über die Data Eigenschaft verfügbar gemacht wird.

using Microsoft.Agents.AI.Workflows;

// Simple event with a string payload
internal sealed class ProgressEvent(string step) : WorkflowEvent(step) { }

// Event with a structured payload
internal sealed class MetricsEvent(MetricsData metrics) : WorkflowEvent(metrics) { }

Erstellen Sie in Python benutzerdefinierte Ereignisse mithilfe der WorkflowEvent Klasse direkt mit einer benutzerdefinierten Typdiskriminatorzeichenfolge. Die Parameter type und data enthalten alle Informationen.

from agent_framework import WorkflowEvent

# Create a custom event with a custom type string and payload
event = WorkflowEvent(type="progress", data="Step 1 complete")

# Custom event with a structured payload
event = WorkflowEvent(type="metrics", data={"latency_ms": 42, "tokens": 128})

Note

Die Ereignistypen "started", "status"und "failed" sind für Framework-Lebenszyklusbenachrichtigungen reserviert. Wenn ein Executor versucht, einen dieser Typen auszugeben, wird das Ereignis ignoriert und eine Warnung protokolliert.

Definieren Sie ein benutzerdefiniertes Ereignis, indem Sie einen Typ erstellen, der die workflow.Event Schnittstelle implementiert. Die Data Methode gibt die Ereignisnutzlast zurück.

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

type ProgressEvent struct {
    Step string
}

func (e ProgressEvent) Data() any {
    return e.Step
}

var _ workflow.Event = ProgressEvent{}

Auslösen von benutzerdefinierten Ereignissen

Geben Sie benutzerdefinierte Ereignisse aus dem Nachrichtenhandler eines Executors aus, indem Sie die AddEventAsync für den IWorkflowContext aufrufen.

using Microsoft.Agents.AI.Workflows;

internal sealed class ProgressEvent(string step) : WorkflowEvent(step) { }

internal sealed partial class CustomExecutor() : Executor("CustomExecutor")
{
    [MessageHandler]
    private async ValueTask HandleAsync(string message, IWorkflowContext context)
    {
        await context.AddEventAsync(new ProgressEvent("Validating input"));

        // Executor logic...

        await context.AddEventAsync(new ProgressEvent("Processing complete"));
    }
}

Geben Sie benutzerdefinierte Ereignisse aus einem Handler aus, indem Sie add_event für den WorkflowContext aufrufen:

from agent_framework import (
    handler,
    Executor,
    WorkflowContext,
    WorkflowEvent,
)

class CustomExecutor(Executor):

    @handler
    async def handle(self, message: str, ctx: WorkflowContext[str]) -> None:
        await ctx.add_event(WorkflowEvent(type="progress", data="Validating input"))

        # Executor logic...

        await ctx.add_event(WorkflowEvent(type="progress", data="Processing complete"))

Geben Sie benutzerdefinierte Ereignisse aus dem Handler eines Executors aus, indem Sie die AddEvent für den workflow.Context aufrufen.

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

customExecutor := workflow.NewExecutor("CustomExecutor", func(ctx *workflow.Context, message string) error {
    if err := ctx.AddEvent(ProgressEvent{Step: "Validating input"}); err != nil {
        return err
    }

    // Executor logic...

    return ctx.AddEvent(ProgressEvent{Step: "Processing complete"})
}).Bind()

Benutzerdefinierte Ereignisse erfordern WithOutputFromnicht : AddEvent gibt sie direkt aus. Verwenden Sie YieldOutput und legen Sie eine Ausgabequelle fest, wenn Sie auch eine normale Workflowausgabe benötigen.

Verwenden von benutzerdefinierten Ereignissen

Verwenden Sie den Musterabgleich, um im Ereignisdatenstrom nach Ihrem benutzerdefinierten Ereignistyp zu filtern:

await foreach (WorkflowEvent evt in run.WatchStreamAsync())
{
    switch (evt)
    {
        case ProgressEvent progress:
            Console.WriteLine($"Progress: {progress.Data}");
            break;

        case WorkflowOutputEvent output:
            Console.WriteLine($"Done: {output.Data}");
            return;
    }
}

Filtern nach der benutzerdefinierten Typendiskriminatorzeichenfolge:

async for event in workflow.run(input_message, stream=True):
    if event.type == "progress":
        print(f"Progress: {event.data}")
    elif event.type == "output":
        print(f"Done: {event.data}")
        return

Verwenden Sie einen Typwechsel oder eine Typ assertion, um im Ereignisdatenstrom nach dem benutzerdefinierten Ereignistyp zu filtern:

for evt, err := range run.WatchStream(ctx) {
    if err != nil {
        return err
    }

    switch e := evt.(type) {
    case ProgressEvent:
        fmt.Printf("Progress: %v\n", e.Data())
    case workflow.OutputEvent:
        fmt.Printf("Done: %v\n", e.Output)
    case workflow.ErrorEvent:
        return e.Error
    case workflow.ExecutorFailedEvent:
        return fmt.Errorf("executor %q failed: %w", e.ExecutorID, e.Error)
    }
}

Nächste Schritte

Workflow-Generator & Ausführung

Verwandte Themen:

• Agents in Workflows
• Status-Management.
• Prüfpunkte & Fortsetzen
• Beobachtbarkeit