次の方法で共有

イベント

ワークフロー イベント システムは、ワークフローの実行に可観測性を提供します。 イベントは実行中の重要なポイントで生成され、ストリーミングを介してリアルタイムで使用できます。

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

// 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 an 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"                # Executor emitted data (e.g., AgentResponse)

# 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イベントは通常、人間の承認を必要とするツール呼び出しのContentを含むtype == "function_approval_request"ペイロードを伝達します。 これらのイベントの処理の詳細については、 Human-in-the-Loop を参照してください。

イベントの消費

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_stream(input_message):
    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 == "output":
        print(f"Workflow produced output: {event.data}")
        return
    elif event.type == "error":
        print(f"Workflow error: {event.data}")
        return

カスタム イベント

カスタム イベントを使用すると、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})

注

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

カスタム イベントの出力

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

カスタムイベントの処理

パターン マッチングを使用して、イベント ストリーム内のカスタム イベントの種類をフィルター処理します。

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

次のステップ

ワークフロー ビルダーと実行

関連トピック:

  • Last updated on