Events

System zdarzeń przepływu pracy zapewnia obserwowalność wykonania przepływu pracy. Zdarzenia są emitowane w kluczowych punktach podczas wykonywania i mogą być odbierane w czasie rzeczywistym za pośrednictwem transmisji strumieniowej.

Wbudowane typy zdarzeń

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

Uwaga / Notatka

Gdy agenci używają narzędzi wymagających zatwierdzenia, RequestInfoEvent zwykle przenosi ToolApprovalRequestContent pakiet funkcji do wywołań narzędzi, które wymagają zatwierdzenia przez człowieka. Aby uzyskać szczegółowe informacje na temat obsługi tych zdarzeń, zobacz 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

Uwaga / Notatka

Gdy agenci używają narzędzi wymaganych do zatwierdzenia, request_info zdarzenia zwykle niosą Content ładunek z elementem type == "function_approval_request" dla wywołań narzędzi, które wymagają zatwierdzenia przez człowieka. Aby uzyskać szczegółowe informacje na temat obsługi tych zdarzeń, zobacz Human-in-the-Loop .

Uwaga / Notatka

"output" i "intermediate" to dwa dyskryminatory wyjściowe. Element wykonawczy oznaczony jako źródło wyjścia terminala emituje zdarzenia "output" (obsługiwane przez WorkflowRunResult.get_outputs()). Źródło oznaczone jako pośrednie źródło wyjściowe emituje zdarzenia "intermediate" (przetwarzane przez WorkflowRunResult.get_intermediate_outputs()). Typ "data" jest przestarzałym aliasem i "intermediate" zostanie usunięty w przyszłej wersji. Preferuj filtrowanie "intermediate" w nowym kodzie.

Zdarzenia języka Go implementują workflow.Event i uwidaczniają ich dane zdarzenia za pośrednictwem elementu Data(). Użyj przełącznika typu, aby sprawdzić pola typowane:

Typ zdarzenia	Ładunek danych i cel
`workflow.StartedEvent`	Uruchamia cykl przetwarzania danych wejściowych. `Message` zawiera dane wejściowe.
`workflow.OutputEvent`	Wyznaczony egzekutor zwraca wartość w `Output`. `Tags` identyfikuje pośrednie dane wyjściowe.
`workflow.ErrorEvent`	Zgłasza błąd w `Error`. `SubWorkflowID` identyfikuje zagnieżdżony przepływ pracy po ustawieniu.
`workflow.WorkflowWarningEvent`	Zgłasza ostrzeżenie w `Message`.
`workflow.ExecutorInvokedEvent`	Egzekutor rozpoczyna przetwarzanie. `ExecutorID` identyfikuje go i `Message` zawiera dane wejściowe.
`workflow.ExecutorCompletedEvent`	Egzekutor kończy przetwarzanie. `Result` zawiera wynik.
`workflow.ExecutorFailedEvent`	Moduł wykonawczy ulega awarii. `ExecutorID` i `Error` identyfikują błąd.
`workflow.SuperStepStartedEvent`	Rozpoczyna się superstep. `StepNumber` i `StartInfo` opisują to.
`workflow.SuperStepCompletedEvent`	Superkrok zostaje ukończony. `CompletionInfo` opisuje zadania oczekujące, żądania i dowolny punkt kontrolny.
`workflow.RequestInfoEvent`	`Request` zawiera żądanie zewnętrzne, na które host może odpowiedzieć.
`workflow.RequestHaltEvent`	Egzekutor żąda zatrzymania. `Result` zawiera jego ładunek danych, jeśli istnieje.

Język Go używa OutputEvent zarówno dla wyjścia terminalowego, jak i danych wyjściowych pośrednich. Wywołaj IsIntermediate(), aby je odróżnić; nie korzysta z oddzielnych "output" i "intermediate" dyskryminatorów w Pythonie.

Agenci hostowani używają również elementu OutputEvent z ładunkiem *agent.ResponseUpdate lub *agent.Response po włączeniu odpowiednich zdarzeń. Język Go nie definiuje oddzielnych typów AgentResponseEvent i AgentResponseUpdateEvent. Aby uzyskać informacje na temat konfiguracji, zobacz Egzekutory agenta.

Korzystanie ze zdarzeń

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

Użyj przełącznika typu dla zdarzeń przepływu pracy języka Go. Obsługuj błędy iteratora i zdarzenia awarii oddzielnie: odbieranie danych wyjściowych nie gwarantuje pomyślnego działania pozostałej części przepływu pracy.

Obserwowanie zdarzeń

inproc.Default.Run przechodzi do następnego przystanku i gromadzi zdarzenia. Uruchomienie może zostać wstrzymane dla danych wejściowych zewnętrznych; przerwanie niekoniecznie oznacza ukończenie.

