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

Note

エージェントが承認が必要なツールを使用する場合、 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

Note

エージェントが承認が必要なツールを使用する場合、request_infoイベントは通常、人間の承認を必要とするツール呼び出しのContentを含むtype == "function_approval_request"ペイロードを伝達します。これらのイベントの処理の詳細については、 Human-in-the-Loop を参照してください。

Note

"output" と "intermediate" は、2 つの出力識別子です。 ターミナル出力ソースとして指定された Executor は、("output" によって使用される) WorkflowRunResult.get_outputs()イベントを出力します。 中間出力ソースとして指定された 1 つは、("intermediate" によって使用される) WorkflowRunResult.get_intermediate_outputs()イベントを出力します。 "data"の種類は、"intermediate"の非推奨のエイリアスであり、今後のリリースで削除される予定です。新しいコードで"intermediate"をフィルター処理することをお勧めします。

Go イベントは workflow.Event を実装し、Data() を介してペイロードを公開します。型スイッチを使用して、型指定されたフィールドを調べます。

イベントの種類	ペイロードと目的
`workflow.StartedEvent`	入力処理サイクルを開始します。 `Message` には入力が含まれています。
`workflow.OutputEvent`	指定された Executor は、 `Output`で値を生成します。 `Tags` は中間出力を識別します。
`workflow.ErrorEvent`	エラーを `Error` に表示します。 `SubWorkflowID` は、設定時に入れ子になったワークフローを識別します。
`workflow.WorkflowWarningEvent`	`Message`で警告を表示します。
`workflow.ExecutorInvokedEvent`	Executor が処理を開始します。 `ExecutorID` によってそれが識別され、 `Message` にその入力が含まれます。
`workflow.ExecutorCompletedEvent`	Executor は処理を完了します。 `Result` には結果が含まれます。
`workflow.ExecutorFailedEvent`	Executor の実行に失敗しました。 `ExecutorID` と `Error` でエラーが特定されます。
`workflow.SuperStepStartedEvent`	スーパーステップが開始されます。 `StepNumber` と `StartInfo` がそれを説明します。
`workflow.SuperStepCompletedEvent`	スーパーステップが完了しました。 `CompletionInfo` は、保留中の作業、要求、チェックポイントについて記述します。
`workflow.RequestInfoEvent`	`Request` には、ホストが応答できる外部要求が含まれています。
`workflow.RequestHaltEvent`	Executer が停止を要求しました。 `Result` にはペイロード (存在する場合) が含まれます。

Go では、ターミナル出力と中間出力の両方に OutputEvent が使用されます。 IsIntermediate()を呼び出して区別します。Pythonの個別の"output"と"intermediate"判別子は使用しません。

また、ホストされるエージェントは、対応するイベントが有効な場合に、*agent.ResponseUpdate または *agent.Response のペイロードと共に OutputEvent を使用します。 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 は次の停止位置に進み、イベントを蓄積します。実行は外部入力に対して一時停止できます。停止したからといって完了とは限りません。

次の関数は、Executor アクティビティと出力を出力し、失敗を返し、実行を閉じます。

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 を使用します。ストリーミング実行を一度に処理できるイベント反復子は 1 つだけです。要求の処理と再開については、 Human-in-the-loop ワークフローを参照してください。

カスタムイベント

カスタムイベントを使用すると、Executor は、アプリケーションのニーズに合わせて調整されたワークフローの実行中にドメイン固有のシグナルを出力できます。ユースケースの例を次にいくつか示します。

進行状況を追跡 する - 発信者が状態の更新を表示できるように、中間ステップを報告します。
診断の出力 — ワークフロー出力を変更せずに、警告、メトリック、またはデバッグ情報を表示します。
リレードメインデータ — 構造化ペイロード (データベースの書き込み、ツール呼び出しなど) をリスナーにリアルタイムでプッシュします。

カスタムイベントの定義

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

Note

イベントの種類 "started"、 "status"、および "failed" は、フレームワークのライフサイクル通知用に予約されています。 Executor がこれらの型のいずれかを出力しようとすると、イベントは無視され、警告がログに記録されます。

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を呼び出して、Executor のメッセージハンドラーからカスタムイベントを生成します。

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

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

AddEventでworkflow.Contextを呼び出して、Executor ハンドラーからカスタムイベントを生成します。

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

Events

組み込みのイベントの種類

イベントの使用

イベントを観察する

ストリーミング イベント

カスタム イベント

カスタム イベントの定義

カスタム イベントの出力

カスタム イベントの処理

次のステップ

フィードバック

その他のリソース

ストリーミングイベント

カスタムイベント

カスタムイベントの定義

カスタムイベントの出力

カスタムイベントの処理