Events

Arbetsflödets händelsesystem ger insyn i genomförandet av arbetsflöden. Händelser genereras vid viktiga punkter under körningen och kan användas i realtid via strömning.

Inbyggda händelsetyper

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

Anmärkning

När agenter använder verktyg som kräver godkännande bär RequestInfoEvent vanligtvis en ToolApprovalRequestContent nyttolast för verktygsanrop som kräver mänskligt godkännande. Mer information om hur du hanterar dessa händelser finns i 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

Anmärkning

När agenter använder verktyg som kräver godkännande request_info, har händelser vanligtvis en nyttolast Content med type == "function_approval_request" för verktygsanrop som kräver mänskligt godkännande. Mer information om hur du hanterar dessa händelser finns i Human-in-the-Loop .

Anmärkning

"output" och "intermediate" är de två utgångsdiskriminatorerna. En exekverare som har utsetts till en källa för terminalutdata avger "output" händelser (som förbrukas av WorkflowRunResult.get_outputs()). En som anges som en mellanliggande utdatakälla genererar "intermediate" händelser (förbrukas av WorkflowRunResult.get_intermediate_outputs()). Typen "data" är ett inaktuellt alias för "intermediate" och tas bort i en framtida version, och föredrar att filtrera på "intermediate" i ny kod.

Go-händelser implementerar workflow.Event och exponerar nyttolasten via Data(). Använd en type switch för att inspektera deras typade fält:

Händelsetyp	Payload och syfte
`workflow.StartedEvent`	Startar en indatabearbetningscykel. `Message` innehåller indata.
`workflow.OutputEvent`	En utsedd exekverare ger ett värde i `Output`. `Tags` identifiering av mellanliggande utdata.
`workflow.ErrorEvent`	Rapportera ett fel i `Error`. `SubWorkflowID` identifierar ett kapslat arbetsflöde när den är angiven.
`workflow.WorkflowWarningEvent`	Rapporterar en varning i `Message`.
`workflow.ExecutorInvokedEvent`	En exekverare börjar bearbeta. `ExecutorID` identifierar det och `Message` innehåller dess indata.
`workflow.ExecutorCompletedEvent`	En exekverare slutför bearbetningen. `Result` innehåller resultatet.
`workflow.ExecutorFailedEvent`	En exekverare misslyckas. `ExecutorID` och `Error` identifierar felet.
`workflow.SuperStepStartedEvent`	En superstep startar. `StepNumber` och `StartInfo` beskriver det.
`workflow.SuperStepCompletedEvent`	Ett supersteg har slutförts. `CompletionInfo` beskriver väntande arbete, begäranden och valfri kontrollpunkt.
`workflow.RequestInfoEvent`	`Request` innehåller en extern begäran som värden kan besvara.
`workflow.RequestHaltEvent`	En exekverare begär en haltering. `Result` innehåller sin nyttolast, i förekommande fall.

Go använder OutputEvent för både terminal- och mellanliggande utdata. Anropa IsIntermediate() för att särskilja dem; den använder inte Pythons separata "output"- och "intermediate"-diskriminatorer.

Värdbaserade agenter använder OutputEventockså , med en *agent.ResponseUpdate eller *agent.Response nyttolast när motsvarande händelser är aktiverade. Go definierar inte separata AgentResponseEvent typer och AgentResponseUpdateEvent typer. Se Agent executors för konfiguration.

Använda händelser

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

Använd en type switch för Go-arbetsflödeshändelser. Hantera iteratorfel och felhändelser separat: att ta emot utdata garanterar inte att resten av arbetsflödet lyckas.

Observera händelser

inproc.Default.Run går vidare till nästa stoppunkt och ackumulerar händelser. Körningen kan pausas för externa indata. Ett stopp innebär inte nödvändigtvis att körningen är slutförd.

Följande funktion skriver ut körningsaktivitet och utdata, returnerar misslyckanden och stänger körningen:

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() returnerar de händelser som ackumulerats sedan föregående anrop. Bokmärket flyttas fram när du anropar det, även om du slutar iterera tidigt. Använd OutgoingEvents() för att spela upp alla ackumulerade händelser i stället för att anropa NewEvents() igen.

Strömmande händelser

Använd RunStreaming för att observera händelser när de anländer. Den här funktionen följer ett arbetsflöde som inte kräver externa indata:

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 fortsätter att vänta när ett arbetsflöde har väntande externa begäranden. Hantera RequestInfoEvent och skicka matchande svar via SendResponse när du använder det för begärandedrivna arbetsflöden.

Använd WatchUntilHalt i stället när konsumenten måste återupptas efter en paus utan att vänta på ett svar. Endast en händelseiterator kan bearbeta en strömmande körning i taget. Mer information om hantering och återupptagande finns i människa-i-loopen-arbetsflöden.

Anpassade händelser

Med anpassade händelser kan köre generera domänspecifika signaler under arbetsflödeskörningen som är skräddarsydda för programmets behov. Några exempel på användningsfall är:

Spåra förloppet – rapportera mellanliggande steg så att anropare kan visa statusuppdateringar.
Generera diagnostik – ytvarningar, mått eller felsökningsinformation utan att ändra arbetsflödets utdata.
Relädomändata – push-strukturerade nyttolaster (t.ex. databasskrivningar, verktygsanrop) till lyssnare i realtid.

Definiera anpassade händelser

Definiera en anpassad händelse genom att underklassa WorkflowEvent. Baskonstruktorn accepterar en valfri object? data nyttolast som exponeras via egenskapen Data .

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

I Python skapar du anpassade händelser med hjälp av WorkflowEvent klassen direkt med en diskriminerande sträng av anpassad typ. Parametrarna type och data bär all information.

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

Anmärkning

Händelsetyperna "started", "status"och "failed" är reserverade för ramverkets livscykelmeddelanden. Om en exekutor försöker emittera någon av dessa typer ignoreras händelsen och en varning loggas.

Definiera en anpassad händelse genom att skapa en typ som implementerar workflow.Event gränssnittet. Metoden Data returnerar händelsedata.

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

Generera anpassade händelser

Generera anpassade händelser från en exekutors meddelandehanterare genom att anropa AddEventAsync på 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"));
    }
}

Generera anpassade händelser från en hanterare genom att anropa add_event på 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"))

Generera anpassade händelser från en körhanterare genom att anropa AddEvent på 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()

Anpassade händelser kräver WithOutputFrominte : AddEvent genererar dem direkt. Använd YieldOutput och ange en utdatakälla när du också behöver vanliga arbetsflödesutdata.

Använda anpassade händelser

Använd mönstermatchning för att filtrera efter din anpassade händelsetyp i händelseströmmen:

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

Filtrera på den anpassade typdiskriminatorsträngen:

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

Använd en typväxel eller typkontroll för att filtrera efter din anpassade händelsetyp i händelseströmmen:

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ästa steg

Arbetsflödesbyggare och exekvering

Relaterade ämnen:

Feedback

Var den här sidan till hjälp?

Uppdaterades senast 2026-10-29

Events

Inbyggda händelsetyper

Använda händelser

Observera händelser

Strömmande händelser

Anpassade händelser

Definiera anpassade händelser

Generera anpassade händelser

Använda anpassade händelser

Nästa steg

Feedback

Ytterligare resurser