Events

El sistema de eventos de flujo de trabajo proporciona observabilidad en la ejecución del flujo de trabajo. Los eventos se emiten en puntos clave durante la ejecución y se pueden consumir en tiempo real a través del streaming.

Tipos de eventos integrados

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

Cuando los agentes utilizan herramientas que requieren aprobación, RequestInfoEvent suele llevar una carga útil ToolApprovalRequestContent para las llamadas a herramientas que requieren aprobación humana. Consulte Human-in-the-Loop para obtener más información sobre cómo controlar estos eventos.

# 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

Cuando los agentes utilizan herramientas que requieren aprobación, los eventos request_info suelen llevar una carga útil Content con type == "function_approval_request" para las llamadas a herramientas que requieren aprobación humana. Consulte Human-in-the-Loop para obtener más información sobre cómo controlar estos eventos.

Note

"output" y "intermediate" son los dos discriminadores de salida. Un ejecutor designado como fuente de salida del terminal emite eventos "output" (consumidos por WorkflowRunResult.get_outputs()). Una fuente designada como fuente de salida intermedia emite eventos "intermediate" (consumidos por WorkflowRunResult.get_intermediate_outputs()). El tipo "data" es un alias en desuso de "intermediate" y se eliminará en una versión futura; es preferible filtrar por "intermediate" en el código nuevo.

Los eventos de Go implementan workflow.Event y exponen su carga a través de Data(). Usa un switch de tipo para inspeccionar sus campos con tipo:

Tipo de evento	Carga útil y propósito
workflow.StartedEvent	Inicia un ciclo de procesamiento de entrada. Message contiene la entrada de datos.
workflow.OutputEvent	Un ejecutor designado produce un valor en Output. Tags identifica la salida intermedia.
workflow.ErrorEvent	Notifica un error en Error. SubWorkflowID identifica un flujo de trabajo anidado cuando se establece.
workflow.WorkflowWarningEvent	Notifica una advertencia en Message.
workflow.ExecutorInvokedEvent	Un ejecutor empieza a procesar. ExecutorID lo identifica y Message contiene su entrada.
workflow.ExecutorCompletedEvent	Un ejecutor completa el procesamiento. Result contiene su resultado.
workflow.ExecutorFailedEvent	Un ejecutor falla. ExecutorID e Error identifican el fallo.
workflow.SuperStepStartedEvent	Se inicia un superpaso. StepNumber y StartInfo lo describen.
workflow.SuperStepCompletedEvent	Se completa un superpaso. CompletionInfo describe el trabajo pendiente, las solicitudes y cualquier punto de control.
workflow.RequestInfoEvent	Request contiene una solicitud externa que el host puede responder.
workflow.RequestHaltEvent	Un ejecutor solicita una detención. Result contiene su payload, si existe.

Go usa OutputEvent tanto para la salida terminal como para la intermedia. Llame a IsIntermediate() para distinguirlas; no usa los discriminadores "output" y "intermediate" separados de Python.

Los agentes hospedados también usan OutputEvent, con una carga de *agent.ResponseUpdate o *agent.Response cuando se habilitan los eventos correspondientes. Go no define tipos AgentResponseEvent y AgentResponseUpdateEvent separados. Consulte Ejecutores de agentes para la configuración.

Consumo de eventos

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

Usa un switch de tipo para eventos de flujo de trabajo de Go. Gestione los errores del iterador y los eventos de error por separado: recibir una salida no garantiza que el resto del flujo de trabajo se complete correctamente.

Observar eventos

inproc.Default.Run avanza a la siguiente parada y acumula eventos. La ejecución puede pausarse para la entrada externa; una detención no es necesariamente la finalización.

La siguiente función imprime la actividad y la salida del ejecutor, devuelve fallos y cierra la ejecución:

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() devuelve los eventos acumulados desde su llamada anterior. Su marcador avanza al llamarlo, incluso si deja de iterar temprano. Use OutgoingEvents() para reproducir todos los eventos acumulados en lugar de volver a llamar a NewEvents().

Eventos de streaming

Usa RunStreaming para observar eventos a medida que llegan. Esta función sigue un flujo de trabajo que no requiere entrada externa:

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 sigue esperando cuando un flujo de trabajo tiene solicitudes externas pendientes. Gestione RequestInfoEvent y envíe las respuestas coincidentes mediante SendResponse cuando lo use para flujos de trabajo controlados por solicitudes.

Use WatchUntilHalt en lugar cuando el consumidor deba devolver en una pausa sin esperar una respuesta. Solo un iterador de eventos puede consumir una ejecución de streaming a la vez. Vea Flujos de trabajo con intervención humana para la gestión de solicitudes y la reanudación.

Eventos personalizados

Los eventos personalizados permiten a los ejecutores emitir señales específicas del dominio durante la ejecución del flujo de trabajo adaptadas a las necesidades de la aplicación. Algunos casos de uso de ejemplo son:

• Seguimiento del progreso : informe de los pasos intermedios para que los autores de llamadas puedan mostrar las actualizaciones de estado.
• Emitir diagnósticos — presentar advertencias, métricas o información de depuración sin cambiar la salida del proceso de trabajo.
• Datos del dominio de retransmisión: envía cargas útiles estructuradas (por ejemplo, escrituras en la base de datos, llamadas a herramientas) a los oyentes en tiempo real.

Definir eventos personalizados

Defina un evento personalizado mediante la subclase WorkflowEvent. El compilador base acepta una carga útil opcional object? data que se expone a través de la propiedad 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) { }

En Python, cree eventos personalizados mediante la WorkflowEvent clase directamente con una cadena de discriminador de tipo personalizado. Los type parámetros y data llevan toda la información.

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

Los tipos "started"de evento , "status"y "failed" están reservados para las notificaciones del ciclo de vida del marco. Si un ejecutor intenta emitir uno de estos tipos, se omite el evento y se registra una advertencia.

Defina un evento personalizado mediante la creación de un tipo que implemente la workflow.Event interfaz. El Data método devuelve la carga del 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{}

Emisión de eventos personalizados

Emita eventos personalizados desde el controlador de mensajes de un ejecutor llamando a AddEventAsync en :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"));
    }
}

Emita eventos personalizados desde un controlador llamando a add_event en :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"))

Emita eventos personalizados desde un controlador de ejecutor llamando a AddEvent en :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()

Los eventos personalizados no requieren WithOutputFrom: AddEvent los emite directamente. Use YieldOutput y designe un origen de salida cuando necesite también la salida ordinaria del flujo de trabajo.

Consumo de eventos personalizados

Use la coincidencia de patrones para filtrar el tipo de evento personalizado en el flujo de eventos:

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

Filtrar por la cadena discriminadora de tipo personalizada:

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

Use un modificador de tipo o una aserción de tipo para filtrar el tipo de evento personalizado en el flujo de eventos:

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

Pasos siguientes

Generador de flujos de trabajo y ejecución

Temas relacionados:

• Agentes en flujos de trabajo
• Administración de estados
• Puntos de control & Reanudación
• Observabilidad