تنسيقات مهام سير عمل Microsoft Agent Framework - أرجواني

تم تصميم التنسيق الأرجواني استنادا إلى نظام Magentic-One الذي اخترعه AutoGen. وهو نمط مرن متعدد العوامل للأغراض العامة مصمم للمهام المعقدة المفتوحة التي تتطلب تعاونا ديناميكيا. في هذا النمط، يقوم مدير Magentic مخصص بتنسيق فريق من الوكلاء المتخصصين، وتحديد العامل الذي يجب أن يعمل بعد ذلك بناء على السياق المتطور، وتقدم المهام، وقدرات العامل.

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

تنسيق أرجواني

Tip

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

Note

في ورقة Magentic-One ، تم تصميم 4 وكلاء متخصصين للغاية لحل مجموعة محددة جدا من المهام. في التنسيق Magentic في إطار عمل العامل، يمكنك تحديد وكلاء متخصصين خاصين بك ليناسبوا احتياجات التطبيق الخاصة بك. ومع ذلك، يتم اختبار مدى جودة أداء التنسيق الماجنتي خارج تصميم Magentic-One الأصلي.

ما ستتعلمه

كيفية إعداد مدير Magentic لتنسيق عوامل متخصصة متعددة
كيفية التعامل مع أحداث الدفق باستخدام WorkflowEvent
كيفية تنفيذ مراجعة خطة الإنسان في التكرار الحلقي
كيفية تتبع تعاون العامل وتقدمه من خلال المهام المعقدة

تعريف وكلاءك المتخصصين

في التنسيق Magentic، يمكنك تحديد العوامل المتخصصة التي يمكن للمدير تحديدها ديناميكيا استنادا إلى متطلبات المهمة:

#pragma warning disable MAAIW001  // Magentic types are experimental
#pragma warning disable OPENAI001 // HostedCodeInterpreterTool is experimental

using Azure.AI.Projects;
using Azure.Identity;
using Microsoft.Agents.AI;
using Microsoft.Agents.AI.Workflows;
using Microsoft.Agents.AI.Workflows.Specialized.Magentic;
using Microsoft.Extensions.AI;

string endpoint = Environment.GetEnvironmentVariable("AZURE_AI_PROJECT_ENDPOINT")
    ?? throw new InvalidOperationException("AZURE_AI_PROJECT_ENDPOINT is not set.");
string deploymentName = Environment.GetEnvironmentVariable("AZURE_AI_MODEL_DEPLOYMENT_NAME") ?? "gpt-5.4-mini";

AIProjectClient projectClient = new(new Uri(endpoint), new DefaultAzureCredential());

AIAgent researcherAgent = projectClient.AsAIAgent(
    deploymentName,
    name: "ResearcherAgent",
    description: "Specialist in research and information gathering.",
    instructions: "You are a researcher. Find relevant information without doing additional computation or quantitative analysis.");

AIAgent coderAgent = projectClient.AsAIAgent(
    deploymentName,
    name: "CoderAgent",
    description: "A helpful assistant that writes and executes code to analyze data.",
    instructions: "You solve quantitative questions by writing and running code. Show the analysis and the computation process clearly.",
    tools: [new HostedCodeInterpreterTool()]);

AIAgent managerAgent = projectClient.AsAIAgent(
    deploymentName,
    name: "MagenticManager",
    description: "Orchestrator that coordinates the research and coding workflow.",
    instructions: "You coordinate the team to complete complex tasks efficiently.");

import os

from agent_framework import Agent
from agent_framework.foundry import FoundryChatClient
from azure.identity import AzureCliCredential

client = FoundryChatClient(
    project_endpoint=os.environ["FOUNDRY_PROJECT_ENDPOINT"],
    model=os.environ["FOUNDRY_MODEL"],
    credential=AzureCliCredential(),
)

researcher_agent = Agent(
    name="ResearcherAgent",
    description="Specialist in research and information gathering",
    instructions=(
        "You are a Researcher. You find information without additional computation or quantitative analysis."
    ),
    client=client,
)

coder_agent = Agent(
    name="CoderAgent",
    description="A helpful assistant that writes and executes code to process and analyze data.",
    instructions="You solve questions using code. Please provide detailed analysis and computation process.",
    client=client,
    tools=client.get_code_interpreter_tool(),
)

