Události

Systém událostí pracovního postupu poskytuje pozorovatelnost při provádění pracovního postupu. Události jsou vyvolávány v klíčových bodech během provádění a dají se využívat v reálném čase přes streamování.

Předdefinované typy událostí

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

Poznámka:

Když agenti používají nástroje vyžadující schválení, RequestInfoEvent obvykle nese ToolApprovalRequestContent datový obsah pro volání nástrojů, které vyžadují lidské schválení. Podrobnosti o zpracování těchto událostí najdete v tématu Human-in-the-Loop .

# 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

Poznámka:

Když agenti používají nástroje s vyžadovaným schválením, request_info události obvykle nesou Content zatížení s type == "function_approval_request" pro volání nástrojů, které vyžadují lidské schválení. Podrobnosti o zpracování těchto událostí najdete v tématu Human-in-the-Loop .

Poznámka:

"output" a "intermediate" jsou to dva diskriminátory výstupu. Vykonavatel určený jako zdroj výstupu terminálu vysílá události "output" (které zpracovává WorkflowRunResult.get_outputs()). Ten, který je označen jako mezilehlý výstupní zdroj, vysílá "intermediate" události (které zpracovává WorkflowRunResult.get_intermediate_outputs()). Typ "data" je zastaralý alias pro "intermediate" a bude odebrán v budoucí verzi. Preferujte filtrování "intermediate" v novém kódu.

Události Go implementují workflow.Event a zpřístupňují svou datovou část prostřednictvím Data(). Pomocí přepínače typu zkontrolujte jejich typová pole:

Typ události	Payload a účel
workflow.StartedEvent	Spustí cyklus zpracování vstupu. Message obsahuje vstup.
workflow.OutputEvent	Určený vykonavatel vrací hodnotu v Output. Tags identifikuje mezivýstup.
workflow.ErrorEvent	Hlásí chybu v souboru Error. SubWorkflowID identifikuje vnořený pracovní postup, je-li nastaven.
workflow.WorkflowWarningEvent	Nahlásí upozornění v souboru Message.
workflow.ExecutorInvokedEvent	Executor začne zpracovávat. ExecutorID identifikuje ho a Message obsahuje jeho vstup.
workflow.ExecutorCompletedEvent	Vykonavatel dokončí zpracování. Result obsahuje výsledek.
workflow.ExecutorFailedEvent	Executor selže. ExecutorID a Error identifikují selhání.
workflow.SuperStepStartedEvent	Začíná superstep. StepNumber a StartInfo ho popisují.
workflow.SuperStepCompletedEvent	superstep je dokončen. CompletionInfo popisuje čekající práci, požadavky a jakýkoli kontrolní bod.
workflow.RequestInfoEvent	Request obsahuje externí požadavek, na který může hostitel odpovědět.
workflow.RequestHaltEvent	Vykonavatel požádá o zastavení. Result obsahuje svou datovou část, pokud existuje.

Go používá OutputEvent pro terminálový i průběžný výstup. Zavolejte IsIntermediate(), abyste je rozlišili; nepoužívá samostatné "output" a "intermediate" oddělovače jazyka Python.

Hostovaní agenti také používají OutputEvent, s datovou částí *agent.ResponseUpdate nebo *agent.Response, když jsou povoleny příslušné události. Go nedefinuje samostatné AgentResponseEvent typy a AgentResponseUpdateEvent typy. Informace o konfiguraci najdete v Agent executors.

Zpracovávání událostí

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

Pro události workflow Go použijte přepínač typu. Zpracovávejte chyby iterátoru a události selhání odděleně: příjem výstupu nezaručuje, že zbytek pracovního postupu bude úspěšný.

Sledování událostí

inproc.Default.Run přejde na další zastávku a nashromáždí události. Běh lze pozastavit pro externí vstup; zastavení nutně neznamená dokončení.

Následující funkce vytiskne aktivitu a výstup executoru, vrátí selhání a uzavře run:

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() vrátí události kumulované od předchozího volání. Jeho ukazatel se posune, když ji zavoláte, i když přestanete iterovat brzy. Použijte OutgoingEvents() k přehrání všech kumulovaných událostí místo opětovného volání NewEvents().

Události streamování

Slouží RunStreaming k pozorování událostí, jak přicházejí. Tato funkce se řídí pracovním postupem, který nevyžaduje externí vstup:

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 zůstává čekat, když pracovní postup čeká na vyřízení externích požadavků. Zpracovávejte RequestInfoEvent a odesílejte odpovídající odpovědi prostřednictvím SendResponse, když jej používáte pro pracovní postupy řízené požadavky.

Místo toho použijte WatchUntilHalt, když se klient musí pokračovat po pozastavení bez čekání na odpověď. Pouze jeden iterátor událostí může najednou zpracovávat streamované spuštění. Viz pracovní postupy se zapojením člověka pro zpracování požadavků a pokračování.

Vlastní události

Vlastní události umožňují exekutorům generovat signály specifické pro doménu během provádění pracovního postupu přizpůsobené potřebám vaší aplikace. Mezi příklady případů použití patří:

• Sledovat průběh – hlásit přechodné kroky, aby volající mohli zobrazovat aktualizace stavu.
• Emitovat diagnostiku – zobrazit upozornění, metriky nebo informace o ladění bez změny výstupu pracovního postupu.
• Přenosová data domény – odesílání strukturovaných datových zátěží (např. zápisy do databáze, volání nástrojů) posluchačům v reálném čase.

Definování vlastních událostí

Definujte vlastní událost podtřídou WorkflowEvent. Základní konstruktor přijímá volitelnou object? data datovou část, která je vystavena prostřednictvím Data vlastnosti.

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

V jazyce Python vytvořte vlastní události přímo pomocí třídy s vlastním řetězcem rozlišení typu. Parametry type a data obsahují všechny informace.

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

Poznámka:

Typy událostí "started", "status" a "failed" jsou vyhrazeny pro oznámení životního cyklu rámce. Pokud se exekutor pokusí vygenerovat jeden z těchto typů, událost se ignoruje a zaprotokoluje se upozornění.

Definujte vlastní událost vytvořením typu, který implementuje workflow.Event rozhraní. Metoda Data vrátí datovou část události.

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

Generování vlastních událostí

Generování vlastních událostí z obslužné rutiny zprávy exekutoru voláním AddEventAsync na 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"));
    }
}

Generování vlastních událostí z obslužné rutiny voláním add_event na 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"))

Generování vlastních událostí z obslužné rutiny exekutoru voláním AddEvent na 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()

Vlastní události nevyžadují WithOutputFrom: AddEvent je emitují přímo. Použijte YieldOutput a vyberte zdroj výstupu, pokud potřebujete také běžný výstup pracovního postupu.

Využívání vlastních událostí

K filtrování vlastního typu události ve streamu událostí použijte porovnávání vzorů:

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

Filtr na řetězec diskriminátoru vlastního typu:

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

Pomocí přepínače typu nebo typové aserce filtrujte ve streamu událostí váš vlastní typ události:

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

Další kroky

Tvůrce pracovních postupů a provádění

Související témata:

• Agenti v pracovních postupech
• Správa stavu
• Kontrolní body a pokračování
• Pozorovatelnost