Events

Sistem acara alur kerja memberikan kemampuan observasi terhadap eksekusi alur kerja. Peristiwa dipancarkan di titik-titik penting selama eksekusi dan dapat dikonsumsi secara real time melalui streaming.

Jenis Peristiwa Bawaan

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

Nota

Saat agen menggunakan alat yang memerlukan persetujuan, RequestInfoEvent biasanya membawa ToolApprovalRequestContent payload untuk pemanggilan alat yang membutuhkan persetujuan manusia. Lihat Human-in-the-Loop untuk detail tentang penanganan peristiwa ini.

# 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

Nota

Saat agen menggunakan alat yang memerlukan persetujuan, request_info event biasanya membawa Content payload dengan type == "function_approval_request" untuk panggilan alat yang membutuhkan persetujuan manusia. Lihat Human-in-the-Loop untuk detail tentang penanganan peristiwa ini.

Nota

"output" dan "intermediate" adalah dua diskriminator output. Pelaksana yang ditetapkan sebagai sumber keluaran terminal menghasilkan "output" event (dikonsumsi oleh WorkflowRunResult.get_outputs()). Satu yang ditetapkan sebagai sumber keluaran perantara menghasilkan "intermediate" kejadian (yang dikonsumsi oleh WorkflowRunResult.get_intermediate_outputs()). Tipe "data" adalah alias usang untuk "intermediate" dan akan dihapus dalam rilis mendatang; dalam kode baru, sebaiknya lakukan pemfilteran pada "intermediate".

Peristiwa Go mengimplementasikan workflow.Event dan mengekspos payload mereka melalui Data(). Gunakan type switch untuk memeriksa field bertipe:

Jenis peristiwa	Muatan dan tujuan
`workflow.StartedEvent`	Memulai siklus pemrosesan input. `Message` berisi input.
`workflow.OutputEvent`	Eksekutor yang ditunjuk menghasilkan nilai dalam `Output`. `Tags` untuk mengidentifikasi output perantara.
`workflow.ErrorEvent`	Melaporkan kesalahan di `Error`. `SubWorkflowID` mengidentifikasi alur kerja berlapis jika disetel.
`workflow.WorkflowWarningEvent`	Melaporkan peringatan di `Message`.
`workflow.ExecutorInvokedEvent`	Executor mulai memproses. `ExecutorID` mengidentifikasinya, dan `Message` berisi inputnya.
`workflow.ExecutorCompletedEvent`	Eksekutor menyelesaikan pemrosesan. `Result` berisi hasilnya.
`workflow.ExecutorFailedEvent`	Eksekutor gagal. `ExecutorID` dan `Error` mengidentifikasi kegagalan.
`workflow.SuperStepStartedEvent`	Sebuah superstep dimulai. `StepNumber` dan `StartInfo` menggambarkannya.
`workflow.SuperStepCompletedEvent`	Sebuah superstep selesai. `CompletionInfo` menjelaskan pekerjaan tertunda, permintaan, dan checkpoint apa pun.
`workflow.RequestInfoEvent`	`Request` berisi permintaan eksternal yang dapat dijawab host.
`workflow.RequestHaltEvent`	Eksekutor meminta penghentian. `Result` berisi payload-nya, jika ada.

Go menggunakan OutputEvent untuk terminal dan output perantara. Panggil IsIntermediate() untuk membedakannya; itu tidak menggunakan diskriminator "output" dan "intermediate" Python yang terpisah.

Agen yang dihosting juga menggunakan OutputEvent, dengan *agent.ResponseUpdate payload atau *agent.Response saat peristiwa terkait diaktifkan. Go tidak menentukan jenis AgentResponseUpdateEvent dan AgentResponseEvent yang terpisah. Lihat Agent executors untuk konfigurasi.

Mengkonsumsi Peristiwa

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

Gunakan type switch untuk peristiwa alur kerja Go. Menangani kesalahan iterator dan peristiwa kegagalan secara terpisah: menerima output tidak menjamin bahwa sisa alur kerja berhasil.

Mengamati peristiwa

inproc.Default.Run melanjutkan ke penghentian berikutnya dan mengakumulasi peristiwa. Eksekusi dapat dijeda untuk input eksternal; penghentian tidak selalu berarti proses telah selesai.

