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의 separate "output" 및 "intermediate" discriminator를 사용하지 않습니다.

호스트된 에이전트는 또한 *agent.Response를 사용하며, 해당 이벤트가 활성화된 경우 OutputEvent 또는 *agent.ResponseUpdate 페이로드를 사용합니다. 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 는 워크플로에 보류 중인 외부 요청이 있을 때 계속 대기합니다. 요청 기반 워크플로에 사용할 때 SendResponse을(를) 통해 RequestInfoEvent을(를) 처리하고 일치하는 응답을 전송하세요.

소비자가 응답을 기다리지 않고 일시 정지 후 다시 돌아와야 할 때 대신 WatchUntilHalt을(를) 사용하세요. 한 번에 하나의 이벤트 반복자만 스트리밍 실행을 처리할 수 있습니다. 요청 처리 및 재개를 위한 휴먼 인 더 루프 워크플로를 참조하세요.

사용자 지정 이벤트

사용자 지정 이벤트를 통해 실행기는 애플리케이션의 요구에 맞게 워크플로 실행 중에 도메인별 신호를 내보냅니다. 몇 가지 예제 사용 사례는 다음과 같습니다.

진행률 추적 - 호출자가 상태 업데이트를 표시할 수 있도록 중간 단계를 보고합니다.
진단 정보 내보내기 — 워크플로의 출력을 변경하지 않고 경고, 메트릭, 또는 디버그 정보를 표시합니다.
릴레이 도메인 데이터 - 실시간으로 수신기에 구조적 페이로드(예: 데이터베이스 쓰기, 도구 호출)를 푸시합니다.

사용자 지정 이벤트 정의

서브클래싱하여 사용자 지정 이벤트를 정의합니다 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) { }

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

사용자 지정 이벤트 내보내기

다음의 AddEventAsyncIWorkflowContext를 호출하여 실행기의 메시지 처리기에서 사용자 지정 이벤트를 발생시킵니다.

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

다음을 호출 AddEventworkflow.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-10-30 에 마지막으로 업데이트됨