# Create a manager agent for orchestration
manager_agent = Agent(
    name="MagenticManager",
    description="Orchestrator that coordinates the research and coding workflow",
    instructions="You coordinate a team to complete complex tasks efficiently.",
    client=client,
)

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

استخدم منشئ سير عمل Magentic لتكوين سير العمل مع مدير ومجموعة من المشاركين. يعرض المنشئ أيضا حدود الحلقة الداخلية (جولات التنسيق القصوى، والحد الأقصى للأكشاك المتتالية قبل إعادة التخطيط، وإعادة تعيين الحد الأقصى للخطة) وعلامة لمراجعة خطة الإنسان في الحلقة.

Workflow workflow = new MagenticWorkflowBuilder(managerAgent)
    .AddParticipants([researcherAgent, coderAgent])
    .WithName("Magentic Orchestration Workflow")
    .WithDescription("Coordinates a researcher and coder to solve a complex analytical task.")
    .RequirePlanSignoff(false)
    .WithMaxRounds(10)
    .WithMaxStalls(3)
    .WithMaxResets(2)
    .Build();

from agent_framework.orchestrations import MagenticBuilder

workflow = MagenticBuilder(
    participants=[researcher_agent, coder_agent],
    intermediate_output_from=[researcher_agent, coder_agent],
    manager_agent=manager_agent,
    max_round_count=10,
    max_stall_count=3,
    max_reset_count=2,
).build()

Tip

يتم تنفيذ مدير قياسي استنادا إلى تصميم Magentic-One، مع المطالبات الثابتة المأخوذة من الورق الأصلي. يمكنك تخصيص سلوك المدير عن طريق تمرير المطالبات الخاصة بك عبر معلمات الدالة MagenticBuilder الإنشائية. لتخصيص المدير بشكل أكبر، يمكنك أيضا تنفيذ مديرك الخاص عن طريق تصنيف الفئة الفرعية MagenticManagerBase .

مخرجات متوسطة

Note

ينطبق هذا المقطع حاليا على المحور Python فقط.

intermediate_output_from=[...] التمرير إلى MagenticBuilder تعيين مشاركين محددين كمصادر إخراج وسيطة. تنبعث "intermediate" من مكالماتهم yield_output أحداث، بينما تظل إجابة المدير النهائية المركب حدثا "output" (محطة طرفية). بدون هذه المعلمة (الافتراضي)، يتم عرض المحطة الطرفية AgentResponse للمدير فقط.

هذا مفيد بشكل خاص لسير العمل Magentic لأن:

غالبا ما تكون المهام طويلة الأمد مع العديد من جولات تعاون العامل
يمكنك عرض مساهمة كل عامل في الوقت الفعلي مع تقدم سير العمل في وضع الدفق
يوفر رؤية لخطوات المنطق الوسيط لسير العمل

تشغيل سير العمل باستخدام تدفق الأحداث

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

const string TaskPrompt =
    "I am preparing a report on the energy efficiency of different machine learning model architectures. " +
    "Compare the estimated training and inference energy consumption of ResNet-50, BERT-base, and GPT-2 " +
    "on standard datasets (for example, ImageNet for ResNet, GLUE for BERT, WebText for GPT-2). " +
    "Then, estimate the CO2 emissions associated with each, assuming training on an Azure Standard_NC6s_v3 " +
    "VM for 24 hours. Provide tables for clarity, and recommend the most energy-efficient model " +
    "per task type (image classification, text classification, and text generation).";

await using StreamingRun run = await InProcessExecution.RunStreamingAsync(
    workflow,
    new List<ChatMessage> { new(ChatRole.User, TaskPrompt) });

await run.TrySendMessageAsync(new TurnToken(emitEvents: true));

string? lastResponseId = null;
WorkflowOutputEvent? finalOutput = null;

