Conceitos principais de fluxos de trabalho do Microsoft Agent Framework - Eventos

Este documento fornece uma visão detalhada do sistema de eventos de fluxos de trabalho no Microsoft Agent Framework.

Visão geral

Há eventos internos que fornecem observabilidade na execução do fluxo de trabalho.

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
AgentRunResponseEvent   // An agent run produces output
AgentRunUpdateEvent     // An agent run produces a streaming update

// Superstep events
SuperStepStartedEvent   // Superstep begins
SuperStepCompletedEvent // Superstep completes

// Request events
RequestInfoEvent        // A request is issued

# Workflow lifecycle events
WorkflowStartedEvent    # Workflow execution begins
WorkflowOutputEvent     # Workflow produces an output
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
AgentRunEvent           # An agent run produces output
AgentRunUpdateEvent     # An agent run produces a streaming update

# Superstep events
SuperStepStartedEvent   # Superstep begins
SuperStepCompletedEvent # Superstep completes

# Request events
RequestInfoEvent        # A request is issued

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 (
    ExecutorCompleteEvent,
    ExecutorInvokeEvent,
    WorkflowOutputEvent,
    WorkflowErrorEvent,
)

async for event in workflow.run_stream(input_message):
    match event:
        case ExecutorInvokeEvent() as invoke:
            print(f"Starting {invoke.executor_id}")
        case ExecutorCompleteEvent() as complete:
            print(f"Completed {complete.executor_id}: {complete.data}")
        case WorkflowOutputEvent() as output:
            print(f"Workflow produced output: {output.data}")
            return
        case WorkflowErrorEvent() as error:
            print(f"Workflow error: {error.exception}")
            return

Eventos personalizados

Os usuários podem definir e emitir eventos personalizados durante a execução do fluxo de trabalho para melhorar a observabilidade.

using Microsoft.Agents.AI.Workflows;
using Microsoft.Agents.AI.Workflows.Reflection;

internal sealed class CustomEvent(string message) : WorkflowEvent(message) { }

internal sealed class CustomExecutor() : ReflectingExecutor<CustomExecutor>("CustomExecutor"), IMessageHandler<string>
{
    public async ValueTask HandleAsync(string message, IWorkflowContext context)
    {
        await context.AddEventAsync(new CustomEvent($"Processing message: {message}"));
        // Executor logic...
    }
}

from agent_framework import (
    handler,
    Executor,
    WorkflowContext,
    WorkflowEvent,
)

class CustomEvent(WorkflowEvent):
    def __init__(self, message: str):
        super().__init__(message)

class CustomExecutor(Executor):

    @handler
    async def handle(self, message: str, ctx: WorkflowContext[str]) -> None:
        await ctx.add_event(CustomEvent(f"Processing message: {message}"))
        # Executor logic...

Próximas Etapas

• Saiba como usar agentes em fluxos de trabalho para criar fluxos de trabalho inteligentes.
• Saiba como usar fluxos de trabalho como agentes.
• Saiba como lidar com solicitações e respostas em fluxos de trabalho.
• Saiba como gerenciar o estado em fluxos de trabalho.
• Saiba como criar pontos de verificação e retomar a partir deles.
• Aprenda a monitorizar fluxos de trabalho.
• Aprenda sobre isolamento de estado nos fluxos de trabalho.
• Aprenda a visualizar fluxos de trabalho.