Evenementen

Het werkstroomgebeurtenissysteem biedt waarneembaarheid in de uitvoering van de werkstroom. Gebeurtenissen worden tijdens de uitvoering op belangrijke punten verzonden en kunnen in realtime worden gebruikt via streaming.

Ingebouwde gebeurtenistypen

// 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

Opmerking

Wanneer agents gebruikmaken van hulpprogramma's die goedkeuring vereisen, voert RequestInfoEvent doorgaans een ToolApprovalRequestContent payload uit voor hulpprogramma-aanroepen waarvoor menselijke goedkeuring is vereist. Zie Human-in-the-Loop voor meer informatie over het afhandelen van deze gebeurtenissen.

# 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

Opmerking

Wanneer agents hulpprogramma's die goedkeuring vereisen gebruiken, bevatten request_info gebeurtenissen doorgaans een Content payload met type == "function_approval_request" voor toolaanroepen die menselijke goedkeuring vereisen. Zie Human-in-the-Loop voor meer informatie over het afhandelen van deze gebeurtenissen.

Opmerking

"output" en "intermediate" zijn de twee uitvoerdiscriminatoren. Een executor die is aangewezen als een bron voor terminaluitvoer genereert "output"-gebeurtenissen (verwerkt door WorkflowRunResult.get_outputs()). Eén aangewezen als een tussenliggende uitvoerbron verzendt "intermediate" gebeurtenissen (verbruikt door WorkflowRunResult.get_intermediate_outputs()). Het "data" type is een afgeschafte alias voor "intermediate" en wordt verwijderd in een toekomstige versie. Filter liever op "intermediate" in nieuwe code.

Go-gebeurtenissen implementeren workflow.Event en maken hun payload beschikbaar via Data(). Gebruik een typeschakelaar om de getypeerde velden te controleren:

Gebeurtenistype	Payload en doel
workflow.StartedEvent	Hiermee start u een invoerverwerkingscyclus. Message bevat de invoer.
workflow.OutputEvent	Een aangewezen executor levert een waarde op in Output. Tags identificeert tussenliggende uitvoer.
workflow.ErrorEvent	Rapporteert een fout in Error. SubWorkflowID identificeert een geneste werkstroom wanneer deze is ingesteld.
workflow.WorkflowWarningEvent	Meldt een waarschuwing in Message.
workflow.ExecutorInvokedEvent	Een executor begint met de verwerking. ExecutorID identificeert het en Message bevat de invoer.
workflow.ExecutorCompletedEvent	Een executor voert de verwerking uit. Result bevat zijn resultaat.
workflow.ExecutorFailedEvent	De executor faalt. ExecutorID en Error identificeren de fout.
workflow.SuperStepStartedEvent	Een superstep begint. StepNumber en StartInfo beschrijven het.
workflow.SuperStepCompletedEvent	Een superstep is voltooid. CompletionInfo beschrijft in behandeling zijnde werk, verzoeken en een ijkpunt.
workflow.RequestInfoEvent	Request bevat een extern verzoek dat de host kan beantwoorden.
workflow.RequestHaltEvent	Een executor vraagt een stop aan. Result bevat de payload, indien van toepassing.

Go gebruikt OutputEvent voor zowel terminal- als tussenliggende uitvoer. Roep IsIntermediate() aan om ze te onderscheiden; het maakt geen gebruik van Python's afzonderlijke "output" en "intermediate" discriminatoren.

Gehoste agents gebruiken OutputEventook , met een *agent.ResponseUpdate of *agent.Response payload wanneer de bijbehorende gebeurtenissen zijn ingeschakeld. Go definieert geen afzonderlijke AgentResponseEvent en AgentResponseUpdateEvent typen. Zie Agent-executors voor configuratie.

Gebeurtenissen gebruiken

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

Gebruik een type switch voor workflowgebeurtenissen in Go. Los iteratorfouten en foutevenementen af: het ontvangen van een uitvoer garandeert niet dat de rest van de workflow slaagt.

Gebeurtenissen observeren

inproc.Default.Run gaat verder naar de volgende onderbreking en verzamelt gebeurtenissen. De run kan worden gepauzeerd voor externe invoer; een stop is niet noodzakelijkerwijs voltooiing.