await foreach (WorkflowEvent workflowEvent in run.WatchStreamAsync())
{
    switch (workflowEvent)
    {
        case AgentResponseUpdateEvent updateEvent:
            // Stream per-participant deltas. Group by ResponseId / MessageId / ExecutorId so
            // each new contiguous response prints its executor header once.
            string responseId = updateEvent.Update.ResponseId
                ?? updateEvent.Update.MessageId
                ?? updateEvent.ExecutorId;
            if (!string.Equals(responseId, lastResponseId, StringComparison.Ordinal))
            {
                if (lastResponseId is not null)
                {
                    Console.WriteLine();
                }
                Console.Write($"- {updateEvent.ExecutorId}: ");
                lastResponseId = responseId;
            }
            Console.Write(updateEvent.Update.Text);
            break;

        case MagenticPlanCreatedEvent planCreated:
            Console.WriteLine($"\n[Magentic Initial Plan]\n{planCreated.FullTaskLedger.Text}");
            break;

        case MagenticReplannedEvent replanned:
            Console.WriteLine($"\n[Magentic Replanned]\n{replanned.FullTaskLedger.Text}");
            break;

        case MagenticProgressLedgerUpdatedEvent progressUpdated:
            MagenticProgressLedger ledger = progressUpdated.ProgressLedger;
            Console.WriteLine(
                $"\n[Magentic Progress Ledger] satisfied={ledger.IsRequestSatisfied}, " +
                $"inLoop={ledger.IsInLoop}, progressing={ledger.IsProgressBeingMade}, " +
                $"nextSpeaker={ledger.NextSpeaker}, instruction={ledger.InstructionOrQuestion}");
            break;

        case WorkflowOutputEvent outputEvent when outputEvent.Is<List<ChatMessage>>():
            finalOutput = outputEvent;
            break;

        case WorkflowErrorEvent workflowError:
            Console.Error.WriteLine(workflowError.Exception?.ToString() ?? "Unknown workflow error.");
            break;

        case ExecutorFailedEvent executorFailed:
            Console.Error.WriteLine(
                $"Executor '{executorFailed.ExecutorId}' failed: " +
                (executorFailed.Data?.ToString() ?? "unknown error"));
            break;
    }
}

if (finalOutput?.As<List<ChatMessage>>() is { } transcript)
{
    Console.WriteLine("\n\n=== Final Conversation Transcript ===\n");
    foreach (ChatMessage message in transcript)
    {
        Console.WriteLine($"{message.AuthorName ?? message.Role.ToString()}: {message.Text}");
    }
}

import json
import asyncio
from typing import cast

from agent_framework import (
    AgentResponseUpdate,
    Message,
    WorkflowEvent,
)
from agent_framework.orchestrations import MagenticProgressLedger

task = (
    "I am preparing a report on the energy efficiency of different machine learning model architectures. "
    "Compare the estimated training and inference energy consumption of ResNet-50, BERT-base, and GPT-2 "
    "on standard datasets (for example, ImageNet for ResNet, GLUE for BERT, WebText for GPT-2). "
    "Then, estimate the CO2 emissions associated with each, assuming training on an Azure Standard_NC6s_v3 "
    "VM for 24 hours. Provide tables for clarity, and recommend the most energy-efficient model "
    "per task type (image classification, text classification, and text generation)."
)

# Keep track of the last executor to format output nicely in streaming mode
last_message_id: str | None = None
stream = workflow.run(task, stream=True)
async for event in stream:
    if event.type in ("intermediate", "output") and isinstance(event.data, AgentResponseUpdate):
        message_id = event.data.message_id
        if message_id != last_message_id:
            if last_message_id is not None:
                print("\n")
            print(f"- {event.executor_id}:", end=" ", flush=True)
            last_message_id = message_id
        print(event.data, end="", flush=True)

    elif event.type == "magentic_orchestrator":
        print(f"\n[Magentic Orchestrator Event] Type: {event.data.event_type.name}")
        if isinstance(event.data.content, Message):
            print(f"Please review the plan:\n{event.data.content.text}")
        elif isinstance(event.data.content, MagenticProgressLedger):
            print(f"Please review progress ledger:\n{json.dumps(event.data.content.to_dict(), indent=2)}")
        else:
            print(f"Unknown data type in MagenticOrchestratorEvent: {type(event.data.content)}")

        # Block to allow user to read the plan/progress before continuing
        # Note: this is for demonstration only and is not the recommended way to handle human interaction.
        # Please refer to `with_plan_review` for proper human interaction during planning phases.
        await asyncio.get_event_loop().run_in_executor(None, input, "Press Enter to continue...")

result = await stream.get_final_response()
if outputs := result.get_outputs():
    print(outputs[-1])

يظهر Magentic ثلاثة أحداث منسق تضع علامة على مراحل التخطيط والتقدم:

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

