Partager via

Concepts fondamentaux des flux de travail Microsoft Agent Framework - Événements

Ce document fournit un aperçu détaillé du système d’événements des flux de travail dans Microsoft Agent Framework.

Aperçu

Il existe des événements intégrés qui fournissent une observabilité dans l’exécution du flux de travail.

Types d’événements intégrés

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

Consommation d’événements

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

Événements personnalisés

Les utilisateurs peuvent définir et émettre des événements personnalisés pendant l’exécution du flux de travail pour améliorer l’observabilité.

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

Étapes suivantes

  • Last updated on