المنفذون

المنفذون هم كتل الإنشاء الأساسية التي تعالج الرسائل في سير العمل. وهي وحدات معالجة مستقلة تتلقى رسائل مكتوبة، وتنفذ عمليات، ويمكن أن تنتج رسائل إخراج أو أحداث.

نظرة عامة

يحتوي كل منفذ على معرف فريد ويمكنه التعامل مع أنواع رسائل معينة. يمكن أن يكون المنفذون:

• مكونات المنطق المخصصة — معالجة البيانات أو استدعاء واجهات برمجة التطبيقات أو تحويل الرسائل
• عوامل الذكاء الاصطناعي — استخدم LLMs لإنشاء استجابات (راجع العوامل في مهام سير العمل)

Important

الطريقة الموصى بها لتعريف معالجات رسائل المنفذ في C# هي استخدام السمة [MessageHandler] على الأساليب داخل partial فئة مشتقة من Executor. يستخدم هذا إنشاء مصدر وقت التحويل البرمجي لتسجيل المعالج، ما يوفر أداء أفضل، والتحقق من صحة وقت التحويل البرمجي، وتوافق AOT الأصلي.

بنية المنفذ الأساسية

يشتق المنفذون من Executor الفئة الأساسية ويستخدمون السمة [MessageHandler] لتعريف أساليب المعالج. يجب وضع علامة partial على الفئة لتمكين إنشاء المصدر.

using Microsoft.Agents.AI.Workflows;

internal sealed partial class UppercaseExecutor() : Executor("UppercaseExecutor")
{
    [MessageHandler]
    private ValueTask<string> HandleAsync(string message, IWorkflowContext context)
    {
        string result = message.ToUpperInvariant();
        return ValueTask.FromResult(result); // Return value is automatically sent to connected executors
    }
}

يمكنك أيضا إرسال الرسائل يدويا دون إرجاع قيمة:

internal sealed partial class UppercaseExecutor() : Executor("UppercaseExecutor")
{
    [MessageHandler]
    private async ValueTask HandleAsync(string message, IWorkflowContext context)
    {
        string result = message.ToUpperInvariant();
        await context.SendMessageAsync(result); // Manually send messages to connected executors
    }
}

Tip

يمكن للمنفذين الاحتفاظ بحالة قابلة للتغيير. إذا تمت مشاركة منفذ ذي حالة عبر عمليات تشغيل سير العمل، فيجب تنفيذه IResettableExecutor لمسح الحالة القديمة بين عمليات التشغيل. راجع المنفذون القابلون لإعادة التعيين للحصول على التفاصيل.

أنواع إدخال متعددة

معالجة أنواع إدخال متعددة عن طريق تعريف أساليب متعددة [MessageHandler] :

internal sealed partial class SampleExecutor() : Executor("SampleExecutor")
{
    [MessageHandler]
    private ValueTask<string> HandleStringAsync(string message, IWorkflowContext context)
    {
        return ValueTask.FromResult(message.ToUpperInvariant());
    }

    [MessageHandler]
    private ValueTask<int> HandleIntAsync(int message, IWorkflowContext context)
    {
        return ValueTask.FromResult(message * 2);
    }
}

المنفذون Function-Based

إنشاء منفذ من دالة باستخدام أسلوب الملحق BindExecutor :

Func<string, string> uppercaseFunc = s => s.ToUpperInvariant();
var uppercase = uppercaseFunc.BindExecutor("UppercaseExecutor");

كائن IWorkflowContext

IWorkflowContext يوفر أساليب للتفاعل مع سير العمل أثناء التنفيذ:

• SendMessageAsync — إرسال رسائل إلى المنفذين المتصلين
• YieldOutputAsync — إنتاج مخرجات سير العمل التي تم إرجاعها/دفقها إلى المتصل

internal sealed partial class OutputExecutor() : Executor("OutputExecutor")
{
    [MessageHandler]
    private async ValueTask HandleAsync(string message, IWorkflowContext context)
    {
        await context.YieldOutputAsync("Hello, World!");
    }
}

