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 有效负载，用于需要人工审批的工具调用。有关处理这些事件的详细信息，请参阅 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

注释

当智能体使用需要审批的工具时，对于需要人工审批的工具调用，request_info 事件通常会携带一个包含 type == "function_approval_request" 的 Content 有效负载。有关处理这些事件的详细信息，请参阅 Human-in-the-Loop 。

注释

"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`	一个 superstep 开始了。 `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 类型。有关配置，请参阅 Agent 执行器。

使用事件

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

发出自定义事件

通过在 IWorkflowContext 上调用 AddEventAsync，从执行程序的消息处理程序发出自定义事件：

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

通过在 AddEvent 上调用 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()

自定义事件不需要 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-11-02