Met de volgende functie worden de executoractiviteit en uitvoer afgedrukt, worden mislukte runs geretourneerd en wordt de run afgesloten:

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() geeft de gebeurtenissen terug die zijn verzameld sinds de vorige aanroep ervan. De bladwijzer gaat verder wanneer u deze aanroept, zelfs als u vroegtijdig stopt met itereren. Gebruik OutgoingEvents() om alle verzamelde gebeurtenissen opnieuw af te spelen in plaats van NewEvents() opnieuw aan te roepen.

Streaming-gebeurtenissen

Gebruik RunStreaming om gebeurtenissen te observeren wanneer ze binnenkomen. Met deze functie wordt een workflow bewaakt waarvoor geen externe invoer is vereist:

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 blijft wachten wanneer een werkstroom externe aanvragen in behandeling heeft. Verwerk RequestInfoEvent en verzend overeenkomende antwoorden via SendResponse wanneer u deze gebruikt voor requestgestuurde workflows.

Gebruik WatchUntilHalt in plaats daarvan wanneer de consumer tijdens een pauze moet terugkeren zonder te wachten op een antwoord. Slechts één gebeurtenis-iterator kan tegelijkertijd een streaming-uitvoering verwerken. Zie Human-in-the-loop-werkstromen voor de afhandeling van aanvragen en het hervatten van werkstromen.

Aangepaste gebeurtenissen

Met aangepaste gebeurtenissen kunnen uitvoerders domeinspecifieke signalen verzenden tijdens de uitvoering van de werkstroom, afgestemd op de behoeften van uw toepassing. Enkele voorbeelden van gebruiksvoorbeelden zijn:

• Voortgang bijhouden : tussenliggende stappen rapporteren, zodat bellers statusupdates kunnen weergeven.
• Diagnostische gegevens verzenden : waarschuwingen, metrische gegevens of foutopsporingsgegevens zonder de uitvoer van de werkstroom te wijzigen.
• Relaydomeingegevens — push gestructureerde payloads (bijvoorbeeld databaseschrijfbewerkingen, tool-aanroepen) naar listeners in realtime.

Aangepaste gebeurtenissen definiëren

Definieer een aangepaste gebeurtenis door subklassen te definiëren WorkflowEvent. De basisconstructor accepteert een optionele object? data nettolading die wordt weergegeven via de Data eigenschap.

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) { }

Maak in Python aangepaste gebeurtenissen door direct de WorkflowEvent klasse te gebruiken, samen met een discriminator-tekenreeks voor aangepaste typen. De type en data parameters bevatten alle informatie.

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

Opmerking

De gebeurtenistypen "started", "status"en "failed" zijn gereserveerd voor meldingen over de levenscyclus van frameworks. Als een uitvoerder probeert een van deze typen uit te zenden, wordt de gebeurtenis genegeerd en wordt een waarschuwing vastgelegd.

Definieer een aangepaste gebeurtenis door een type te maken waarmee de workflow.Event interface wordt geïmplementeerd. De methode Data geeft de payload van de gebeurtenis terug.

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{}

Aangepaste gebeurtenissen verzenden

Aangepaste gebeurtenissen verzenden vanuit de uitvoerder's berichthandler door AddEventAsync aan te roepen op IWorkflowContext:

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

Stuur aangepaste gebeurtenissen vanuit een handler door add_event aan te roepen op WorkflowContext.

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

Aangepaste gebeurtenissen genereren vanuit een handler van een executor door AddEvent aan te roepen op de workflow.Context:

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

Aangepaste gebeurtenissen vereisen WithOutputFromniet: AddEvent verzendt ze rechtstreeks. Gebruik YieldOutput en wijs een uitvoerbron aan wanneer u ook gewone werkstroomuitvoer nodig hebt.

Aangepaste gebeurtenissen gebruiken

Gebruik patroonkoppeling om te filteren op uw aangepaste gebeurtenistype in de gebeurtenisstroom:

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;
    }
}

Filter op de discriminator-tekenreeks voor het aangepaste type:

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

Gebruik een typeswitch of typebevestiging om te filteren op uw aangepaste gebeurtenistype in de gebeurtenisstroom:

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

Volgende stappen 

Werkstroombouwer en uitvoering

Verwante onderwerpen:

• Agents in werkprocessen
• Statusbeheer
• Controlepunten en hervatten
• Observatievermogen