إذا لم يرسل المعالج رسائل ولا ينتج عنه مخرجات، فيمكنه ببساطة تنفيذ تأثيرات جانبية:

internal sealed partial class LogExecutor() : Executor("LogExecutor")
{
    [MessageHandler]
    private void Handle(string message, IWorkflowContext context)
    {
        Console.WriteLine("Doing some work...");
    }
}

تعريف أنواع البروتوكولات

يعلن بروتوكول المنفذ عن أنواع الرسائل التي قد يرسلها إلى المنفذين المتصلين وأنواع الإخراج التي قد ينتجها. يتحقق سير العمل من صحة الاستدعاءات إلى SendMessageAsync هذه الإعلانات ومعارضتها YieldOutputAsync ويطرح InvalidOperationException عندما يستخدم المنفذ نوعا غير معلن.

يستخدم [SendsMessage] للإعلان عن أنواع الرسائل المرسلة وتعريف [YieldsOutput] أنواع الإخراج التي تم إرجاعها. تصف هذه السمات قدرات المنفذ؛ لا يرسلون القيم أو ينتجونها بأنفسهم. تطبيق كل سمة عدة مرات عندما يستخدم المنفذ أنواعا متعددة.

بالنسبة للمنفذين الذين يعانون من معالج مكتوب واحد، اشتق من Executor<TInput> أو Executor<TInput, TOutput> وتجاوز HandleAsync:

internal sealed record ProcessRequest(string Text);
internal sealed record ProgressUpdate(string Status);

[SendsMessage(typeof(ProgressUpdate))]
[YieldsOutput(typeof(string))]
internal sealed partial class ProcessingExecutor()
    : Executor<ProcessRequest>("ProcessingExecutor")
{
    public override async ValueTask HandleAsync(
        ProcessRequest message,
        IWorkflowContext context,
        CancellationToken cancellationToken = default)
    {
        await context.SendMessageAsync(
            new ProgressUpdate("Processing started"),
            cancellationToken);

        await context.YieldOutputAsync(
            message.Text.ToUpperInvariant(),
            cancellationToken);
    }
}

عند الإشارة إلى منشئ مصدر مهام سير العمل، يجب تعريف partial فئة مع [SendsMessage] أو [YieldsOutput] حتى يتمكن المولد من إضافة تكوين البروتوكول الخاص به.

بالنسبة للمنفذين الذين تم إنشاؤهم من المصدر باستخدام [MessageHandler] أساليب، قم بتعريف الأنواع المستخدمة من قبل معالج واحد مع الوسيطات الخاصة به Send والوسيطات Yield المسماة، مثل [MessageHandler(Send = [typeof(ProgressUpdate)], Yield = [typeof(string)])]. استخدم مستوى [SendsMessage] الفئة وعندما [YieldsOutput] تنطبق الإعلانات على المنفذ بأكمله.

تتم إضافة أنواع إرجاع المعالجات غير الفارغة تلقائيا إلى أنواع البروتوكولات المرسلة والمنتجة عند ExecutorOptions.AutoSendMessageHandlerResultObject تمكينها.ExecutorOptions.AutoYieldOutputHandlerResultObject يتم تمكين كلا الخيارين بشكل افتراضي. لذلك، هناك حاجة إلى إعلانات صريحة في المقام الأول للأنواع الإضافية المنبعثة مباشرة من خلال SendMessageAsync أو YieldOutputAsync.

[YieldsOutput] يسمح للمنفذ بإخراج نوع، ولكنه لا يعين المنفذ كمصدر إخراج طرفي. قم بتسجيل المنفذ مع WorkflowBuilder.WithOutputFrom لقيمه المتحققة للسطح إلى المتصل سير العمل.

بنية المنفذ الأساسية

يرث المنفذون من Executor الفئة الأساسية. يستخدم كل منفذ أساليب مزينة بالمصمم @handler . يجب أن تحتوي المعالجات على تعليقات توضيحية للنوع المناسب لتحديد أنواع الرسائل التي تعالجها.

from agent_framework import (
    Executor,
    WorkflowContext,
    handler,
)