Poniższa funkcja wyświetla aktywność i dane wyjściowe egzekutora, zwraca niepowodzenia i zamyka przebieg:

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() zwraca zdarzenia skumulowane od poprzedniego wywołania. Jej wskaźnik postępu przesuwa się przy każdym wywołaniu, nawet jeśli przestaniesz iterować wcześnie. Użyj polecenia OutgoingEvents() , aby odtworzyć wszystkie skumulowane zdarzenia zamiast wywołać NewEvents() ponownie.

Zdarzenia przesyłania strumieniowego

Służy RunStreaming do obserwowania zdarzeń w miarę ich napływania. Ta funkcja obsługuje przepływ pracy, który nie wymaga zewnętrznych danych wejściowych:

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 program ciągle czeka, gdy przepływ pracy ma oczekujące żądania zewnętrzne. Obsługuj RequestInfoEvent i wysyłaj pasujące odpowiedzi za pomocą metody SendResponse, gdy jej używasz w przepływach pracy opartych na żądaniach.

Zamiast tego należy użyć WatchUntilHalt, gdy odbiorca musi wrócić po wstrzymaniu bez oczekiwania na odpowiedź. W danym momencie tylko jeden iterator zdarzeń może przetwarzać przebieg strumieniowy. Zobacz Przepływy pracy z udziałem człowieka w pętli, aby zapoznać się z obsługą żądań i wznawianiem.

Zdarzenia niestandardowe

Zdarzenia niestandardowe pozwalają wykonawcom emitować sygnały specyficzne dla domeny podczas wykonywania przepływu pracy dostosowane do potrzeb aplikacji. Oto przykładowe przypadki użycia:

Śledzenie postępu — zgłaszanie kroków pośrednich, dzięki czemu osoby wywołujące mogą wyświetlać aktualizacje stanu.
Emituj dane diagnostyczne — przekaż ostrzeżenia, metryki lub informacje debugowania bez zmieniania danych wyjściowych przepływu pracy.
Przekazywanie danych domeny — przesyłanie struktur danych (np. zapisy bazy danych, wywołania narzędzi) do odbiorników w czasie rzeczywistym.

Definiowanie zdarzeń niestandardowych

Zdefiniuj zdarzenie niestandardowe według podklasy WorkflowEvent. Konstruktor bazowy akceptuje opcjonalny object? data ładunek udostępniany za pośrednictwem Data właściwości.

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

W Pythonie utwórz zdarzenia niestandardowe, używając klasy WorkflowEvent bezpośrednio z ciągiem znaków dyskryminującym typ niestandardowy. Parametry type i data zawierają wszystkie informacje.

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

Uwaga / Notatka

Typy zdarzeń "started", "status"i "failed" są zarezerwowane dla powiadomień cyklu życia platformy. Jeśli egzekutor próbuje emitować jeden z tych typów, zdarzenie jest ignorowane, a ostrzeżenie jest rejestrowane.

Zdefiniuj zdarzenie niestandardowe, tworząc typ, który implementuje workflow.Event interfejs. Metoda Data zwraca ładunek zdarzenia.

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

Emitowanie zdarzeń niestandardowych

Emituj zdarzenia niestandardowe z programu obsługi komunikatów egzekutora, wywołując AddEventAsync w 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"));
    }
}

Emituj niestandardowe zdarzenia za pomocą programu obsługi poprzez wywołanie 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"))

Wysyłaj niestandardowe zdarzenia z procedury obsługi executora, wywołując AddEvent w elemencie 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()

Zdarzenia niestandardowe nie wymagają WithOutputFrom: AddEvent emituje je bezpośrednio. Użyj YieldOutput i określ źródło wyjściowe, gdy potrzebujesz również zwykłych danych wyjściowych przepływu pracy.

Korzystanie ze zdarzeń niestandardowych

Skorzystaj z dopasowania wzorców, aby filtrować niestandardowy typ zdarzenia w strumieniu zdarzeń.

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

Filtruj ciąg rozróżniający typ niestandardowy:

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

Użyj przełącznika typów lub asercji typu, aby filtrować strumień zdarzeń pod kątem własnego typu zdarzenia:

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

Następne kroki

Konstruktor przepływu pracy i wykonywanie

Powiązane tematy:

Opinia

Czy ta strona była pomocna?

Ostatnia aktualizacja o 2026-10-29

Events

Wbudowane typy zdarzeń

Korzystanie ze zdarzeń

Obserwowanie zdarzeń

Zdarzenia przesyłania strumieniowego

Zdarzenia niestandardowe

Definiowanie zdarzeń niestandardowych

Emitowanie zdarzeń niestandardowych

Korzystanie ze zdarzeń niestandardowych

Następne kroki

Opinia

Dodatkowe zasoby