Events

Il sistema di eventi del flusso di lavoro fornisce l'osservabilità nell'esecuzione del flusso di lavoro. Gli eventi vengono emessi in punti chiave durante l'esecuzione e possono essere consumati in tempo reale tramite streaming.

Tipi di evento predefiniti

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

Annotazioni

Quando gli agenti usano strumenti che richiedono approvazione, RequestInfoEvent in genere trasporta un ToolApprovalRequestContent payload per le chiamate di strumenti che richiedono l'approvazione umana. Per informazioni dettagliate sulla gestione di questi eventi, vedere 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

Annotazioni

Quando gli agenti usano strumenti che richiedono l'approvazione, gli eventi request_info in genere contengono un payload Content con type == "function_approval_request" per le chiamate agli strumenti che richiedono l'approvazione umana. Per informazioni dettagliate sulla gestione di questi eventi, vedere Human-in-the-Loop .

Annotazioni

"output" e "intermediate" sono i due discriminatori di output. Un executor designato come origine di output terminale genera eventi "output" (utilizzati da WorkflowRunResult.get_outputs()). Uno designato come origine di output intermedia genera "intermediate" eventi (utilizzati da WorkflowRunResult.get_intermediate_outputs()). Il tipo "data" è un alias deprecato di "intermediate" e verrà rimosso in una versione futura; nel nuovo codice, preferire il filtro su "intermediate".

Gli eventi Go implementano workflow.Event ed espongono il payload tramite Data(). Usare un type switch per controllare i campi tipizzati:

Tipo di evento	Payload e scopo
workflow.StartedEvent	Avvia un ciclo di elaborazione dell'input. Message contiene l'input.
workflow.OutputEvent	Un executor designato restituisce un valore in Output. Tags identifica l'output intermedio.
workflow.ErrorEvent	Segnala un errore in Error. SubWorkflowID identifica un flusso di lavoro annidato se impostato.
workflow.WorkflowWarningEvent	Segnala un avviso in Message.
workflow.ExecutorInvokedEvent	Un executor avvia l'elaborazione. ExecutorID lo identifica e Message contiene il relativo input.
workflow.ExecutorCompletedEvent	Un executor completa l'elaborazione. Result contiene il risultato.
workflow.ExecutorFailedEvent	Un executor non riesce. ExecutorID e Error identificano il guasto.
workflow.SuperStepStartedEvent	Inizia un superstep. StepNumber e StartInfo lo descrivono.
workflow.SuperStepCompletedEvent	Un superstep viene completato. CompletionInfo descrive il lavoro in sospeso, le richieste e qualsiasi checkpoint.
workflow.RequestInfoEvent	Request contiene una richiesta esterna a cui l'host può rispondere.
workflow.RequestHaltEvent	Un executor richiede un arresto. Result contiene il payload, se presente.

Go usa OutputEvent per l'output terminale e intermedio. Chiama IsIntermediate() per distinguerli; non usa i discriminatori separati "output" e "intermediate" di Python.

Gli agenti ospitati usano anche OutputEvent, con un carico utile *agent.ResponseUpdate o *agent.Response quando gli eventi corrispondenti sono abilitati. Go non definisce tipi AgentResponseUpdateEvent e AgentResponseEvent separati. Per la configurazione, vedere Executor dell'agente.

Utilizzo degli eventi

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

Usare un'opzione di tipo per gli eventi del flusso di lavoro Go. Gestire separatamente gli errori dell'iteratore e gli eventi di errore: la ricezione di un output non garantisce che il resto del flusso di lavoro abbia esito positivo.

Osservare gli eventi

inproc.Default.Run avanza fino all'arresto successivo e accumula eventi. L'esecuzione può essere sospesa per l'input esterno; un arresto non equivale necessariamente al completamento.

La funzione seguente stampa l'attività e l'output dell'executor, restituisce i fallimenti e chiude l'esecuzione:

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() restituisce gli eventi accumulati dalla chiamata precedente. Il segnalibro avanza quando viene chiamato, anche se si interrompe l'iterazione in anticipo. Usare OutgoingEvents() per riprodurre tutti gli eventi accumulati invece di chiamare NewEvents() di nuovo.

Eventi di streaming

Usare RunStreaming per osservare gli eventi non appena arrivano. Questa funzione segue un flusso di lavoro che non richiede l'input esterno:

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 rimane in attesa quando un flusso di lavoro ha richieste esterne in sospeso. Gestire RequestInfoEvent e inviare risposte corrispondenti tramite SendResponse quando lo si usa per i flussi di lavoro basati sulle richieste.

Usare WatchUntilHalt invece quando il consumer deve tornare durante una pausa senza attendere una risposta. Un solo iteratore di eventi può consumare un'esecuzione di streaming alla volta. Vedere Flussi di lavoro con intervento umano nel ciclo per la gestione delle richieste e la ripresa.

Eventi personalizzati

Gli eventi personalizzati consentono agli executor di generare segnali specifici del dominio durante l'esecuzione del flusso di lavoro in base alle esigenze dell'applicazione. Alcuni casi d'uso di esempio includono:

• Tenere traccia dello stato : segnalare i passaggi intermedi in modo che i chiamanti possano visualizzare gli aggiornamenti dello stato.
• Genera diagnostica : visualizza avvisi, metriche o informazioni di debug senza modificare l'output del flusso di lavoro.
• Dati del dominio di inoltro — trasmettere payload strutturati (ad esempio, scritture di database, chiamate di strumenti) agli ascoltatori in tempo reale.

Definizione di eventi personalizzati

Definire un evento personalizzato sottoclassando WorkflowEvent. Il costruttore di base accetta un payload facoltativo object? data, esposto tramite la proprietà 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) { }

In Python creare eventi personalizzati usando la WorkflowEvent classe direttamente con una stringa discriminatoria del tipo personalizzata. I type parametri e data contengono tutte le informazioni.

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

Annotazioni

I tipi di "started"evento , "status"e "failed" sono riservati per le notifiche del ciclo di vita del framework. Se un esecutore tenta di generare uno di questi tipi, l'evento viene ignorato e viene registrato un messaggio di avviso.

Definire un evento personalizzato creando un tipo che implementa l'interfaccia workflow.Event . Il Data metodo restituisce il payload dell'evento.

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

Emissione di eventi personalizzati

Generare eventi personalizzati dal gestore di messaggi di un executor chiamando AddEventAsync su 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"));
    }
}

Generare eventi personalizzati da un gestore chiamando add_event su 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"))

Generare eventi personalizzati da un gestore executor chiamando AddEvent su 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()

Gli eventi personalizzati non richiedono WithOutputFrom: AddEvent li emette direttamente. Usare YieldOutput e designare un'origine di output quando è necessario anche l'output normale del flusso di lavoro.

Utilizzo di eventi personalizzati

Usare i criteri di ricerca per filtrare il tipo di evento personalizzato nel flusso di eventi:

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

Filtrare in base alla stringa discriminatoria del tipo personalizzato:

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

Usare uno switch di tipo o un'asserzione di tipo per filtrare il tipo di evento personalizzato nel flusso di eventi:

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

Passaggi successivi

Generatore di flussi di lavoro ed esecuzione

Argomenti correlati:

• Agenti nei flussi di lavoro
• Gestione dello Stato
• Checkpoint & Ripresa
• Osservabilità