class UpperCase(Executor):

    @handler
    async def to_upper_case(self, text: str, ctx: WorkflowContext[str]) -> None:
        """Convert the input to uppercase and forward it to the next node."""
        await ctx.send_message(text.upper())

المنفذون Function-Based

إنشاء منفذ من دالة باستخدام @executor مصمم الديكور:

from agent_framework import (
    WorkflowContext,
    executor,
)

@executor(id="upper_case_executor")
async def upper_case(text: str, ctx: WorkflowContext[str]) -> None:
    """Convert the input to uppercase and forward it to the next node."""
    await ctx.send_message(text.upper())

أنواع إدخال متعددة

معالجة أنواع إدخال متعددة عن طريق تعريف معالجات متعددة:

class SampleExecutor(Executor):

    @handler
    async def to_upper_case(self, text: str, ctx: WorkflowContext[str]) -> None:
        await ctx.send_message(text.upper())

    @handler
    async def double_integer(self, number: int, ctx: WorkflowContext[int]) -> None:
        await ctx.send_message(number * 2)

معلمات النوع الصريح

كبديل لكتابة التعليقات التوضيحية، يمكنك تحديد الأنواع بشكل صريح عبر معلمات مصمم الديكور:

Important

عند استخدام معلمات النوع الصريحة، يجب تحديد جميع الأنواع عبر مصمم الديكور — لا يمكنك خلط المعلمات الصريحة مع التعليقات التوضيحية للنوع. المعلمة input مطلوبة؛ output وهي workflow_output اختيارية.

class ExplicitTypesExecutor(Executor):

    @handler(input=str, output=str)
    async def to_upper_case(self, text, ctx) -> None:
        await ctx.send_message(text.upper())

    @handler(input=str | int, output=str)
    async def handle_mixed(self, message, ctx) -> None:
        await ctx.send_message(str(message).upper())

    @handler(input=str, output=int, workflow_output=bool)
    async def process_with_workflow_output(self, message, ctx) -> None:
        await ctx.send_message(len(message))
        await ctx.yield_output(True)

كائن WorkflowContext

WorkflowContext يوفر أساليب للتفاعل مع سير العمل أثناء التنفيذ:

• send_message — إرسال رسائل إلى المنفذين المتصلين
• yield_output — إنتاج مخرجات سير العمل التي تم إرجاعها/دفقها إلى المتصل

class OutputExecutor(Executor):

    @handler
    async def handle(self, message: str, ctx: WorkflowContext[Never, str]) -> None:
        await ctx.yield_output("Hello, World!")

إذا لم يرسل المعالج رسائل ولا ينتج مخرجات، فلا يلزم وجود معلمة نوع:

class LogExecutor(Executor):

    @handler
    async def handle(self, message: str, ctx: WorkflowContext) -> None:
        print("Doing some work...")

تعيين منفذي الإخراج الطرفي والمتوسط

أي المنفذين يساهمون في الإجابة الطرفية لسير العمل والتي تنبعث منها تقدم المراقبة هو قرار وقت البناء الذي تم تكوينه على WorkflowBuilder، وليس علامة لكل انبعاث.

• output_from— المنفذون الذين تنتج "output" مكالماتهم ctx.yield_output(...) أحداثا ويتم إرجاعهم بواسطة WorkflowRunResult.get_outputs().
• intermediate_output_from— المنفذون الذين تنتج "intermediate" مكالماتهم ctx.yield_output(...) أحداثا ويتم إرجاعهم بواسطة WorkflowRunResult.get_intermediate_outputs().

from agent_framework import WorkflowBuilder

workflow = WorkflowBuilder(
    start_executor=analysis_executor,
    output_from=[summary_executor],
    intermediate_output_from=[analysis_executor],
).build()

Important

ctx.yield_output(...) ليس لديه علم لكل انبعاث. يتم تسمية "output" نفس المكالمة أو "intermediate" تستند فقط إلى تعيين المنشئ. لا ctx.yield_intermediate(...) توجد واجهة برمجة تطبيقات - لا يختلف التعيين لكل عائد.

