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

يربط سير العمل المنفذينوالحواف معا في رسم بياني موجه ويدير التنفيذ. وهو ينسق استدعاء المنفذ وتوجيه الرسائل وتدفق الأحداث.

إنشاء مهام سير العمل

يتم إنشاء مهام سير العمل باستخدام WorkflowBuilder الفئة التي توفر واجهة برمجة تطبيقات بطلاقة لتعريف بنية سير العمل:

using Microsoft.Agents.AI.Workflows;

var processor = new DataProcessor();
var validator = new Validator();
var formatter = new Formatter();

// Build workflow
WorkflowBuilder builder = new(processor); // Set starting executor
builder.AddEdge(processor, validator);
builder.AddEdge(validator, formatter);
var workflow = builder.Build();

يتم إنشاء مهام سير العمل باستخدام WorkflowBuilder الفئة :

from agent_framework import WorkflowBuilder

processor = DataProcessor()
validator = Validator()
formatter = Formatter()

# Build workflow
builder = WorkflowBuilder(start_executor=processor)
builder.add_edge(processor, validator)
builder.add_edge(validator, formatter)
workflow = builder.build()

أثناء التحقق من صحة وقت الإنشاء، يمكن للمعالج استخدام تعليق توضيحي للحاوية المجردة، مثل listأو dicttupleSequence. يمكنه قبول الإخراج الذي يستخدم الحاوية ذات المعلمات المقابلة، مثل list[str].

workflow توفر الحزمة نموذج تنفيذ يستند إلى الرسم البياني حيث يتم توصيل المنفذين بواسطة الحواف.

• المنفذ - وحدة معالجة تتلقى الإدخال وتنتج الإخراج
• Edge - يربط إخراج أحد المنفذين بإدخال آخر
• Builder - إنشاء مهام سير العمل عن طريق تعريف المنفذين والحواف
• تشغيل - تنفيذ سير عمل مع إدخال معين

تنشئ هذه الدالة سير عمل لمعالجة النص دون الحاجة إلى نموذج أو بيانات اعتماد:

import (
    "slices"
    "strings"

    "github.com/microsoft/agent-framework-go/workflow"
)

func buildWorkflow() (*workflow.Workflow, error) {
    uppercase := workflow.NewExecutor(
        "UppercaseExecutor", strings.ToUpper,
    ).Bind()

    reverse := workflow.NewExecutor(
        "ReverseExecutor", func(input string) string {
            runes := []rune(input)
            slices.Reverse(runes)
            return string(runes)
        },
    ).Bind()

    return workflow.NewBuilder(uppercase).
        AddEdge(uppercase, reverse).
        WithOutputFrom(reverse).
        Build()
}

WithOutputFrom(reverse) يعرض نتيجة المنفذ العكسي كإخراج سير عمل. لا تزال الحافة تقدم النتيجة الكبيرة إلى المنفذ العكسي بغض النظر عن تعيين الإخراج.

قم بتمرير سير العمل الذي تم إرجاعه إلى buildWorkflow أي من دالتي التنفيذ أدناه. بالنسبة للإدخل "Hello, World!"، اطبع Workflow output: !DLROW ,OLLEHكلاهما .

تنفيذ سير العمل

تدعم مهام سير العمل كلا من وضعي التنفيذ الدفق وغير المتدفق:

using Microsoft.Agents.AI.Workflows;

// Streaming execution — get events as they happen
StreamingRun run = await InProcessExecution.RunStreamingAsync(workflow, inputMessage);
await foreach (WorkflowEvent evt in run.WatchStreamAsync())
{
    if (evt is ExecutorCompletedEvent executorComplete)
    {
        Console.WriteLine($"{executorComplete.ExecutorId}: {executorComplete.Data}");
    }

    if (evt is WorkflowOutputEvent outputEvt)
    {
        Console.WriteLine($"Workflow completed: {outputEvt.Data}");
    }
}

// Non-streaming execution — wait for completion
Run result = await InProcessExecution.RunAsync(workflow, inputMessage);
foreach (WorkflowEvent evt in result.NewEvents)
{
    if (evt is WorkflowOutputEvent outputEvt)
    {
        Console.WriteLine($"Final result: {outputEvt.Data}");
    }
}

