Events

O sistema de eventos de workflow proporciona observabilidade na execução do workflow. Os eventos são emitidos em pontos-chave durante a execução e podem ser consumidos em tempo real através de 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

Observação

Quando os agentes utilizam ferramentas que requerem aprovação, RequestInfoEvent normalmente transporta um ToolApprovalRequestContent pacote de dados para execuções de ferramentas que precisam de aprovação humana. Consulte Human-in-the-Loop para detalhes sobre como lidar com estes 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

Observação

Quando os agentes utilizam ferramentas que exigem aprovação, request_info os eventos normalmente incluem uma Content carga útil nas type == "function_approval_request" chamadas de ferramentas que requerem aprovação humana. Consulte Human-in-the-Loop para detalhes sobre como lidar com estes eventos.

Observação

"output" e "intermediate" são os dois discriminadores de saída. Um executor designado como fonte de saída terminal emite "output" eventos (consumidos por WorkflowRunResult.get_outputs()). Uma designada como fonte de saída intermédia emite "intermediate" eventos (consumidos por WorkflowRunResult.get_intermediate_outputs()). O tipo "data" é um nome alternativo obsoleto para "intermediate" e será removido numa versão futura; prefira filtrar por "intermediate" em código novo.

Os eventos Go implementam workflow.Event e expõem a sua carga útil através de Data(). Use uma instrução switch de tipo para inspecionar os seus campos tipados:

Tipo de evento	Payload e propósito
`workflow.StartedEvent`	Inicia um ciclo de processamento de entrada. `Message` contém os dados de entrada.
`workflow.OutputEvent`	Um executor designado produz um valor em `Output`. `Tags` identifica a saída intermédia.
`workflow.ErrorEvent`	Comunica um erro em `Error`. `SubWorkflowID` identifica um fluxo de trabalho aninhado quando ativado.
`workflow.WorkflowWarningEvent`	Apresenta um aviso em `Message`.
`workflow.ExecutorInvokedEvent`	Um executor inicia o processamento. `ExecutorID` identifica-o e `Message` contém os seus dados de entrada.
`workflow.ExecutorCompletedEvent`	Um executor conclui o processamento. `Result` contém o seu resultado.
`workflow.ExecutorFailedEvent`	Um executante falha. `ExecutorID` e `Error` identificam a falha.
`workflow.SuperStepStartedEvent`	Inicia-se um superstep. `StepNumber` e `StartInfo` descrevem-no.
`workflow.SuperStepCompletedEvent`	Um superpasso é concluído. `CompletionInfo` Descreve trabalho pendente, pedidos e qualquer ponto de controlo.
`workflow.RequestInfoEvent`	`Request` contém um pedido externo ao qual o host pode responder.
`workflow.RequestHaltEvent`	Um executor solicita a paragem. `Result` contém a sua carga útil, se houver.

Go usa OutputEvent tanto para saída terminal como intermédia. Chama IsIntermediate() para os distinguir; não usa os discriminadores "output" e "intermediate" separados do Python.

Os agentes alojados também usam OutputEvent, com um *agent.ResponseUpdate payload ou *agent.Response quando os eventos correspondentes estão ativados. Go não define tipos separados AgentResponseEventAgentResponseUpdateEvent . Consulte executores de agente para configuração.

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

Utilize um switch de tipo para eventos de fluxo de trabalho Go. Trate os erros do iterador e os eventos de falha separadamente: receber um resultado não garante que o resto do fluxo de trabalho tenha sucesso.

Observar eventos

inproc.Default.Run avança para a próxima paragem e acumula eventos. A execução pode fazer uma pausa para input externo; uma paragem não é necessariamente a conclusão.

A seguinte função imprime a atividade e a saída do executor, devolve falhas e fecha a execução:

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() retorna os eventos acumulados desde a chamada anterior. O seu marcador avança quando o chamas, mesmo que pares de iterar cedo. Use OutgoingEvents() para repetir todos os eventos acumulados em vez de chamar NewEvents() novamente.

Eventos em streaming

Usa RunStreaming para observar os acontecimentos à medida que chegam. Esta função segue um fluxo de trabalho que não requer 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 Fica sempre à espera quando um fluxo de trabalho tem pedidos externos pendentes. Trate RequestInfoEvent e envie respostas correspondentes através de SendResponse quando o usar para fluxos de trabalho condicionados por pedidos.

Use WatchUntilHalt em vez disso quando o consumidor tiver de retomar num ponto de pausa sem esperar por uma resposta. Apenas um iterador de eventos pode consumir uma execução de streaming de cada vez. Consulte os fluxos de trabalho Human-in-the-loop para o tratamento e retoma de pedidos.

Eventos personalizados

Os eventos personalizados permitem aos executores emitir sinais específicos do domínio durante a execução do fluxo de trabalho, adaptados às necessidades da sua aplicação. Alguns exemplos de casos de uso incluem:

Acompanhe o progresso — reporte os passos intermédios para que os chamadores possam mostrar atualizações de estado.
Emita diagnósticos — apresente avisos, métricas ou informações de depuração sem alterar a saída do processo de trabalho.
Dados do domínio de retransmissão — enviar cargas úteis estruturadas (por exemplo, operações de escrita em bases de dados, chamadas de ferramentas) aos destinatários em tempo real.

Definição de Eventos Personalizados

Defina um evento personalizado subclassificando WorkflowEvent. O construtor base aceita um payload opcional object? data que é exposto pela propriedade 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) { }

Em Python, cria eventos personalizados usando a WorkflowEvent classe diretamente com uma cadeia discriminadora de tipo personalizada. Os type parâmetros e data transportam toda a informação.

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

Observação

Os tipos "started"de eventos , "status", e "failed" estão reservados para notificações do ciclo de vida da estrutura. Se um executor tentar emitir um destes tipos, o evento é ignorado e um aviso é registado.

Defina um evento personalizado criando um tipo que implemente a workflow.Event interface. O Data método devolve a carga útil do 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{}

Emissão de Eventos Personalizados

Emita eventos personalizados a partir do manipulador de mensagens de um executor chamando AddEventAsync no 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 a partir de um handler chamando add_event o 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"))

Emitir eventos personalizados a partir de um processador do executor ao chamar AddEvent no 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()

Eventos personalizados não exigem WithOutputFrom: AddEvent emite-os diretamente. Utilize YieldOutput e designe uma fonte de saída quando também precisar de saída de fluxo de trabalho comum.

Consumir Eventos Personalizados

Use a correspondência de padrões para filtrar o tipo de evento personalizado no fluxo 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;
    }
}

Filtro na cadeia discriminadora de tipo personalizado:

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

Utilize um type switch ou uma asserção de tipo para filtrar o seu tipo de evento personalizado no fluxo 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)
    }
}

Passos seguintes

Construtor e Execução de Fluxos de Trabalho

Tópicos relacionados:

Comentários

Esta página foi útil?

Última atualização a 2026-10-29

Events

Tipos de eventos integrados

Consumo de eventos

Observar eventos

Eventos em streaming

Eventos personalizados

Definição de Eventos Personalizados

Emissão de Eventos Personalizados

Consumir Eventos Personalizados

Passos seguintes

Comentários

Recursos adicionais