كلتا القائمتين اختياريتان. إذا تم توفير أي من قائمة تحديد الإخراج، فلا يزال بإمكان المنفذ الذي يظهر في أي من القائمةين إرسال رسائل إلى منفذي انتقال البيانات من الخادم عبر ctx.send_message(...)، ولكن يتم إخفاء مكالماته yield_output . إذا تم حذف كلتا القائمتين، فإن كل yield_output قائمة لا تزال تنبعث "output" للتوافق.

بنية المنفذ الأساسية

المنفذون هم وحدات المعالجة في سير العمل. فهم يتلقون المدخلات، وينفذون العمل، وينتجون الإخراج.

أنواع إدخال متعددة

تسجيل معالجات متعددة عن طريق تكوين المسارات على المنفذ:

sample := (&workflow.Executor{
    ID: "SampleExecutor",
    ConfigureProtocol: func(pb *workflow.ProtocolBuilder) (*workflow.ProtocolBuilder, error) {
        pb.RouteBuilder.
            AddHandlerRaw(reflect.TypeFor[string](), reflect.TypeFor[string](), func(_ *workflow.Context, msg any) (any, error) {
                return strings.ToUpper(msg.(string)), nil
            }).
            AddHandlerRaw(reflect.TypeFor[int](), reflect.TypeFor[int](), func(_ *workflow.Context, msg any) (any, error) {
                return msg.(int) * 2, nil
            })
        return pb, nil
    },
}).Bind()

المنفذون Function-Based

أبسط طريقة لإنشاء منفذ هي مع workflow.NewExecutor(...).Bind():

uppercase := workflow.NewExecutor("UppercaseExecutor", func(input string) string {
    return strings.ToUpper(input)
}).Bind()

يسجل منفذو الوظائف تلقائيا نوع الإدخال ويمكنهم الإرسال التلقائي والقيم التي تم إرجاعها تلقائيا.

سير العمل. عنصر السياق

يمكن للمعالجات قبول *workflow.Context التفاعل مع سير العمل أثناء التنفيذ:

output := workflow.NewExecutor("OutputExecutor", func(ctx *workflow.Context, message string) error {
    return ctx.YieldOutput("Hello, World!")
}).Bind()

يعرض السياق أيضا واجهات برمجة التطبيقات مثل SendMessageوAddEventReadStatePostRequestQueueStateUpdate.

منفذو العامل

يمكن استخدام الوكلاء كمنفذين لسير العمل عبر agentworkflow.New:

agentExecutor := agentworkflow.New(myAgent, agentworkflow.Config{
    EmitUpdateEvents: true,
})

دورة حياة المنفذ

يدعم المنفذون خطافات دورة الحياة من خلال الحقول على workflow.Executor:

خطاف	الغرض
ConfigureProtocol	إعداد توجيه الرسائل وأنواع الإرسال/العائد المعلنة
InitializeFunc	الإعداد عند إنشاء مثيل منفذ للتشغيل
ResetFunc	إعادة تعيين الحالة المحلية للمنفذ قبل إعادة الاستخدام
OnCheckpointFunc	حفظ الحالة عند نقطة التحقق
OnCheckpointRestoredFunc	استعادة الحالة من نقطة التحقق
OnMessageDeliveryStartingFunc	تشغيل قبل تسليم فائقة الرسائل
OnMessageDeliveryFinishedFunc	تشغيل بعد انتهاء عملية فائقة من تسليم الرسائل

stateful := workflow.NewExecutor("StatefulExecutor", handleMessage).Extend(&workflow.Executor{
    InitializeFunc: func(ctx *workflow.Context) error {
        return nil
    },
    ResetFunc: func() error {
        return nil
    },
    OnCheckpointFunc: func(ctx *workflow.Context) error {
        return ctx.QueueStateUpdate("StatefulExecutorState", "", currentState)
    },
    OnCheckpointRestoredFunc: func(ctx *workflow.Context) error {
        restored, err := ctx.ReadState("StatefulExecutorState", "")
        if err != nil {
            return err
        }
        currentState = restored
        return nil
    },
}).Bind()

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

حواف