في Python يتم نقلها داخل واحد MagenticOrchestratorEvent الذي يميز PLAN_CREATEDتعداده event_type و REPLANNEDو.PROGRESS_LEDGER_UPDATED في .NET يتم إصدارها على شكل ثلاثة أنواع مميزة — MagenticPlanCreatedEventو MagenticReplannedEventو MagenticProgressLedgerUpdatedEvent — وكلها مشتقة من MagenticOrchestratorEvent.

Advanced: Human-in-the-Loop Plan Review

تمكين human-in-the-loop (HITL) للسماح للمستخدمين بمراجعة الخطة المقترحة للمدير والموافقة عليها قبل التنفيذ. وهذا مفيد لضمان توافق الخطة مع توقعات المستخدمين ومتطلباتهم.

هناك خياران لمراجعة الخطة:

مراجعة: يقدم المستخدم ملاحظات لمراجعة الخطة، ما يؤدي إلى تشغيل المدير لإعادة التخطيط استنادا إلى الملاحظات.
الموافقة: يوافق المستخدم على الخطة as-is، مما يسمح بمتابعة سير العمل.

تمكين مراجعة الخطة عند إنشاء سير عمل Magentic. تختلف الإعدادات الافتراضية بين اللغات: في Python، تكون مراجعة الخطة متوقفة عن التشغيل بشكل افتراضي (enable_plan_review=False) ويمكنك الاشتراك بشكل صريح؛ في .NET، تكون مراجعة الخطة قيد التشغيل بشكل افتراضي (RequirePlanSignoff افتراضيات إلى true)، وتم إلغاء الاشتراك في المثال الأساسي السابق في هذه الصفحة بحيث يمكن تشغيلها من طرف إلى طرف دون تفاعل. توضح التعليمات البرمجية أدناه كيفية الاشتراك والتعامل مع طلبات المراجعة الناتجة.

يتم عرض الإيقاف المؤقت لمراجعة الخطة من خلال آلية الطلب/الاستجابة لسير العمل مع MagenticPlanReviewRequest البيانات. يمكنك التعامل مع هذه في دفق الحدث واستئناف سير العمل بمجرد MagenticPlanReviewResponse موافقة الإنسان على الخطة أو مراجعتها.

Tip

تعرف على المزيد حول الطلبات والاستجابات في دليل الطلبات والاستجابات .

Workflow workflow = new MagenticWorkflowBuilder(managerAgent)
    .AddParticipants([researcherAgent, coderAgent])
    .RequirePlanSignoff(true)
    .WithMaxRounds(10)
    .WithMaxStalls(1)
    .WithMaxResets(2)
    .Build();

CheckpointManager checkpointManager = CheckpointManager.CreateInMemory();
InProcessExecutionEnvironment environment = ExecutionEnvironment.InProcess_Lockstep
    .ToWorkflowExecutionEnvironment()
    .WithCheckpointing(checkpointManager);

await using StreamingRun run = await environment.OpenStreamingAsync(workflow);
await run.TrySendMessageAsync(new List<ChatMessage> { new(ChatRole.User, TaskPrompt) });
await run.TrySendMessageAsync(new TurnToken(emitEvents: true));

ExternalRequest? pendingRequest = null;
CheckpointInfo? lastCheckpoint = null;
WorkflowOutputEvent? finalOutput = null;

async Task<WorkflowOutputEvent?> DrainAsync(StreamingRun activeRun)
{
    WorkflowOutputEvent? output = null;
    await foreach (WorkflowEvent evt in activeRun.WatchStreamAsync(blockOnPendingRequest: false))
    {
        switch (evt)
        {
            case AgentResponseUpdateEvent updateEvent:
                Console.Write(updateEvent.Update.Text);
                break;
            case RequestInfoEvent requestInfo
                when requestInfo.Request.Data.As<MagenticPlanReviewRequest>() is not null:
                pendingRequest = requestInfo.Request;
                break;
            case SuperStepCompletedEvent stepCompleted:
                lastCheckpoint = stepCompleted.CompletionInfo?.Checkpoint ?? lastCheckpoint;
                break;
            case WorkflowOutputEvent outputEvent when outputEvent.Is<List<ChatMessage>>():
                output = outputEvent;
                break;
        }
    }
    return output;
}

