Events

O sistema de eventos de fluxo de trabalho fornece observabilidade na execução do fluxo de trabalho. Os eventos são emitidos em pontos-chave durante a execução e podem ser consumidos em tempo real por meio do streaming.

Tipos de eventos incorporados

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

Quando os agentes usam ferramentas que exigem aprovação, RequestInfoEvent normalmente carrega um payload ToolApprovalRequestContent para chamadas de ferramenta que exigem aprovação humana. Consulte Human-in-the-Loop para obter detalhes sobre como lidar com esses 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

Quando os agentes usam ferramentas que exigem aprovação, request_info os eventos normalmente carregam um payload Content com type == "function_approval_request" para chamadas de ferramenta que requerem aprovação humana. Consulte Human-in-the-Loop para obter detalhes sobre como lidar com esses eventos.

Note

"output" e "intermediate" são os dois discriminadores de saída. Um executor designado como uma fonte de saída do terminal emite os eventos "output" (consumidos por WorkflowRunResult.get_outputs()). Um componente designado como uma fonte de saída intermediária emite "intermediate" eventos (consumidos por WorkflowRunResult.get_intermediate_outputs()). O tipo "data" é um alias obsoleto de "intermediate" e será removido em uma versão futura; prefira filtrar por "intermediate" em códigos novos.

Os eventos go implementam workflow.Event e expõem sua carga por meio de Data(). Use um switch de tipo para inspecionar seus campos tipados:

Tipo de evento	Carga útil e finalidade
`workflow.StartedEvent`	Inicia um ciclo de processamento de entrada. `Message` contém a entrada.
`workflow.OutputEvent`	Um executor designado produz um valor em `Output`. `Tags` identifica a saída intermediária.
`workflow.ErrorEvent`	Relata um erro em `Error`. `SubWorkflowID` indica um fluxo de trabalho aninhado quando está definido.
`workflow.WorkflowWarningEvent`	Relata um aviso em `Message`.
`workflow.ExecutorInvokedEvent`	Um executor inicia o processamento. `ExecutorID` identifica-o e `Message` contém sua entrada.
`workflow.ExecutorCompletedEvent`	Um executor conclui o processamento. `Result` contém seu resultado.
`workflow.ExecutorFailedEvent`	Um executor falhou. `ExecutorID` e `Error` identificam a falha.
`workflow.SuperStepStartedEvent`	Um superstep começa. `StepNumber` e `StartInfo` descrevem-no.
`workflow.SuperStepCompletedEvent`	Uma superetapa concluída. `CompletionInfo` descreve o trabalho pendente, solicitações e qualquer checkpoint.
`workflow.RequestInfoEvent`	`Request` contém uma solicitação externa que o host pode responder.
`workflow.RequestHaltEvent`	Um executor solicita uma interrupção. `Result` contém seu conteúdo, se houver.

O Go usa OutputEvent para saída terminal e intermediária. Chame IsIntermediate() para distingui-los; ele não usa os discriminadores separados "intermediate" e "output" do Python.

Os agentes hospedados também usam OutputEvent, com um *agent.ResponseUpdate payload ou *agent.Response payload quando os eventos correspondentes estão habilitados. O Go não define tipos separados AgentResponseUpdateEvent e AgentResponseEvent. Consulte executores do Agente para configuração.

Consumindo 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

Use um switch de tipo para eventos de fluxo de trabalho Go. Lide com erros de iterador e eventos de falha separadamente: receber uma saída não garante que o restante do fluxo de trabalho seja bem-sucedido.

Observar eventos

inproc.Default.Run avança para a próxima parada e acumula eventos. A execução pode pausar para entrada externa; uma parada não é necessariamente a conclusão.

A função a seguir imprime a atividade e a saída do executor, retorna 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 sua chamada anterior. Seu marcador avança quando você o chama, mesmo que você pare de iterar mais cedo. Use OutgoingEvents() para reproduzir todos os eventos acumulados em vez de chamar NewEvents() novamente.

Eventos de streaming

Use RunStreaming para observar eventos à medida que eles chegam. Essa 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 continua aguardando quando um fluxo de trabalho tem solicitações externas pendentes. Manipule RequestInfoEvent e envie respostas correspondentes por meio de SendResponse ao usá-la para fluxos de trabalho controlados por solicitação.

Em vez disso, use WatchUntilHalt quando o consumidor deve retornar em um ponto de pausa sem aguardar uma resposta. Somente um iterador de eventos pode consumir uma execução de streaming de cada vez. Consulte fluxos de trabalho com humano no circuito para tratamento e retomada de solicitações.

Eventos personalizados

Os eventos personalizados permitem que os executores emitam sinais específicos do domínio durante a execução do fluxo de trabalho adaptadas às necessidades do aplicativo. Alguns exemplos de caso de uso incluem:

Acompanhar o progresso – relatar etapas intermediárias para que os chamadores possam mostrar atualizações de status.
Emitir diagnósticos — apresentar avisos, métricas ou informações de depuração sem alterar o output do workflow.
Retransmissão de dados de domínio – envie por push os payloads estruturados (por exemplo, gravações de banco de dados, chamadas de ferramenta) para ouvintes em tempo real.

Definindo eventos personalizados

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

No Python, crie eventos personalizados usando a WorkflowEvent classe diretamente com uma cadeia de caracteres discriminatória de tipo personalizado. Os parâmetros type e data carregam todas as informações.

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

Os tipos de evento "started", "status", e "failed" são reservados para notificações do ciclo de vida do framework. Se um executor tentar emitir um desses tipos, o evento será ignorado e um aviso será registrado.

Defina um evento personalizado criando um tipo que implementa a workflow.Event interface. O Data método retorna o conteúdo 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{}

Emitindo eventos personalizados

Emita eventos personalizados a partir do manipulador de mensagens de um executor ao chamar 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 de um manipulador chamando add_event no 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 em um manipulador de executor chamando AddEvent em 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. Use YieldOutput e designe uma fonte de saída quando você também precisar de saída de fluxo de trabalho comum.

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

Filtrar por cadeia de caracteres 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

Use uma opção de tipo ou declaração de tipo para filtrar 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)
    }
}

Próximas Etapas

Criador de Workflow & Execução

Tópicos relacionados:

Comentários

Esta página foi útil?

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

Events

Tipos de eventos incorporados

Consumindo eventos

Observar eventos

Eventos de streaming

Eventos personalizados

Definindo eventos personalizados

Emitindo eventos personalizados

Consumindo eventos personalizados

Próximas Etapas

Comentários

Recursos adicionais