Fungsi berikut mencetak aktivitas dan output executor, mengembalikan kegagalan, dan menutup 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() mengembalikan peristiwa yang diakumulasikan sejak pemanggilan sebelumnya. Bookmark-nya maju ketika Anda memanggilnya, bahkan jika Anda berhenti melakukan iterasi lebih awal. Gunakan OutgoingEvents() untuk memutar ulang semua peristiwa yang terakumulasi alih-alih memanggil NewEvents() lagi.

Acara siaran

Gunakan RunStreaming untuk mengamati event saat mereka tiba. Fungsi ini mengikuti alur kerja yang tidak memerlukan input eksternal:

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 terus menunggu ketika alur kerja memiliki permintaan eksternal yang tertunda. Tangani RequestInfoEvent dan kirim respons yang cocok melalui SendResponse saat Anda menggunakannya untuk alur kerja berbasis permintaan.

Gunakan WatchUntilHalt sebagai gantinya ketika konsumen harus melanjutkan kembali dari jeda tanpa menunggu respons. Hanya satu iterator peristiwa yang dapat memproses streaming run pada satu waktu. Lihat Alur kerja human-in-the-loop untuk penanganan permintaan dan pelanjutannya.

Peristiwa Kustom

Peristiwa kustom memungkinkan pelaksana memancarkan sinyal khusus domain selama eksekusi alur kerja yang disesuaikan dengan kebutuhan aplikasi Anda. Beberapa contoh kasus penggunaan:

Lacak kemajuan — laporkan langkah menengah sehingga penelepon dapat menampilkan pembaruan status.
Keluarkan diagnostik — tampilkan peringatan, metrik, atau informasi debug tanpa mengubah keluaran alur kerja.
Data domain relai — dorong payload terstruktur (misalnya, penulisan database, panggilan alat) ke pendengar secara real time.

Menentukan Peristiwa Kustom

Tentukan peristiwa kustom dengan subkelas WorkflowEvent. Konstruktor dasar menerima payload opsional object? data yang diekspos melalui properti 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) { }

Di Python, buat peristiwa kustom menggunakan WorkflowEvent kelas secara langsung dengan string diskriminator jenis kustom. Parameter type dan data membawa semua informasi.

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

Nota

Jenis "started"peristiwa , "status", dan "failed" dicadangkan untuk pemberitahuan siklus hidup kerangka kerja. Jika pelaksana mencoba memancarkan salah satu jenis ini, peristiwa diabaikan dan peringatan dicatat.

Tentukan peristiwa kustom dengan membuat jenis yang mengimplementasikan workflow.Event antarmuka. Metode Data mengembalikan payload peristiwa.

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

Memancarkan Peristiwa Kustom

Keluarkan peristiwa kustom dari handler pesan pengeksekusi dengan memanggil AddEventAsync di 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"));
    }
}

Keluarkan peristiwa kustom dari handler dengan memanggil add_event di 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"))

Picu peristiwa kustom dari handler eksekutor dengan memanggil AddEvent pada 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()

Peristiwa kustom tidak memerlukan WithOutputFrom: AddEvent memancarkannya secara langsung. Gunakan YieldOutput dan tetapkan sumber output saat Anda juga memerlukan output alur kerja biasa.

Memanfaatkan Peristiwa Khusus

Gunakan pencocokan pola untuk memfilter jenis peristiwa kustom Anda di aliran peristiwa:

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

Memfilter berdasarkan string diskriminator jenis kustom.

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

Gunakan sakelar jenis atau ketik pernyataan untuk memfilter jenis peristiwa kustom Anda di aliran peristiwa:

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

Langkah berikutnya

Pembuat & Pelaksanaan Alur Kerja

Topik terkait:

Saran dan Komentar

Apakah halaman ini membantu?

Terakhir diperbarui pada 2026-10-29

Events

Jenis Peristiwa Bawaan

Mengkonsumsi Peristiwa

Mengamati peristiwa

Acara siaran

Peristiwa Kustom

Menentukan Peristiwa Kustom

Memancarkan Peristiwa Kustom

Memanfaatkan Peristiwa Khusus

Langkah berikutnya

Saran dan Komentar

Sumber Daya Tambahan: