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" في التعليمات البرمجية الجديدة.

انتقل إلى تنفيذ 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 يحتوي على البيانات الأساسية الخاصة به، إن وجدت.

يستخدم OutputEvent Go لكل من الإخراج الطرفي والمتوسط. دعوة IsIntermediate() لتمييزها؛ فهي لا تستخدم Python منفصلة "output" وتمييزية"intermediate".

يستخدم OutputEventالوكلاء المستضافون أيضا ، مع *agent.ResponseUpdate حمولة أو *agent.Response عند تمكين الأحداث المقابلة. لا تحدد Go أنواعا و منفصلة AgentResponseEventAgentResponseUpdateEvent . راجع عامل المنفذين للتكوين .

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

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 يستمر في الانتظار عندما يحتوي سير العمل على طلبات خارجية معلقة. معالجة RequestInfoEvent الاستجابات المطابقة SendResponse وإرسالها عند استخدامها لسير العمل المستند إلى الطلب.

استخدم WatchUntilHalt بدلا من ذلك عندما يجب أن يعود المستهلك مؤقتا دون انتظار استجابة. يمكن لمكرر حدث واحد فقط استهلاك تشغيل دفق في كل مرة. راجع مهام سير العمل البشرية في الحلقة لمعالجة الطلب واستئنافه.

أحداث مخصصة

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

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

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

حدد حدثا مخصصا عن طريق التصنيف 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 الأسلوب بإرجاع حمولة الحدث.

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

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

إرسال أحداث مخصصة من معالج رسائل المنفذ عن طريق استدعاء 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:

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

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

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

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

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