Aracılığıyla paylaş

Events

İş akışı olay sistemi, iş akışı yürütmeye gözlemlenebilirlik sağlar. Olaylar, yürütme sırasında önemli noktalarda üretilir ve streaming aracılığıyla gerçek zamanlı olarak işlenebilir.

Yerleşik Olay Türleri

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

Olayları Tüketme

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

Özel Olaylar

Özel olaylar, yürütücülerin iş akışı yürütme sırasında uygulamanızın gereksinimlerine göre uyarlanmış etki alanına özgü sinyaller yaymasına olanak tanır. Bazı örnek kullanım örnekleri şunlardır:

  • İlerleme durumunu izleme — ara adımları bildirerek arayanların durum güncelleştirmelerini gösterebilmesini sağlayın.
  • Tanılama verilerini yayımlayın — iş akışı çıktısını değiştirmeden uyarıları, metrikleri veya hata ayıklama bilgilerini yüzeye çıkarın.
  • Aktarma alanı verileri — yapılandırılmış yükleri (örneğin, veritabanı yazma işlemleri, araç çağrıları) dinleyicilere gerçek zamanlı olarak iletme.

Özel Olayları Tanımlama

Bir alt sınıf oluşturarak WorkflowEvent özel bir olay tanımlayın. Temel oluşturucu, Data özelliği aracılığıyla kullanıma sunulan isteğe bağlı bir object? data yükü kabul eder.

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'da, sınıfını WorkflowEvent doğrudan özel tür ayrıştırıcı dizesiyle kullanarak özel olaylar oluşturun. type ve data parametreleri tüm bilgileri taşır.

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

Uyarı

, "started"ve "status" olay türleri "failed"çerçeve yaşam döngüsü bildirimleri için ayrılmıştır. Yürütücü bu türlerden birini tetiklemeye çalışırsa, olay yoksayılır ve bir uyarı kaydedilir.

Özel Olayları Tetikleme

AddEventAsync üzerinde IWorkflowContext çağrıda bulunarak bir yürütücünün ileti işleyicisinden özel olaylar yayarak:

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

Bir işleyiciden özel olayları yaymak için WorkflowContext üzerinde add_event çağırın:

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

Özel Olayları Kullanma

Olay akışındaki özel olay türünüz için filtrelemek için desen eşleştirmeyi kullanın:

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

Özel tür ayrıştırıcı dizesini filtreleyin:

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

Sonraki Adımlar

İş Akışı Oluşturucusu ve Yürütme

İlgili konular:

  • Last updated on