finalOutput = await DrainAsync(run);

// Loop until the workflow finishes or the user accepts a plan that runs to completion.
while (finalOutput is null && pendingRequest is not null)
{
    MagenticPlanReviewRequest reviewRequest = pendingRequest.Data.As<MagenticPlanReviewRequest>()!;

    Console.WriteLine("\n\n[Magentic Plan Review Request]");
    if (reviewRequest.CurrentProgress is { } progress)
    {
        Console.WriteLine(
            $"Current progress: satisfied={progress.IsRequestSatisfied}, " +
            $"inLoop={progress.IsInLoop}, progressing={progress.IsProgressBeingMade}");
    }
    if (reviewRequest.IsStalled)
    {
        Console.WriteLine("(Replan triggered by stall detection.)");
    }
    Console.WriteLine($"Proposed plan:\n{reviewRequest.Plan.Text}\n");
    Console.Write("Press Enter to approve, or type feedback to request a revision: ");

    string reply = Console.ReadLine() ?? string.Empty;
    MagenticPlanReviewResponse reviewResponse = string.IsNullOrWhiteSpace(reply)
        ? reviewRequest.Approve()
        : reviewRequest.Revise(reply);

    ExternalResponse response = pendingRequest.CreateResponse(reviewResponse);
    pendingRequest = null;

    await using StreamingRun resumed = await environment.ResumeStreamingAsync(workflow, lastCheckpoint!);
    await resumed.SendResponseAsync(response);
    finalOutput = await DrainAsync(resumed);
}

if (finalOutput?.As<List<ChatMessage>>() is { } transcript)
{
    Console.WriteLine("\n\n=== Final Conversation Transcript ===\n");
    foreach (ChatMessage message in transcript)
    {
        Console.WriteLine($"{message.AuthorName ?? message.Role.ToString()}: {message.Text}");
    }
}

import json
import asyncio
from typing import cast

from agent_framework import (
    AgentResponseUpdate,
    Agent,
    Message,
    WorkflowEvent,
)
from agent_framework.orchestrations import (
    MagenticBuilder,
    MagenticPlanReviewRequest,
    MagenticPlanReviewResponse,
)

workflow = MagenticBuilder(
    participants=[researcher_agent, coder_agent],
    intermediate_output_from=[researcher_agent, coder_agent],
    enable_plan_review=True,
    manager_agent=manager_agent,
    max_round_count=10,
    max_stall_count=1,
    max_reset_count=2,
).build()

pending_request: WorkflowEvent | None = None
pending_responses: dict[str, MagenticPlanReviewResponse] | None = None
final_response: object | None = None

while not final_response:
    if pending_responses is not None:
        stream = workflow.run(stream=True, responses=pending_responses)
    else:
        stream = workflow.run(task, stream=True)

    last_message_id: str | None = None
    async for event in stream:
        if event.type in ("intermediate", "output") and isinstance(event.data, AgentResponseUpdate):
            message_id = event.data.message_id
            if message_id != last_message_id:
                if last_message_id is not None:
                    print("\n")
                print(f"- {event.executor_id}:", end=" ", flush=True)
                last_message_id = message_id
            print(event.data, end="", flush=True)

        elif event.type == "request_info" and event.request_type is MagenticPlanReviewRequest:
            pending_request = event

    result = await stream.get_final_response()
    if outputs := result.get_outputs():
        final_response = outputs[-1]

    pending_responses = None

    # Handle plan review request if any
    if pending_request is not None:
        event_data = cast(MagenticPlanReviewRequest, pending_request.data)

        print("\n\n[Magentic Plan Review Request]")
        if event_data.current_progress is not None:
            print("Current Progress Ledger:")
            print(json.dumps(event_data.current_progress.to_dict(), indent=2))
            print()
        print(f"Proposed Plan:\n{event_data.plan.text}\n")
        print("Please provide your feedback (press Enter to approve):")

        reply = await asyncio.get_event_loop().run_in_executor(None, input, "> ")
        if reply.strip() == "":
            print("Plan approved.\n")
            pending_responses = {pending_request.request_id: event_data.approve()}
        else:
            print("Plan revised by human.\n")
            pending_responses = {pending_request.request_id: event_data.revise(reply)}
        pending_request = None

MagenticPlanReviewRequest يحمل الخطة المقترحة، ودفتر الأستاذ الحالي للتقدم (null / Noneفي المراجعة الأولية ويتم ملؤها على خطط إعادة تشغيل المماطلة)، وعلامة تشير إلى ما إذا كان قد تم تشغيل إعادة الخطة عن طريق الكشف عن المكب. بناء الاستجابة عن طريق الاتصال إما approve() بقبول الخطة as-is، أو revise(...) مع الملاحظات لمطالبة المدير بإعادة التخطيط.

المفاهيم الأساسية

التنسيق الديناميكي: يختار مدير Magentic ديناميكيا العامل الذي يجب أن يعمل بعد ذلك بناء على السياق المتطور.
إخراج المحطة الطرفية: يحمل إخراج سير عمل المحطة الطرفية الإجابة النهائية المركب للمدير (AgentResponseفي Python؛ و WorkflowOutputEvent مع List<ChatMessage> حمولة في .NET).
أحداث المنسق: تظهر MagenticOrchestratorEvent المراحل الرئيسية التي تم إنشاؤها بواسطة الخطة وإعادة تخطيطها وتحديثها بواسطة دفتر الأستاذ المتقدم (حدث واحد مع event_type قائمة تعداد في Python؛ وثلاثة أنواع مشتقة في .NET). يتم تسليم دلتا الدفق لكل مشارك من خلال أحداث تحديث استجابة العامل القياسية لإطار العمل.
التحسين التكراري: يمكن للنظام تقسيم المشاكل المعقدة وتحسين الحلول بشكل متكرر من خلال جولات متعددة.
تتبع التقدم والكشف عن الأكشاك: يتعقب دفتر الأستاذ التقدم ما إذا كان الطلب راضيا، وما إذا كان الفريق في حلقة، وما إذا كان يتم إحراز تقدم. تؤدي الجولات غير المتقدمة المتتالية إلى زيادة عداد حجرة الانتظار، وتجاوز الحد الأقصى الذي تم تكوينه يؤدي إلى إعادة تعيين تلقائي وإعادة تخطيط.
التعاون المرن: يمكن استدعاء الوكلاء عدة مرات بأي ترتيب كما يحدده المدير.
الرقابة البشرية: مراجعة اختيارية لخطة الإنسان في الحلقة عبر MagenticPlanReviewRequest / MagenticPlanReviewResponse.
المخرجات المتوسطة (Python فقط، في الوقت الحالي): تعيين المشاركين الذين يجب أن تظهر مكالماتهم yield_output كأحداث "intermediate" إلى جانب إخراج المحطة الطرفية للمدير.

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

يتبع التنسيق Magentic نمط التنفيذ هذا:

مرحلة التخطيط: يقوم المدير بتحليل المهمة وإنشاء خطة أولية
مراجعة الخطة الاختيارية: إذا تم تمكينها، يمكن للبشر مراجعة الخطة والموافقة عليها/تعديلها
تحديد العامل: يحدد المدير العامل الأنسب لكل مهام فرعية
التنفيذ: ينفذ العامل المحدد جزء المهمة الخاص به
تقييم التقدم: يقوم المدير بتقييم التقدم وتحديث الخطة
الكشف عن الأكشاك: إذا توقف التقدم، أعد التخطيط التلقائي باستخدام عملية مراجعة بشرية اختيارية
التكرار: تتكرر الخطوات من 3 إلى 6 حتى تكتمل المهمة أو يتم الوصول إلى الحدود
التركيب النهائي: يقوم المدير بتجميع جميع مخرجات العامل في نتيجة نهائية

مثال كامل

راجع العينات الكاملة في مستودع Agent Framework Samples.

Note

سيتوفر الدعم لهذه الميزة قريبا. راجع مستودع Agent Framework Go للحصول على أحدث حالة.

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

تنسيق التسليم

الملاحظات

هل كانت هذه الصفحة مفيدة؟

Last updated on 2026-08-25

تنسيقات مهام سير عمل Microsoft Agent Framework - أرجواني

ما ستتعلمه

تعريف وكلاءك المتخصصين

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

مخرجات متوسطة

تشغيل سير العمل باستخدام تدفق الأحداث

Advanced: Human-in-the-Loop Plan Review

المفاهيم الأساسية

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

مثال كامل

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

الملاحظات

الموارد الإضافية