# Streaming execution — get events as they happen
async for event in workflow.run(input_message, stream=True):
    if event.type == "output":
        print(f"Workflow completed: {event.data}")

# Non-streaming execution — wait for completion
events = await workflow.run(input_message)
print(f"Final result: {events.get_outputs()}")

استخدم RunStreaming عندما تريد الأحداث عند حدوثها. قم بتمرير سياق المتصل حتى يصل الإلغاء إلى سير العمل، وتحقق من خطأه بعد انتهاء التكرار، وأغلق التشغيل عند الانتهاء:

import (
    "context"
    "errors"
    "fmt"

    "github.com/microsoft/agent-framework-go/workflow"
    "github.com/microsoft/agent-framework-go/workflow/inproc"
)

func runStreaming(
    ctx context.Context, wf *workflow.Workflow,
) (err error) {
    stream, err := inproc.Default.RunStreaming(ctx, wf, "Hello, World!")
    if err != nil {
        return err
    }
    defer func() {
        err = errors.Join(err, stream.Close(context.WithoutCancel(ctx)))
    }()

    for evt, streamErr := range stream.WatchStream(ctx) {
        if streamErr != nil {
            return streamErr
        }
        switch event := evt.(type) {
        case workflow.OutputEvent:
            fmt.Printf("Workflow output: %v\n", event.Output)
        case workflow.ErrorEvent:
            return event.Error
        case workflow.ExecutorFailedEvent:
            return fmt.Errorf("executor %q failed: %w",
                event.ExecutorID, event.Error)
        }
    }
    return ctx.Err()
}

WatchStream الانتظار بينما الطلبات الخارجية معلقة. بالنسبة إلى مهام سير العمل التي تحتاج إلى إدخال بشري، قم بمعالجة أحداث الطلب أو استخدمها WatchUntilHalt للرجوع مؤقتا. راجع أحداث سير العمل.

استخدم Run للتقدم إلى التوقف التالي ثم فحص الأحداث التي تم جمعها. يمكن أن يتوقف التشغيل أثناء انتظار الإدخال الخارجي، ليس فقط عند اكتمال عمله:

import (
    "context"
    "errors"
    "fmt"

    "github.com/microsoft/agent-framework-go/workflow"
    "github.com/microsoft/agent-framework-go/workflow/inproc"
)

func runToHalt(
    ctx context.Context, wf *workflow.Workflow,
) (err error) {
    run, err := inproc.Default.Run(ctx, wf, "Hello, World!")
    if err != nil {
        return err
    }
    defer func() {
        err = errors.Join(err, run.Close(context.WithoutCancel(ctx)))
    }()

    for evt := range run.NewEvents() {
        switch event := evt.(type) {
        case workflow.OutputEvent:
            fmt.Printf("Workflow output: %v\n", event.Output)
        case workflow.ErrorEvent:
            return event.Error
        case workflow.ExecutorFailedEvent:
            return fmt.Errorf("executor %q failed: %w",
                event.ExecutorID, event.Error)
        }
    }

    for evt := range run.OutgoingEvents() {
        if evt, ok := evt.(workflow.ExecutorCompletedEvent); ok {
            fmt.Printf("%s: %v\n", evt.ExecutorID, evt.Result)
        }
    }
    return ctx.Err()
}

NewEvents() تقدم الإشارة المرجعية الخاصة بها في كل مرة تقوم فيها باستدعائها. تستخدم OutgoingEvents() الحلقة الثانية لإعادة تشغيل أحداث المنفذ المتراكمة بدلا من محاولة استهلاكها مرة أخرى مع NewEvents().

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

راجع أحداث سير العمل للمستهلكين الكاملين وتصنيف الإخراج ودلالات الإيقاف المؤقت.

Tip

راجع أمثلة سير العمل للحصول على عينات كاملة قابلة للتشغيل.

التحقق من صحة سير العمل

يقوم منشئو سير العمل بالتحقق من صحة الرسم البياني قبل التنفيذ. تعتمد عمليات التحقق الدقيقة على SDK وبيانات تعريف النوع المتوفرة:

• توافق النوع: يتحقق من بروتوكولات الرسائل المعلن عنها بين المنفذين المتصلين حيث تتوفر بيانات التعريف.
• اتصال الرسم البياني: يتحقق من إمكانية الوصول إلى المنفذين المسجلين من منفذ البداية.
• ربط المنفذ: يتحقق من وجود تسجيلات المنفذ المطلوبة.
• التحقق من صحة الحافة: يتحقق من وجود اتصالات غير صالحة والحواف المكررة وفقا لقواعد المنشئ.

ترفض Go Build() الروابط التي لم يتم حلها، والمنفذين المسجلين غير القابلين للوصول، والحواف غير المشروطة المكررة، وأنواع الرسائل المعلنة غير المتوافقة. لا تزال المسارات الشرطية والبروتوكولات الديناميكية تتطلب اختبار وقت التشغيل.

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

BuildWithOptions(workflow.BuildOptions{DisableOrphanValidation: true}) يعطل فحص المنفذ المعزول فقط. لا يزال التحقق من الصحة الآخر قيد التشغيل.

نموذج التنفيذ: خطوات فائقة

يستخدم إطار العمل نموذج تنفيذ Pregel معدل — نهج متوازي متزامن مجمع (BSP) مع معالجة تستند إلى فائقة.

كيف يعمل Supersteps

يتم تنظيم تنفيذ سير العمل في عمليات فائقة منفصلة. كل فائقة:

1. جمع كافة الرسائل المعلقة من الخطوة السابقة
2. توجيه الرسائل إلى المنفذين المستهدفين استنادا إلى تعريفات الحافة
3. تشغيل جميع المنفذين المستهدفين في وقت واحد ضمن مسافة فائقة
4. انتظار اكتمال جميع المنفذين قبل التقدم (حاجز المزامنة)
5. وضع أي رسائل جديدة منبعثة من المنفذين في قائمة الانتظار للرسالة الفائقة التالية

Superstep N:
┌─────────────────┐    ┌─────────────────┐    ┌─────────────────┐
│  Collect All    │───▶│  Route Messages │───▶│  Execute All    │
│  Pending        │    │  Based on Type  │    │  Target         │
│  Messages       │    │  & Conditions   │    │  Executors      │
└─────────────────┘    └─────────────────┘    └─────────────────┘
                                                       │
                                                       │ (barrier: wait for all)
┌─────────────────┐    ┌─────────────────┐             │
│  Start Next     │◀───│  Emit Events &  │◀────────────┘
│  Superstep      │    │  New Messages   │
└─────────────────┘    └─────────────────┘

حاجز المزامنة

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

يؤثر هذا على أنماط توزيع المهام: إذا قمت بالتطوير إلى مسارات متعددة — واحد مع سلسلة من المنفذين وآخر مع منفذ واحد طويل الأمد — لا يمكن أن يتقدم المسار المتسلسل حتى يكتمل المنفذ طويل الأمد.

لماذا الفوقية؟

يوفر نموذج BSP ضمانات مهمة:

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

لا تصلح هذه الضمانات التنفيذ أو ترتيب الأحداث بين المنفذين المتزامنين ضمن خط واحد. مزامنة الوصول إلى الحالة المشتركة المملوكة للتطبيق بدلا من افتراض أن حاجز التراكب يمنع جميع الأعراق.

العمل مع نموذج Superstep

إذا كنت بحاجة إلى مسارات متوازية مستقلة حقا لا تمنع بعضها البعض، فدمج الخطوات المتسلسلة في منفذ واحد. بدلا من التسلسل step1 → step2 → step3، ادمج هذا المنطق في منفذ واحد. ثم يتم تنفيذ كلا المسارين المتوازيين ضمن خط واحد.

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

الوكلاء في مهام سير العمل

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

• المنفذون — وحدات المعالجة في سير العمل
• الحواف — الاتصالات بين المنفذين
• الأحداث — إمكانية مراقبة سير العمل
• إدارة الدولة