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

Note

عندما يستخدم الوكلاء الأدوات المطلوبة للموافقة، 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 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 عادة ما تحمل Content الأحداث حمولة مع type == "function_approval_request" لمكالمات الأدوات التي تتطلب موافقة بشرية. راجع Human-in-the-Loop للحصول على تفاصيل حول معالجة هذه الأحداث.

Note

"output" وهما "intermediate" الإخراجان التمييزيان. المنفذ المعين كمصدر إخراج طرفي ينبعث "output" من الأحداث (التي يستهلكها WorkflowRunResult.get_outputs()). واحدة معينة كمصدر إخراج وسيطة"intermediate" تنبعث منها الأحداث (المستهلكة من قبل WorkflowRunResult.get_intermediate_outputs()). "data" النوع هو اسم مستعار مهمل ل "intermediate" وسيتم إزالته في إصدار مستقبلي؛ يفضل التصفية "intermediate" في التعليمات البرمجية الجديدة.

استهلاك الأحداث

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

أحداث مخصصة

تتيح الأحداث المخصصة للمنفذين إصدار إشارات خاصة بالمجال أثناء تنفيذ سير العمل مصممة خصيصا لاحتياجات التطبيق الخاص بك. تتضمن بعض أمثلة حالات الاستخدام ما يلي:

• تعقب التقدم — الإبلاغ عن الخطوات المتوسطة حتى يتمكن المتصلون من إظهار تحديثات الحالة.
• إرسال التشخيصات — تحذيرات السطح أو المقاييس أو معلومات تتبع الأخطاء دون تغيير إخراج سير العمل.
• ترحيل بيانات المجال — دفع الحمولات المنظمة (على سبيل المثال، عمليات كتابة قاعدة البيانات، استدعاءات الأدوات) إلى المستمعين في الوقت الفعلي.

تعريف الأحداث المخصصة

حدد حدثا مخصصا عن طريق التصنيف 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})

Note

يتم حجز أنواع "started""status"الأحداث و و "failed" لإشعارات دورة حياة إطار العمل. إذا حاول منفذ إرسال أحد هذه الأنواع، يتم تجاهل الحدث ويتم تسجيل تحذير.

حدد حدثا مخصصا عن طريق إنشاء نوع ينفذ الواجهة workflow.Event . يقوم Data الأسلوب بإرجاع حمولة الحدث.

type ProgressEvent struct {
    Step string
}

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

إصدار أحداث مخصصة

إرسال أحداث مخصصة من معالج رسائل المنفذ عن طريق استدعاء AddEventAsync على IWorkflowContext:

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

إرسال أحداث مخصصة من معالج المنفذ عن طريق استدعاء AddEvent على workflow.Context:

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

استهلاك الأحداث المخصصة

استخدم مطابقة النمط لتصفية نوع الحدث المخصص في دفق الحدث:

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

Events

تصدر مهام سير العمل أحداثا أثناء التنفيذ. يمكن ملاحظة الأحداث من خلال كائن التشغيل.

مراقبة الأحداث

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
}

الخطوات التالية

منشئ سير العمل وتنفيذه

المواضيع ذات الصلة:

• الوكلاء في مهام سير العمل
• إدارة الدولة
• نقاط التفتيش و استئنافها
• إمكانية المراقبة