워크플로 이벤트 시스템은 워크플로 실행에 대한 가시성을 제공합니다. 이벤트는 실행 중에 주요 지점에서 내보내지고 스트리밍을 통해 실시간으로 사용할 수 있습니다.
기본 제공 이벤트 유형
// 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
# 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
이벤트 소비
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
사용자 지정 이벤트
사용자 지정 이벤트를 통해 실행기는 애플리케이션의 요구에 맞게 워크플로 실행 중에 도메인별 신호를 내보냅니다. 몇 가지 예제 사용 사례는 다음과 같습니다.
- 진행률 추적 - 호출자가 상태 업데이트를 표시할 수 있도록 중간 단계를 보고합니다.
- 진단 정보 내보내기 — 워크플로의 출력을 변경하지 않고 경고, 메트릭, 또는 디버그 정보를 표시합니다.
- 릴레이 도메인 데이터 - 실시간으로 수신기에 구조적 페이로드(예: 데이터베이스 쓰기, 도구 호출)를 푸시합니다.
사용자 지정 이벤트 정의
서브클래싱하여 사용자 지정 이벤트를 정의합니다 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" 프레임워크 수명 주기 알림용으로 예약됩니다. 실행기가 이러한 형식 중 하나를 내보내려고 하면 이벤트가 무시되고 경고가 기록됩니다.
사용자 지정 이벤트 내보내기
다음의 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"));
}
}
처리기 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"))
사용자 지정 이벤트 사용
패턴 일치를 사용하여 이벤트 스트림에서 사용자 지정 이벤트 유형을 필터링합니다.
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
다음 단계:
관련 항목: