워크플로 이벤트

이벤트

워크플로 이벤트 시스템은 워크플로 실행에 대한 관찰 가능성(observability)을 제공합니다. 이벤트는 실행 중 핵심 지점에서 발생하며, 스트리밍을 통해 실시간으로 소비할 수 있어요.

내장 이벤트 유형

::: zone pivot="programming-language-csharp"

// 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에서 다룹니다.

::: zone-end

::: zone pivot="programming-language-python"

# 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 이벤트는 보통 type == "function_approval_request"인 Content 페이로드를 담습니다. 이런 이벤트를 처리하는 방법은 Human-in-the-Loop에서 다룹니다.

[!NOTE] "output"과 "intermediate"가 두 가지 출력 구분자입니다. 터미널 출력 소스로 지정된 실행기는 "output" 이벤트를 발생시키고(WorkflowRunResult.get_outputs()로 소비), 중간 출력 소스로 지정된 실행기는 "intermediate" 이벤트를 발생시킵니다(WorkflowRunResult.get_intermediate_outputs()로 소비). "data" 유형은 "intermediate"의 deprecated 별칭으로, 이후 릴리스에서 제거될 예정입니다. 새 코드에서는 "intermediate"로 필터링하는 걸 권장해요.

::: zone-end

이벤트 소비하기

::: zone pivot="programming-language-csharp"

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

::: zone-end

::: zone pivot="programming-language-python"

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

::: zone-end

커스텀 이벤트

커스텀 이벤트는 실행기가 워크플로 실행 중 여러분의 애플리케이션 요구에 맞는 도메인 특화 신호를 발생시키게 해줍니다. 예를 들면 이런 식으로 쓰일 수 있어요.

진행 상황 추적 — 중간 단계를 보고해서 호출자가 상태 업데이트를 보여줄 수 있게 합니다.
진단 이벤트 발행 — 워크플로 출력을 바꾸지 않으면서 경고·메트릭·디버그 정보를 표면화합니다.
도메인 데이터 중계 — 구조화된 페이로드(예: 데이터베이스 쓰기, 도구 호출)를 실시간으로 리스너에게 밀어 보냅니다.

커스텀 이벤트 정의하기

::: zone pivot="programming-language-csharp"

WorkflowEvent를 서브클래싱해 커스텀 이벤트를 정의합니다. 기본 생성자는 Data 속성으로 노출되는 선택적 object? 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) { }

::: zone-end

::: zone pivot="programming-language-python"

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" 이벤트 유형은 프레임워크 생명주기 알림용으로 예약되어 있습니다. 실행기가 이 유형 중 하나를 발생시키려 하면, 해당 이벤트는 무시되고 경고가 로그에 남습니다.

::: zone-end

::: zone pivot="programming-language-go"

workflow.Event 인터페이스를 구현하는 유형을 만들어 커스텀 이벤트를 정의합니다. Data 메서드가 이벤트 페이로드를 반환해요.

type ProgressEvent struct {
    Step string
}

func (e ProgressEvent) Data() any {
    return e.Step
}

::: zone-end

커스텀 이벤트 발생시키기

::: zone pivot="programming-language-csharp"

실행기의 메시지 핸들러에서 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"));
    }
}

::: zone-end

::: zone pivot="programming-language-python"

핸들러에서 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"))

::: zone-end

::: zone pivot="programming-language-go"

실행기 핸들러에서 workflow.Context의 AddEvent를 호출해 커스텀 이벤트를 발생시킵니다.

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

::: zone-end

커스텀 이벤트 소비하기

::: zone pivot="programming-language-csharp"

이벤트 스트림에서 패턴 매칭으로 커스텀 이벤트 유형을 걸러냅니다.

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

::: zone-end

::: zone pivot="programming-language-python"

커스텀 유형 구분 문자열로 필터링합니다.

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

::: zone-end

::: zone pivot="programming-language-go"

이벤트 스트림에서 타입 스위치나 타입 단언으로 커스텀 이벤트 유형을 걸러냅니다.

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

::: zone-end

::: zone pivot="programming-language-go"

이벤트

워크플로는 실행 중 이벤트를 발생시킵니다. 이벤트는 run 객체를 통해 관찰할 수 있어요.

이벤트 관찰하기

run, err := inproc.Default.Run(ctx, wf, input)
for evt := range run.NewEvents() {
    switch e := evt.(type) {
    case workflow.ExecutorCompletedEvent:
        fmt.Printf("Executor %s completed: %v\n", e.ExecutorID, e.Result)
    case workflow.OutputEvent:
        fmt.Printf("Output from %s: %v\n", e.ExecutorID, e.Output)
    }
}

이벤트 스트리밍

스트리밍 워크플로라면 inproc.Default.RunStreaming와 WatchStream을 사용합니다.

run, err := inproc.Default.RunStreaming(ctx, wf, input)
for evt, err := range run.WatchStream(ctx) {
    if err != nil {
        panic(err)
    }
    // process streaming events
}

::: zone-end

다음 단계

[!div class="nextstepaction"] 워크플로 빌더 & 실행

관련 주제: