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

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

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

يتم إنشاء مهام سير العمل باستخدام 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()

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

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

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

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

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

wf, err := workflow.NewBuilder(uppercase).
    AddEdge(uppercase, reverse).
    WithOutputFrom(reverse).
    Build()
if err != nil {
    return err
}

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

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

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 عندما تريد الأحداث عند حدوثها:

stream, err := inproc.Default.RunStreaming(context.Background(), wf, "Hello, World!")
if err != nil {
    return err
}
defer stream.Close(context.Background())

for evt, err := range stream.WatchStream(context.Background()) {
    if err != nil {
        return err
    }
    if output, ok := evt.(workflow.OutputEvent); ok {
        fmt.Printf("Workflow completed: %v\n", output.Output)
    }
}

استخدم Run عندما تريد انتظار اكتمال سير العمل ثم افحص الأحداث التي تم جمعها:

run, err := inproc.Default.Run(context.Background(), wf, "Hello, World!")
if err != nil {
    return err
}

for evt := range run.NewEvents() {
    if output, ok := evt.(workflow.OutputEvent); ok {
        fmt.Printf("Final result: %v\n", output.Output)
    }
}

يمكنك أيضا فحص أحداث المنفذ التي تم جمعها بواسطة تشغيل غير متدفق:

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

Tip

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

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

يقوم إطار العمل بإجراء تحقق شامل عند إنشاء مهام سير العمل:

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

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

يستخدم إطار العمل نموذج تنفيذ 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، ادمج هذا المنطق في منفذ واحد. ثم يتم تنفيذ كلا المسارين المتوازيين ضمن خط واحد.

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

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

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

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