Events

工作流程事件系統提供對工作流程執行的可觀察性。事件會在執行的關鍵點發布，並可透過串流即時觀看。

內建事件類型

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

備註

當代理程式使用需要核准的工具時，RequestInfoEvent 通常會為需要人工核准的工具呼叫攜帶 ToolApprovalRequestContent 酬載。有關處理這些事件的細節，請參見「人機互動」一節。

# 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

備註

當代理程式使用需要核准的工具時，request_info 事件通常會攜帶包含 type == "function_approval_request" 的 Content 酬載，以用於需要人工核准的工具呼叫。有關處理這些事件的細節，請參見「人機互動」一節。

備註

"output" 和 "intermediate" 是兩個輸出判別子。指定為終端機輸出來源的執行程式會發出 "output" 事件 (由 WorkflowRunResult.get_outputs() 取用)。被指定為中間輸出來源的項目會發出"intermediate"事件（由WorkflowRunResult.get_intermediate_outputs()取用）。 "data" 型別是 "intermediate" 的已淘汰別名，並將於未來版本中移除；在新程式碼中，建議優先使用 "intermediate" 進行篩選。

Go 事件實作 workflow.Event 並透過 Data() 公開其有效負載。使用型別開關檢查它們的具型別欄位：

事件類型	酬載與用途
`workflow.StartedEvent`	啟動輸入處理循環。 `Message` 包含輸入。
`workflow.OutputEvent`	指定執行者會在 `Output` 中產生一個值。 `Tags` 標示中間輸出。
`workflow.ErrorEvent`	在 `Error` 中回報錯誤。 `SubWorkflowID` 設定時可識別巢狀工作流程。
`workflow.WorkflowWarningEvent`	在 `Message` 中回報警告。
`workflow.ExecutorInvokedEvent`	執行器開始處理。 `ExecutorID` 用來識別它，並且 `Message` 包含其輸入。
`workflow.ExecutorCompletedEvent`	執行器完成處理。 `Result` 包含其結果。
`workflow.ExecutorFailedEvent`	執行器失敗。 `ExecutorID` 和 `Error` 會識別該失敗。
`workflow.SuperStepStartedEvent`	超級步驟開始。 `StepNumber` 和 `StartInfo` 會加以描述。
`workflow.SuperStepCompletedEvent`	超級步驟完成。 `CompletionInfo` 描述待處理工作、請求及任何檢查點。
`workflow.RequestInfoEvent`	`Request` 包含一個主機可以回應的外部請求。
`workflow.RequestHaltEvent`	執行程式要求停止。若有酬載，`Result` 會包含其酬載。

Go 對終端和中間輸出都使用 OutputEvent。請呼叫 IsIntermediate() 來區分它們；它不會使用 Python 中獨立的 "output" 和 "intermediate" 鑑別子。

當對應事件啟用時，託管代理也會使用 OutputEvent，並帶有 *agent.ResponseUpdate 或 *agent.Response 酬載。 Go 並不定義不同的 AgentResponseEvent 和 AgentResponseUpdateEvent 類型。關於配置，請參見代理程式執行器。

消費事件

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

請對 Go 工作流程事件使用型別切換。分別處理迭代器錯誤與失敗事件：收到輸出並不保證整個工作流程都能成功。

觀察事件

inproc.Default.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() 回傳自上次呼叫以來累積的事件。即使你提前停止迭代，它也會在你呼叫它時使書籤位置往前推進。使用 OutgoingEvents() 來重播所有累積的事件，而不是再次呼叫 NewEvents()。

串流活動

用 RunStreaming 來觀察事件發生。此函式會觀察不需要外部輸入的工作流程：

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 當工作流程有待處理的外部請求時，它一直等待。當你將其用於請求驅動的工作流程時，處理 RequestInfoEvent 並透過 SendResponse 傳送相符的回應。

當取用者必須在暫停時傳回而不等待回應時，請改用 WatchUntilHalt。一次只能有一個事件反覆運算器取用串流執行。如需要求處理與繼續執行的相關資訊，請參閱人機迴圈工作流程。

自訂事件

自訂事件讓執行者在工作流程執行時，能根據應用程式需求發出專屬領域訊號。一些範例使用案例包括：

追蹤進度 — 報告中間步驟，讓來電者能顯示狀態更新。
發出診斷 — 顯示警告、指標或偵錯資訊，且不改變工作流程的輸出。
中繼領域資料 — 即時將結構化有效載荷（例如資料庫寫入、工具呼叫）推送給聽者。

定義自訂事件

透過子 WorkflowEvent類別定義一個自訂事件。基礎建構器接受一個可以選擇的object? data有效載荷，這個載荷會透過屬性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) { }

在 Python 中，直接使用 WorkflowEvent 類別建立自訂事件，並搭配自訂型別的判別器字串。 type和data參數承載所有資訊。

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

備註

事件類型 "started"、 "status"、 "failed" 保留用於框架生命週期通知。若執行者嘗試發出上述類型，該事件會被忽略並記錄警告。

透過建立實作 workflow.Event 介面的型別來定義自訂事件。 Data 方法會傳回事件酬載。

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

發出自訂事件

透過在AddEventAsync上呼叫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"));
    }
}

透過在 WorkflowContext 上呼叫 add_event，從處理常式發出自訂事件：

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

透過在 workflow.Context 上呼叫 AddEvent，從執行程式處理常式發出自訂事件：

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

自訂事件不必使用 WithOutputFrom：AddEvent 會直接發出這些事件。當你同時需要一般工作流程輸出時，請使用 YieldOutput 並指定一個輸出來源。

取用自訂事件

使用模式匹配來篩選事件串流中的自訂事件類型：

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

自訂類型判別器字串上的濾波器：

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

使用型別切換或型別斷言來篩選事件串流中的自訂事件類型：

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

下一步

工作流程建構器與執行

相關主題：

意見反應

此頁面對您有幫助嗎？

上次更新時間： 2026-10-30