حواف

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

أنواع الحواف

يدعم إطار العمل العديد من أنماط الحافة:

Type	Description	حالة الاستخدام
Direct	اتصالات بسيطة من واحد إلى واحد	البنية الأساسية لبرنامج ربط العمليات التجارية الخطية
شرطي	الحواف ذات الشروط التي تحدد متى تتدفق الرسائل	التوجيه الثنائي (if/else)
تبديل حالة الأحرف	التوجيه إلى منفذين مختلفين استنادا إلى الشروط	التوجيه متعدد الفروع
تحديد متعدد (توزيع المهام)	منفذ واحد يرسل رسائل إلى أهداف متعددة	معالجة متوازية
المروحة	منفذون متعددون يرسلون إلى هدف واحد	تجميع

الحواف المباشرة

أبسط نموذج — قم بتوصيل منفذين دون شروط:

WorkflowBuilder builder = new(sourceExecutor);
builder.AddEdge(sourceExecutor, targetExecutor);

builder = WorkflowBuilder(start_executor=source_executor)
builder.add_edge(source_executor, target_executor)
workflow = builder.build()

wf, err := workflow.NewBuilder(sourceExecutor).
    AddEdge(sourceExecutor, targetExecutor).
    Build()

المروحة في الحواف

جمع الرسائل من مصادر متعددة في هدف واحد:

builder.AddFanInBarrierEdge(sources: [ worker1, worker2, worker3 ], target: aggregatorExecutor);

builder.add_fan_in_edges([worker1, worker2, worker3], aggregator_executor)

workers := []workflow.ExecutorBinding{worker1, worker2, worker3}

wf, err := workflow.NewBuilder(startExecutor).
    AddFanOutEdge(startExecutor, workers).
    AddFanInBarrierEdge(workers, aggregatorExecutor).
    Build()

توفر الأقسام أدناه برامج تعليمية مفصلة للحافات الشرطية والتبديلية ومتعددة التحديدات.

الحواف الشرطية

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

ما ستقوم بإنشاءه

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

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

المفاهيم التي تمت تغطيتها

• الحواف الشرطية

المتطلبات الأساسية

• .NET 8.0 SDK أو أحدث.
• Azure نقطة نهاية خدمة OpenAI ونشرها المكونة.
• Azure CLI تثبيتهاومصادقتها (لمصادقة بيانات الاعتماد Azure).
• الفهم الأساسي ل C# والبرمجة غير المتزامنة.
• تطبيق وحدة تحكم جديد.

ثبِّت حزم NuGet

أولا، قم بتثبيت الحزم المطلوبة لمشروع .NET الخاص بك:

dotnet add package Microsoft.Agents.AI.Workflows --prerelease
dotnet add package Microsoft.Agents.AI.Workflows.Generators --prerelease
dotnet add package Microsoft.Agents.AI.Foundry --prerelease

Microsoft.Agents.AI.Workflows.Generators توفر الحزمة منشئ المصدر الذي يسجل الأساليب [MessageHandler] في partial فئات المنفذ. لا تتدفق حزم منشئ المصدر بشكل عابر، لذا قم بالرجوع إليها مباشرة في المشروع الذي يحدد المنفذين.

تعريف نماذج البيانات

ابدأ بتعريف بنيات البيانات التي ستتدفق عبر سير العمل الخاص بك:

using System.Text.Json.Serialization;

/// <summary>
/// Represents the result of spam detection.
/// </summary>
public sealed class DetectionResult
{
    [JsonPropertyName("is_spam")]
    public bool IsSpam { get; set; }

    [JsonPropertyName("reason")]
    public string Reason { get; set; } = string.Empty;

    // Email ID is generated by the executor, not the agent
    [JsonIgnore]
    public string EmailId { get; set; } = string.Empty;
}

/// <summary>
/// Represents an email.
/// </summary>
internal sealed class Email
{
    [JsonPropertyName("email_id")]
    public string EmailId { get; set; } = string.Empty;

    [JsonPropertyName("email_content")]
    public string EmailContent { get; set; } = string.Empty;
}

/// <summary>
/// Represents the response from the email assistant.
/// </summary>
public sealed class EmailResponse
{
    [JsonPropertyName("response")]
    public string Response { get; set; } = string.Empty;
}

/// <summary>
/// Constants for shared state scopes.
/// </summary>
internal static class EmailStateConstants
{
    public const string EmailStateScope = "EmailState";
}

إنشاء دالات الشرط

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

/// <summary>
/// Creates a condition for routing messages based on the expected spam detection result.
/// </summary>
/// <param name="expectedResult">The expected spam detection result</param>
/// <returns>A function that evaluates whether a message meets the expected result</returns>
private static Func<object?, bool> GetCondition(bool expectedResult) =>
    detectionResult => detectionResult is DetectionResult result && result.IsSpam == expectedResult;

دالة الشرط هذه:

• يأخذ معلمة bool expectedResult (صحيحة للبريد العشوائي، خطأ لغير البريد العشوائي)
• إرجاع دالة يمكن استخدامها كشرط حافة
• التحقق بأمان من أن الرسالة هي DetectionResult ومقارنة الخاصية IsSpam

إنشاء عوامل الذكاء الاصطناعي

إعداد وكلاء الذكاء الاصطناعي الذين سيتعاملون مع الكشف عن البريد العشوائي والمساعدة في البريد الإلكتروني:

using Azure.AI.Projects;
using Azure.Identity;
using Microsoft.Agents.AI;
using Microsoft.Extensions.AI;

/// <summary>
/// Creates a spam detection agent.
/// </summary>
/// <returns>A ChatClientAgent configured for spam detection</returns>
private static ChatClientAgent GetSpamDetectionAgent(IChatClient chatClient) =>
    new(chatClient, new ChatClientAgentOptions
    {
        ChatOptions = new()
        {
            Instructions = "You are a spam detection assistant that identifies spam emails.",
            ResponseFormat = ChatResponseFormat.ForJsonSchema(AIJsonUtilities.CreateJsonSchema(typeof(DetectionResult)))
        }
    });

/// <summary>
/// Creates an email assistant agent.
/// </summary>
/// <returns>A ChatClientAgent configured for email assistance</returns>
private static ChatClientAgent GetEmailAssistantAgent(IChatClient chatClient) =>
    new(chatClient, new ChatClientAgentOptions
    {
        ChatOptions = new()
        {
            Instructions = "You are an email assistant that helps users draft professional responses to emails.",
            ResponseFormat = ChatResponseFormat.ForJsonSchema(AIJsonUtilities.CreateJsonSchema(typeof(EmailResponse)))
        }
    });

تنفيذ المنفذين

إنشاء منفذي سير العمل الذين يتعاملون مع مراحل مختلفة من معالجة البريد الإلكتروني:

using Microsoft.Agents.AI.Workflows;
using System.Text.Json;

/// <summary>
/// Executor that detects spam using an AI agent.
/// </summary>
internal sealed partial class SpamDetectionExecutor : Executor
{
    private readonly AIAgent _spamDetectionAgent;

    public SpamDetectionExecutor(AIAgent spamDetectionAgent) : base("SpamDetectionExecutor")
    {
        this._spamDetectionAgent = spamDetectionAgent;
    }

    [MessageHandler]
    private async ValueTask<DetectionResult> HandleAsync(ChatMessage message, IWorkflowContext context, CancellationToken cancellationToken = default)
    {
        // Generate a random email ID and store the email content to shared state
        var newEmail = new Email
        {
            EmailId = Guid.NewGuid().ToString("N"),
            EmailContent = message.Text
        };
        await context.QueueStateUpdateAsync(newEmail.EmailId, newEmail, scopeName: EmailStateConstants.EmailStateScope);

        // Invoke the agent for spam detection
        var response = await this._spamDetectionAgent.RunAsync(message);
        var detectionResult = JsonSerializer.Deserialize<DetectionResult>(response.Text);

        detectionResult!.EmailId = newEmail.EmailId;
        return detectionResult;
    }
}

/// <summary>
/// Executor that assists with email responses using an AI agent.
/// </summary>
internal sealed partial class EmailAssistantExecutor : Executor
{
    private readonly AIAgent _emailAssistantAgent;

    public EmailAssistantExecutor(AIAgent emailAssistantAgent) : base("EmailAssistantExecutor")
    {
        this._emailAssistantAgent = emailAssistantAgent;
    }

    [MessageHandler]
    private async ValueTask<EmailResponse> HandleAsync(DetectionResult message, IWorkflowContext context, CancellationToken cancellationToken = default)
    {
        if (message.IsSpam)
        {
            throw new ArgumentException("This executor should only handle non-spam messages.");
        }

        // Retrieve the email content from shared state
        var email = await context.ReadStateAsync<Email>(message.EmailId, scopeName: EmailStateConstants.EmailStateScope)
            ?? throw new InvalidOperationException("Email not found.");

        // Invoke the agent to draft a response
        var response = await this._emailAssistantAgent.RunAsync(email.EmailContent);
        var emailResponse = JsonSerializer.Deserialize<EmailResponse>(response.Text);

        return emailResponse!;
    }
}

/// <summary>
/// Executor that sends emails.
/// </summary>
internal sealed partial class SendEmailExecutor : Executor
{
    public SendEmailExecutor() : base("SendEmailExecutor") { }

    [MessageHandler]
    private async ValueTask HandleAsync(EmailResponse message, IWorkflowContext context, CancellationToken cancellationToken = default) =>
        await context.YieldOutputAsync($"Email sent: {message.Response}");
}

/// <summary>
/// Executor that handles spam messages.
/// </summary>
internal sealed partial class HandleSpamExecutor : Executor
{
    public HandleSpamExecutor() : base("HandleSpamExecutor") { }

    [MessageHandler]
    private async ValueTask HandleAsync(DetectionResult message, IWorkflowContext context, CancellationToken cancellationToken = default)
    {
        if (message.IsSpam)
        {
            await context.YieldOutputAsync($"Email marked as spam: {message.Reason}");
        }
        else
        {
            throw new ArgumentException("This executor should only handle spam messages.");
        }
    }
}

إنشاء سير العمل باستخدام الحواف الشرطية

الآن قم بإنشاء البرنامج الرئيسي الذي ينشئ سير العمل وينفذه:

using Microsoft.Extensions.AI;

public static class Program
{
    private static async Task Main()
    {
        // Set up the Azure OpenAI client
        var endpoint = Environment.GetEnvironmentVariable("AZURE_OPENAI_ENDPOINT")
            ?? throw new Exception("AZURE_OPENAI_ENDPOINT is not set.");
        var deploymentName = Environment.GetEnvironmentVariable("AZURE_OPENAI_DEPLOYMENT_NAME") ?? "gpt-4o-mini";
        var chatClient = new AIProjectClient(new Uri(endpoint), new DefaultAzureCredential())
            .GetProjectOpenAIClient().GetProjectResponsesClient().AsIChatClient(deploymentName);

        // Create agents
        AIAgent spamDetectionAgent = GetSpamDetectionAgent(chatClient);
        AIAgent emailAssistantAgent = GetEmailAssistantAgent(chatClient);

        // Create executors
        var spamDetectionExecutor = new SpamDetectionExecutor(spamDetectionAgent);
        var emailAssistantExecutor = new EmailAssistantExecutor(emailAssistantAgent);
        var sendEmailExecutor = new SendEmailExecutor();
        var handleSpamExecutor = new HandleSpamExecutor();

        // Build the workflow with conditional edges
        var workflow = new WorkflowBuilder(spamDetectionExecutor)
            // Non-spam path: route to email assistant when IsSpam = false
            .AddEdge(spamDetectionExecutor, emailAssistantExecutor, condition: GetCondition(expectedResult: false))
            .AddEdge(emailAssistantExecutor, sendEmailExecutor)
            // Spam path: route to spam handler when IsSpam = true
            .AddEdge(spamDetectionExecutor, handleSpamExecutor, condition: GetCondition(expectedResult: true))
            .WithOutputFrom(handleSpamExecutor, sendEmailExecutor)
            .Build();

        // Execute the workflow with sample spam email
        string emailContent = "Congratulations! You've won $1,000,000! Click here to claim your prize now!";
        StreamingRun run = await InProcessExecution.RunStreamingAsync(workflow, new ChatMessage(ChatRole.User, emailContent));
        await run.TrySendMessageAsync(new TurnToken(emitEvents: true));

        await foreach (WorkflowEvent evt in run.WatchStreamAsync().ConfigureAwait(false))
        {
            if (evt is WorkflowOutputEvent outputEvent)
            {
                Console.WriteLine($"{outputEvent}");
            }
        }
    }
}

تحذير

DefaultAzureCredential مناسب للتنمية ولكنه يتطلب دراسة متأنية في الإنتاج. في الإنتاج، ضع في اعتبارك استخدام بيانات اعتماد محددة (على سبيل المثال، ManagedIdentityCredential) لتجنب مشكلات زمن الانتقال، وبحث بيانات الاعتماد غير المقصودة، والمخاطر الأمنية المحتملة من الآليات الاحتياطية.

كيفية عملها

1. إدخال سير العمل: يبدأ spamDetectionExecutor سير العمل بتلقي ChatMessage.

2. تحليل البريد العشوائي: يقوم عامل الكشف عن البريد العشوائي بتحليل البريد الإلكتروني وإرجاع خصائص منظمة DetectionResult مع IsSpam و Reason .

3. التوجيه الشرطي: استنادا إلى IsSpam القيمة:

   • إذا كان البريد العشوائي (IsSpam = true): يوجه إلى HandleSpamExecutor استخدام GetCondition(true)
   • إذا كان شرعيا (IsSpam = false): يوجه إلى EmailAssistantExecutor استخدام GetCondition(false)

4. إنشاء الاستجابة: بالنسبة لرسائل البريد الإلكتروني المشروعة، يقوم مساعد البريد الإلكتروني بصياغة استجابة احترافية.

5. الإخراج النهائي: ينتج عن سير العمل إما إشعار بريد عشوائي أو يرسل استجابة البريد الإلكتروني التي تمت صياغتها.

الميزات الرئيسية للحواف الشرطية

1. شروطType-Safe: GetCondition ينشئ الأسلوب وظائف شرط قابلة لإعادة الاستخدام تقيم محتوى الرسالة بأمان.

2. مسارات متعددة: يمكن أن يكون للمنفذ الواحد حواف صادرة متعددة بشروط مختلفة، ما يتيح منطق التفريع المعقد.

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

4. معالجة الأخطاء: يتحقق المنفذون من صحة إدخالاتهم ويطرحون استثناءات ذات معنى عند تلقي أنواع رسائل غير متوقعة.

5. البنية النظيفة: يتحمل كل منفذ مسؤولية واحدة، ما يجعل سير العمل قابلا للصيانة والاختبار.

تشغيل المثال

عند تشغيل سير العمل هذا باستخدام نموذج البريد الإلكتروني العشوائي:

Email marked as spam: This email contains common spam indicators including monetary prizes, urgency tactics, and suspicious links that are typical of phishing attempts.

حاول تغيير محتوى البريد الإلكتروني إلى شيء شرعي:

string emailContent = "Hi, I wanted to follow up on our meeting yesterday and get your thoughts on the project proposal.";

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

يشكل نمط التوجيه الشرطي هذا الأساس لبناء مهام سير عمل متطورة يمكنها التعامل مع أشجار القرار المعقدة ومنطق الأعمال.

التنفيذ الكامل

لتنفيذ العمل الكامل، راجع هذا النموذج في مستودع إطار عمل العامل.

ما ستقوم بإنشاءه

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

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

المفاهيم التي تمت تغطيتها

• الحواف الشرطية

المتطلبات الأساسية

• Python 3.10 أو أحدث
• تم تثبيت إطار عمل العامل: pip install agent-framework-core
• Azure خدمة OpenAI التي تم تكوينها مع متغيرات البيئة المناسبة
• مصادقة Azure CLI:az login

الخطوة 1: استيراد التبعيات المطلوبة

ابدأ باستيراد المكونات الضرورية لسير العمل الشرطي:

import asyncio
import os
from dataclasses import dataclass
from typing import Any, Literal
from uuid import uuid4

from typing_extensions import Never

from agent_framework import (
    AgentExecutor,
    AgentExecutorRequest,
    AgentExecutorResponse,
    Message,
    WorkflowBuilder,
    WorkflowContext,
    executor,
    Case,
    Default,
)
import os
from agent_framework.openai import OpenAIChatCompletionClient
from azure.identity import AzureCliCredential
from pydantic import BaseModel

الخطوة 2: تعريف نماذج البيانات

إنشاء نماذج Pydantic لتبادل البيانات المنظمة بين مكونات سير العمل:

class DetectionResult(BaseModel):
    """Represents the result of spam detection."""
    # is_spam drives the routing decision taken by edge conditions
    is_spam: bool
    # Human readable rationale from the detector
    reason: str
    # The agent must include the original email so downstream agents can operate without reloading content
    email_content: str


class EmailResponse(BaseModel):
    """Represents the response from the email assistant."""
    # The drafted reply that a user could copy or send
    response: str

الخطوة 3: إنشاء وظائف الشرط

تحديد وظائف الشرط التي ستحدد قرارات التوجيه:

def get_condition(expected_result: bool):
    """Create a condition callable that routes based on DetectionResult.is_spam."""

    # The returned function will be used as an edge predicate.
    # It receives whatever the upstream executor produced.
    def condition(message: Any) -> bool:
        # Defensive guard. If a non AgentExecutorResponse appears, let the edge pass to avoid dead ends.
        if not isinstance(message, AgentExecutorResponse):
            return True

        try:
            # Prefer parsing a structured DetectionResult from the agent JSON text.
            # Using model_validate_json ensures type safety and raises if the shape is wrong.
            detection = DetectionResult.model_validate_json(message.agent_response.text)
            # Route only when the spam flag matches the expected path.
            return detection.is_spam == expected_result
        except Exception:
            # Fail closed on parse errors so we do not accidentally route to the wrong path.
            # Returning False prevents this edge from activating.
            return False

    return condition

الخطوة 4: إنشاء منفذي المعالج

تحديد المنفذين للتعامل مع نتائج التوجيه المختلفة:

@executor(id="send_email")
async def handle_email_response(response: AgentExecutorResponse, ctx: WorkflowContext[Never, str]) -> None:
    """Handle legitimate emails by drafting a professional response."""
    # Downstream of the email assistant. Parse a validated EmailResponse and yield the workflow output.
    email_response = EmailResponse.model_validate_json(response.agent_response.text)
    await ctx.yield_output(f"Email sent:\n{email_response.response}")


@executor(id="handle_spam")
async def handle_spam_classifier_response(response: AgentExecutorResponse, ctx: WorkflowContext[Never, str]) -> None:
    """Handle spam emails by marking them appropriately."""
    # Spam path. Confirm the DetectionResult and yield the workflow output. Guard against accidental non spam input.
    detection = DetectionResult.model_validate_json(response.agent_response.text)
    if detection.is_spam:
        await ctx.yield_output(f"Email marked as spam: {detection.reason}")
    else:
        # This indicates the routing predicate and executor contract are out of sync.
        raise RuntimeError("This executor should only handle spam messages.")


@executor(id="to_email_assistant_request")
async def to_email_assistant_request(
    response: AgentExecutorResponse, ctx: WorkflowContext[AgentExecutorRequest]
) -> None:
    """Transform spam detection response into a request for the email assistant."""
    # Parse the detection result and extract the email content for the assistant
    detection = DetectionResult.model_validate_json(response.agent_response.text)

    # Create a new request for the email assistant with the original email content
    request = AgentExecutorRequest(
        messages=[Message(role="user", contents=[detection.email_content])],
        should_respond=True
    )
    await ctx.send_message(request)

الخطوة 5: إنشاء وكلاء الذكاء الاصطناعي

إعداد Azure عوامل OpenAI مع تنسيق الإخراج المنظم:

async def main() -> None:
    # Create agents
    # AzureCliCredential uses your current az login. This avoids embedding secrets in code.
    chat_client = OpenAIChatCompletionClient(
        model=os.environ["AZURE_OPENAI_CHAT_COMPLETION_MODEL"],
        azure_endpoint=os.environ["AZURE_OPENAI_ENDPOINT"],
        api_version=os.getenv("AZURE_OPENAI_API_VERSION"),
        credential=AzureCliCredential(),
    )

    # Agent 1. Classifies spam and returns a DetectionResult object.
    # response_format enforces that the LLM returns parsable JSON for the Pydantic model.
    spam_detection_agent = AgentExecutor(
        chat_client.as_agent(
            instructions=(
                "You are a spam detection assistant that identifies spam emails. "
                "Always return JSON with fields is_spam (bool), reason (string), and email_content (string). "
                "Include the original email content in email_content."
            ),
            default_options={"response_format": DetectionResult},
        ),
        id="spam_detection_agent",
    )

    # Agent 2. Drafts a professional reply. Also uses structured JSON output for reliability.
    email_assistant_agent = AgentExecutor(
        chat_client.as_agent(
            instructions=(
                "You are an email assistant that helps users draft professional responses to emails. "
                "Your input might be a JSON object that includes 'email_content'; base your reply on that content. "
                "Return JSON with a single field 'response' containing the drafted reply."
            ),
            default_options={"response_format": EmailResponse},
        ),
        id="email_assistant_agent",
    )

الخطوة 6: إنشاء سير العمل الشرطي

إنشاء سير عمل مع الحواف الشرطية التي توجه استنادا إلى نتائج الكشف عن البريد العشوائي:

    # Build the workflow graph.
    # Start at the spam detector.
    # If not spam, hop to a transformer that creates a new AgentExecutorRequest,
    # then call the email assistant, then finalize.
    # If spam, go directly to the spam handler and finalize.
    workflow = (
        WorkflowBuilder(start_executor=spam_detection_agent)
        # Not spam path: transform response -> request for assistant -> assistant -> send email
        .add_edge(spam_detection_agent, to_email_assistant_request, condition=get_condition(False))
        .add_edge(to_email_assistant_request, email_assistant_agent)
        .add_edge(email_assistant_agent, handle_email_response)
        # Spam path: send to spam handler
        .add_edge(spam_detection_agent, handle_spam_classifier_response, condition=get_condition(True))
        .build()
    )

الخطوة 7: تنفيذ سير العمل

تشغيل سير العمل باستخدام نموذج محتوى البريد الإلكتروني:

    # Read Email content from the sample resource file.
    # This keeps the sample deterministic since the model sees the same email every run.
    email_path = os.path.join(os.path.dirname(os.path.dirname(os.path.realpath(__file__))), "resources", "email.txt")

    with open(email_path) as email_file:  # noqa: ASYNC230
        email = email_file.read()

    # Execute the workflow. Since the start is an AgentExecutor, pass an AgentExecutorRequest.
    # The workflow completes when it becomes idle (no more work to do).
    request = AgentExecutorRequest(messages=[Message(role="user", contents=[email])], should_respond=True)
    events = await workflow.run(request)
    outputs = events.get_outputs()
    if outputs:
        print(f"Workflow output: {outputs[0]}")


if __name__ == "__main__":
    asyncio.run(main())

كيفية عمل الحواف الشرطية

1. دالات الشرط: get_condition() تنشئ الدالة دالة تقييم تفحص محتوى الرسالة وترجع True أو False لتحديد ما إذا كان يجب اجتياز الحافة.

2. فحص الرسالة: يمكن للشروط فحص أي جانب من جوانب الرسالة، بما في ذلك البيانات المنظمة من استجابات العامل التي تم تحليلها مع نماذج Pydantic.

3. البرمجة الدفاعية: تتضمن وظيفة الشرط معالجة الأخطاء لمنع فشل التوجيه عند تحليل البيانات المنظمة.

4. التوجيه الديناميكي: استنادا إلى نتيجة الكشف عن البريد العشوائي، يتم توجيه رسائل البريد الإلكتروني تلقائيا إما إلى مساعد البريد الإلكتروني (لرسائل البريد الإلكتروني الشرعية) أو معالج البريد العشوائي (لرسائل البريد الإلكتروني المشبوهة).

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

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

التنفيذ الكامل

لتنفيذ العمل الكامل، راجع نموذج edge_condition.py في مستودع إطار عمل العامل.

إنشاء سير العمل باستخدام الحواف الشرطية

استخدم AddDirectEdge مع دالة شرط لتوجيه الرسائل استنادا إلى قيم وقت التشغيل:

wf, err := workflow.NewBuilder(spamDetector).
    AddDirectEdge(spamDetector, emailAssistant, false, func(msg any) bool {
        result, ok := msg.(DetectionResult)
        return ok && !result.IsSpam
    }).
    AddDirectEdge(spamDetector, spamHandler, false, func(msg any) bool {
        result, ok := msg.(DetectionResult)
        return ok && result.IsSpam
    }).
    WithOutputFrom(emailAssistant, spamHandler).
    Build()

رمز نموذج الحافة الشرطية

لتنفيذ العمل الكامل، راجع نموذج شرط الحافة في مستودع Agent Framework Go.

Switch-Case الحواف

البناء على الحواف الشرطية

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

ما ستبنيه باستخدام Switch-Case

ستقوم بتوسيع سير عمل معالجة البريد الإلكتروني للتعامل مع ثلاثة مسارات قرار:

• → إرسال بريد إلكتروني إلى NotSpam → Email Assistant
• → التعامل مع "منفذ البريد العشوائي"
• غير مؤكد → التعامل مع المنفذ غير المؤكد (حالة افتراضية)

يتمثل التحسين الرئيسي في SwitchBuilder استخدام النمط بدلا من حواف شرطية فردية متعددة، ما يجعل سير العمل أسهل في الفهم والصيانة مع تزايد تعقيد القرار.

المفاهيم التي تمت تغطيتها

• Switch-Case الحواف

نماذج البيانات Switch-Case

تحديث نماذج البيانات لدعم التصنيف ثلاثي الاتجاه:

/// <summary>
/// Represents the possible decisions for spam detection.
/// </summary>
public enum SpamDecision
{
    NotSpam,
    Spam,
    Uncertain
}

/// <summary>
/// Represents the result of spam detection with enhanced decision support.
/// </summary>
public sealed class DetectionResult
{
    [JsonPropertyName("spam_decision")]
    [JsonConverter(typeof(JsonStringEnumConverter))]
    public SpamDecision spamDecision { get; set; }

    [JsonPropertyName("reason")]
    public string Reason { get; set; } = string.Empty;

    // Email ID is generated by the executor, not the agent
    [JsonIgnore]
    public string EmailId { get; set; } = string.Empty;
}

/// <summary>
/// Represents an email stored in shared state.
/// </summary>
internal sealed class Email
{
    [JsonPropertyName("email_id")]
    public string EmailId { get; set; } = string.Empty;

    [JsonPropertyName("email_content")]
    public string EmailContent { get; set; } = string.Empty;
}

/// <summary>
/// Represents the response from the email assistant.
/// </summary>
public sealed class EmailResponse
{
    [JsonPropertyName("response")]
    public string Response { get; set; } = string.Empty;
}

/// <summary>
/// Constants for shared state scopes.
/// </summary>
internal static class EmailStateConstants
{
    public const string EmailStateScope = "EmailState";
}

Condition Factory for Switch-Case

إنشاء مصنع حالة قابل لإعادة الاستخدام يقوم بإنشاء دالات تقييم لكل قرار بريد عشوائي:

/// <summary>
/// Creates a condition for routing messages based on the expected spam detection result.
/// </summary>
/// <param name="expectedDecision">The expected spam detection decision</param>
/// <returns>A function that evaluates whether a message meets the expected result</returns>
private static Func<object?, bool> GetCondition(SpamDecision expectedDecision) =>
    detectionResult => detectionResult is DetectionResult result && result.spamDecision == expectedDecision;

نهج المصنع هذا:

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

عامل الذكاء الاصطناعي المحسن

تحديث عامل الكشف عن البريد العشوائي ليكون أقل ثقة وإرجاع التصنيفات ثلاثية الاتجاه:

/// <summary>
/// Creates a spam detection agent with enhanced uncertainty handling.
/// </summary>
/// <returns>A ChatClientAgent configured for three-way spam detection</returns>
private static ChatClientAgent GetSpamDetectionAgent(IChatClient chatClient) =>
    new(chatClient, new ChatClientAgentOptions
    {
        ChatOptions = new()
        {
            Instructions = "You are a spam detection assistant that identifies spam emails. Be less confident in your assessments.",
            ResponseFormat = ChatResponseFormat.ForJsonSchema<DetectionResult>()
        }
    });

/// <summary>
/// Creates an email assistant agent (unchanged from conditional edges example).
/// </summary>
/// <returns>A ChatClientAgent configured for email assistance</returns>
private static ChatClientAgent GetEmailAssistantAgent(IChatClient chatClient) =>
    new(chatClient, new ChatClientAgentOptions
    {
        ChatOptions = new()
        {
            Instructions = "You are an email assistant that helps users draft responses to emails with professionalism.",
            ResponseFormat = ChatResponseFormat.ForJsonSchema<EmailResponse>()
        }
    });

منفذو سير العمل مع التوجيه المحسن

تنفيذ المنفذين الذين يتعاملون مع التوجيه ثلاثي الاتجاه مع إدارة الحالة المشتركة:

/// <summary>
/// Executor that detects spam using an AI agent with three-way classification.
/// </summary>
internal sealed partial class SpamDetectionExecutor : Executor
{
    private readonly AIAgent _spamDetectionAgent;

    public SpamDetectionExecutor(AIAgent spamDetectionAgent) : base("SpamDetectionExecutor")
    {
        this._spamDetectionAgent = spamDetectionAgent;
    }

    [MessageHandler]
    private async ValueTask<DetectionResult> HandleAsync(ChatMessage message, IWorkflowContext context, CancellationToken cancellationToken = default)
    {
        // Generate a random email ID and store the email content in shared state
        var newEmail = new Email
        {
            EmailId = Guid.NewGuid().ToString("N"),
            EmailContent = message.Text
        };
        await context.QueueStateUpdateAsync(newEmail.EmailId, newEmail, scopeName: EmailStateConstants.EmailStateScope);

        // Invoke the agent for enhanced spam detection
        var response = await this._spamDetectionAgent.RunAsync(message);
        var detectionResult = JsonSerializer.Deserialize<DetectionResult>(response.Text);

        detectionResult!.EmailId = newEmail.EmailId;
        return detectionResult;
    }
}

/// <summary>
/// Executor that assists with email responses using an AI agent.
/// </summary>
internal sealed partial class EmailAssistantExecutor : Executor
{
    private readonly AIAgent _emailAssistantAgent;

    public EmailAssistantExecutor(AIAgent emailAssistantAgent) : base("EmailAssistantExecutor")
    {
        this._emailAssistantAgent = emailAssistantAgent;
    }

    [MessageHandler]
    private async ValueTask<EmailResponse> HandleAsync(DetectionResult message, IWorkflowContext context, CancellationToken cancellationToken = default)
    {
        if (message.spamDecision == SpamDecision.Spam)
        {
            throw new ArgumentException("This executor should only handle non-spam messages.");
        }

        // Retrieve the email content from shared state
        var email = await context.ReadStateAsync<Email>(message.EmailId, scopeName: EmailStateConstants.EmailStateScope);

        // Invoke the agent to draft a response
        var response = await this._emailAssistantAgent.RunAsync(email!.EmailContent);
        var emailResponse = JsonSerializer.Deserialize<EmailResponse>(response.Text);

        return emailResponse!;
    }
}

/// <summary>
/// Executor that sends emails.
/// </summary>
internal sealed partial class SendEmailExecutor : Executor
{
    public SendEmailExecutor() : base("SendEmailExecutor") { }

    [MessageHandler]
    private async ValueTask HandleAsync(EmailResponse message, IWorkflowContext context, CancellationToken cancellationToken = default) =>
        await context.YieldOutputAsync($"Email sent: {message.Response}").ConfigureAwait(false);
}

/// <summary>
/// Executor that handles spam messages.
/// </summary>
internal sealed partial class HandleSpamExecutor : Executor
{
    public HandleSpamExecutor() : base("HandleSpamExecutor") { }

    [MessageHandler]
    private async ValueTask HandleAsync(DetectionResult message, IWorkflowContext context, CancellationToken cancellationToken = default)
    {
        if (message.spamDecision == SpamDecision.Spam)
        {
            await context.YieldOutputAsync($"Email marked as spam: {message.Reason}").ConfigureAwait(false);
        }
        else
        {
            throw new ArgumentException("This executor should only handle spam messages.");
        }
    }
}

/// <summary>
/// Executor that handles uncertain emails requiring manual review.
/// </summary>
internal sealed partial class HandleUncertainExecutor : Executor
{
    public HandleUncertainExecutor() : base("HandleUncertainExecutor") { }

    [MessageHandler]
    private async ValueTask HandleAsync(DetectionResult message, IWorkflowContext context, CancellationToken cancellationToken = default)
    {
        if (message.spamDecision == SpamDecision.Uncertain)
        {
            var email = await context.ReadStateAsync<Email>(message.EmailId, scopeName: EmailStateConstants.EmailStateScope);
            await context.YieldOutputAsync($"Email marked as uncertain: {message.Reason}. Email content: {email?.EmailContent}");
        }
        else
        {
            throw new ArgumentException("This executor should only handle uncertain spam decisions.");
        }
    }
}

إنشاء سير عمل باستخدام نمط Switch-Case

استبدل حواف شرطية متعددة بنمط حالة التبديل الأنظف:

public static class Program
{
    private static async Task Main()
    {
        // Set up the Azure OpenAI client
        var endpoint = Environment.GetEnvironmentVariable("AZURE_OPENAI_ENDPOINT") ?? throw new Exception("AZURE_OPENAI_ENDPOINT is not set.");
        var deploymentName = Environment.GetEnvironmentVariable("AZURE_OPENAI_DEPLOYMENT_NAME") ?? "gpt-4o-mini";
        var chatClient = new AIProjectClient(new Uri(endpoint), new DefaultAzureCredential())
            .GetProjectOpenAIClient()
            .GetProjectResponsesClient()
            .AsIChatClient(deploymentName);

        // Create agents
        AIAgent spamDetectionAgent = GetSpamDetectionAgent(chatClient);
        AIAgent emailAssistantAgent = GetEmailAssistantAgent(chatClient);

        // Create executors
        var spamDetectionExecutor = new SpamDetectionExecutor(spamDetectionAgent);
        var emailAssistantExecutor = new EmailAssistantExecutor(emailAssistantAgent);
        var sendEmailExecutor = new SendEmailExecutor();
        var handleSpamExecutor = new HandleSpamExecutor();
        var handleUncertainExecutor = new HandleUncertainExecutor();

        // Build the workflow using switch-case for cleaner three-way routing
        WorkflowBuilder builder = new(spamDetectionExecutor);
        builder.AddSwitch(spamDetectionExecutor, switchBuilder =>
            switchBuilder
            .AddCase(
                GetCondition(expectedDecision: SpamDecision.NotSpam),
                emailAssistantExecutor
            )
            .AddCase(
                GetCondition(expectedDecision: SpamDecision.Spam),
                handleSpamExecutor
            )
            .WithDefault(
                handleUncertainExecutor
            )
        )
        // After the email assistant writes a response, it will be sent to the send email executor
        .AddEdge(emailAssistantExecutor, sendEmailExecutor)
        .WithOutputFrom(handleSpamExecutor, sendEmailExecutor, handleUncertainExecutor);

        var workflow = builder.Build();

        // Read an email from a text file (use ambiguous content for demonstration)
        string email = Resources.Read("ambiguous_email.txt");

        // Execute the workflow
        StreamingRun run = await InProcessExecution.RunStreamingAsync(workflow, new ChatMessage(ChatRole.User, email));
        await run.TrySendMessageAsync(new TurnToken(emitEvents: true));
        await foreach (WorkflowEvent evt in run.WatchStreamAsync().ConfigureAwait(false))
        {
            if (evt is WorkflowOutputEvent outputEvent)
            {
                Console.WriteLine($"{outputEvent}");
            }
        }
    }
}

مزايا Switch-Case

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

مقارنة الأنماط

قبل (الحواف الشرطية):

var workflow = new WorkflowBuilder(spamDetectionExecutor)
    .AddEdge(spamDetectionExecutor, emailAssistantExecutor, condition: GetCondition(expectedResult: false))
    .AddEdge(spamDetectionExecutor, handleSpamExecutor, condition: GetCondition(expectedResult: true))
    // No clean way to handle a third case
    .WithOutputFrom(handleSpamExecutor, sendEmailExecutor)
    .Build();

بعد (Switch-Case):

WorkflowBuilder builder = new(spamDetectionExecutor);
builder.AddSwitch(spamDetectionExecutor, switchBuilder =>
    switchBuilder
    .AddCase(GetCondition(SpamDecision.NotSpam), emailAssistantExecutor)
    .AddCase(GetCondition(SpamDecision.Spam), handleSpamExecutor)
    .WithDefault(handleUncertainExecutor)  // Clean default case
)
// Continue building the rest of the workflow

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

تشغيل المثال

عند تشغيل سير العمل هذا مع محتوى بريد إلكتروني غامض:

Email marked as uncertain: This email contains promotional language but might be from a legitimate business contact, requiring human review for proper classification.

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

التنفيذ الكامل

لتنفيذ العمل الكامل، راجع هذا النموذج في مستودع إطار عمل العامل.

البناء على الحواف الشرطية

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

ما الذي ستقوم بإنشاءه بعد ذلك

ستقوم بتوسيع سير عمل معالجة البريد الإلكتروني للتعامل مع ثلاثة مسارات قرار:

• → إرسال بريد إلكتروني إلى NotSpam → Email Assistant
• → وضع علامة بريد عشوائي على أنه بريد عشوائي
• علامة → غير مؤكد للمراجعة اليدوية (حالة افتراضية)

يستخدم التحسين الرئيسي مجموعة حافة واحدة لحالة التبديل بدلا من حواف شرطية فردية متعددة، ما يجعل سير العمل أسهل في الفهم والحفاظ عليه مع تزايد تعقيد القرار.

المفاهيم التي تمت تغطيتها

• Switch-Case الحواف

نماذج البيانات المحسنة

تحديث نماذج البيانات لدعم التصنيف ثلاثي الاتجاه:

from typing import Literal

class DetectionResultAgent(BaseModel):
    """Structured output returned by the spam detection agent."""

    # The agent classifies the email into one of three categories
    spam_decision: Literal["NotSpam", "Spam", "Uncertain"]
    reason: str

class EmailResponse(BaseModel):
    """Structured output returned by the email assistant agent."""

    response: str

@dataclass
class DetectionResult:
    """Internal typed payload used for routing and downstream handling."""

    spam_decision: str
    reason: str
    email_id: str

@dataclass
class Email:
    """In memory record of the email content stored in shared state."""

    email_id: str
    email_content: str

Switch-Case Condition Factory

إنشاء مصنع حالة قابل لإعادة الاستخدام يقوم بإنشاء دالات تقييم لكل قرار بريد عشوائي:

def get_case(expected_decision: str):
    """Factory that returns a predicate matching a specific spam_decision value."""

    def condition(message: Any) -> bool:
        # Only match when the upstream payload is a DetectionResult with the expected decision
        return isinstance(message, DetectionResult) and message.spam_decision == expected_decision

    return condition

نهج المصنع هذا:

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

منفذو سير العمل مع الحالة المشتركة

تنفيذ المنفذين الذين يستخدمون الحالة المشتركة لتجنب تمرير محتوى بريد إلكتروني كبير من خلال كل خطوة سير عمل:

EMAIL_STATE_PREFIX = "email:"
CURRENT_EMAIL_ID_KEY = "current_email_id"

@executor(id="store_email")
async def store_email(email_text: str, ctx: WorkflowContext[AgentExecutorRequest]) -> None:
    """Store email content once and pass around a lightweight ID reference."""

    # Persist the raw email content in shared state
    new_email = Email(email_id=str(uuid4()), email_content=email_text)
    ctx.set_state(f"{EMAIL_STATE_PREFIX}{new_email.email_id}", new_email)
    ctx.set_state(CURRENT_EMAIL_ID_KEY, new_email.email_id)

    # Forward email to spam detection agent
    await ctx.send_message(
        AgentExecutorRequest(messages=[Message(role="user", contents=[new_email.email_content])], should_respond=True)
    )

@executor(id="to_detection_result")
async def to_detection_result(response: AgentExecutorResponse, ctx: WorkflowContext[DetectionResult]) -> None:
    """Transform agent response into a typed DetectionResult with email ID."""

    # Parse the agent's structured JSON output
    parsed = DetectionResultAgent.model_validate_json(response.agent_response.text)
    email_id: str = ctx.get_state(CURRENT_EMAIL_ID_KEY)

    # Create typed message for switch-case routing
    await ctx.send_message(DetectionResult(
        spam_decision=parsed.spam_decision,
        reason=parsed.reason,
        email_id=email_id
    ))

@executor(id="submit_to_email_assistant")
async def submit_to_email_assistant(detection: DetectionResult, ctx: WorkflowContext[AgentExecutorRequest]) -> None:
    """Handle NotSpam emails by forwarding to the email assistant."""

    # Guard against misrouting
    if detection.spam_decision != "NotSpam":
        raise RuntimeError("This executor should only handle NotSpam messages.")

    # Retrieve original email content from shared state
    email: Email = ctx.get_state(f"{EMAIL_STATE_PREFIX}{detection.email_id}")
    await ctx.send_message(
        AgentExecutorRequest(messages=[Message(role="user", contents=[email.email_content])], should_respond=True)
    )

@executor(id="finalize_and_send")
async def finalize_and_send(response: AgentExecutorResponse, ctx: WorkflowContext[Never, str]) -> None:
    """Parse email assistant response and yield final output."""

    parsed = EmailResponse.model_validate_json(response.agent_response.text)
    await ctx.yield_output(f"Email sent: {parsed.response}")

@executor(id="handle_spam")
async def handle_spam(detection: DetectionResult, ctx: WorkflowContext[Never, str]) -> None:
    """Handle confirmed spam emails."""

    if detection.spam_decision == "Spam":
        await ctx.yield_output(f"Email marked as spam: {detection.reason}")
    else:
        raise RuntimeError("This executor should only handle Spam messages.")

@executor(id="handle_uncertain")
async def handle_uncertain(detection: DetectionResult, ctx: WorkflowContext[Never, str]) -> None:
    """Handle uncertain classifications that need manual review."""

    if detection.spam_decision == "Uncertain":
        # Include original content for human review
        email: Email | None = ctx.get_state(f"{EMAIL_STATE_PREFIX}{detection.email_id}")
        await ctx.yield_output(
            f"Email marked as uncertain: {detection.reason}. Email content: {getattr(email, 'email_content', '')}"
        )
    else:
        raise RuntimeError("This executor should only handle Uncertain messages.")

إنشاء عامل ذكاء الاصطناعي المحسن

تحديث عامل الكشف عن البريد العشوائي ليكون أقل ثقة وإرجاع التصنيفات ثلاثية الاتجاه:

async def main():
    chat_client = OpenAIChatCompletionClient(
        model=os.environ["AZURE_OPENAI_CHAT_COMPLETION_MODEL"],
        azure_endpoint=os.environ["AZURE_OPENAI_ENDPOINT"],
        api_version=os.getenv("AZURE_OPENAI_API_VERSION"),
        credential=AzureCliCredential(),
    )

    # Enhanced spam detection agent with three-way classification
    spam_detection_agent = AgentExecutor(
        chat_client.as_agent(
            instructions=(
                "You are a spam detection assistant that identifies spam emails. "
                "Be less confident in your assessments. "
                "Always return JSON with fields 'spam_decision' (one of NotSpam, Spam, Uncertain) "
                "and 'reason' (string)."
            ),
            default_options={"response_format": DetectionResultAgent},
        ),
        id="spam_detection_agent",
    )

    # Email assistant remains the same
    email_assistant_agent = AgentExecutor(
        chat_client.as_agent(
            instructions=(
                "You are an email assistant that helps users draft responses to emails with professionalism."
            ),
            default_options={"response_format": EmailResponse},
        ),
        id="email_assistant_agent",
    )

إنشاء سير عمل باستخدام مجموعة Switch-Case Edge

استبدل حواف شرطية متعددة بمجموعة واحدة لحالة التبديل:

    # Build workflow using switch-case for cleaner three-way routing
    workflow = (
        WorkflowBuilder(start_executor=store_email)
        .add_edge(store_email, spam_detection_agent)
        .add_edge(spam_detection_agent, to_detection_result)
        .add_switch_case_edge_group(
            to_detection_result,
            [
                # Explicit cases for specific decisions
                Case(condition=get_case("NotSpam"), target=submit_to_email_assistant),
                Case(condition=get_case("Spam"), target=handle_spam),
                # Default case handles messages when every predicate returns False
                Default(target=handle_uncertain),
            ],
        )
        .add_edge(submit_to_email_assistant, email_assistant_agent)
        .add_edge(email_assistant_agent, finalize_and_send)
        .build()
    )

يتم تحديد الهدف الافتراضي فقط عند إرجاع Falseدالة تقييم كل حالة . إذا أثارت دالة التقييم استثناء، يفشل سير العمل؛ لا يتم توجيه الرسالة إلى الهدف الافتراضي.

التنفيذ والاختبار

تشغيل سير العمل مع محتوى بريد إلكتروني غامض يوضح التوجيه ثلاثي الاتجاه:

    # Use ambiguous email content that might trigger uncertain classification
    email = (
        "Hey there, I noticed you might be interested in our latest offer—no pressure, but it expires soon. "
        "Let me know if you'd like more details."
    )

    # Execute and display results
    events = await workflow.run(email)
    outputs = events.get_outputs()
    if outputs:
        for output in outputs:
            print(f"Workflow output: {output}")

المزايا الرئيسية Switch-Case Edges

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

المقارنة: شرطي مقابل Switch-Case

قبل (الحواف الشرطية):

.add_edge(detector, handler_a, condition=lambda x: x.result == "A")
.add_edge(detector, handler_b, condition=lambda x: x.result == "B")
.add_edge(detector, handler_c, condition=lambda x: x.result == "C")

بعد (Switch-Case):

.add_switch_case_edge_group(
    detector,
    [
        Case(condition=lambda x: x.result == "A", target=handler_a),
        Case(condition=lambda x: x.result == "B", target=handler_b),
        Default(target=handler_c),  # Handles values when every predicate returns False
    ],
)

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

نموذج التعليمات البرمجية Switch-Case

لتنفيذ العمل الكامل، راجع نموذج switch_case_edge_group.py في مستودع إطار عمل العامل.

إنشاء سير عمل باستخدام نمط Switch-Case

يستخدم AddSwitch لتجميع الحالات التي تم ترتيبها وهدف افتراضي اختياري:

builder := workflow.NewBuilder(spamDetector)
builder.AddSwitch(spamDetector).
    AddCase(func(msg any) bool {
        result, ok := msg.(DetectionResult)
        return ok && result.Decision == NotSpam
    }, emailAssistant).
    AddCase(func(msg any) bool {
        result, ok := msg.(DetectionResult)
        return ok && result.Decision == Spam
    }, spamHandler).
    WithDefault(manualReview).
    AddToBuilder(builder).
    AddEdge(emailAssistant, sendEmail).
    WithOutputFrom(sendEmail, spamHandler, manualReview)

wf, err := builder.Build()

نموذج التعليمات البرمجية Switch-Case

لتنفيذ العمل الكامل، راجع نموذج حالة التبديل في مستودع Agent Framework Go.

حواف متعددة التحديد

ما وراء حالة التبديل: توجيه متعدد التحديد

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

سير عمل معالجة البريد الإلكتروني المتقدم

بناء على مثال حالة التبديل، ستقوم بإنشاء نظام معالجة بريد إلكتروني محسن يوضح منطق التوجيه المتطور:

• رسائل البريد الإلكتروني العشوائي → معالج بريد عشوائي واحد (مثل حالة التبديل)
• رسائل البريد الإلكتروني الشرعية → تشغيل مساعد البريد الإلكتروني دائما + ملخص المشغل الشرطي لرسائل البريد الإلكتروني الطويلة
• رسائل البريد الإلكتروني غير المؤكدة → معالج واحد غير مؤكد (مثل switch-case)
• → استمرارية قاعدة البيانات مشغلة لكل من رسائل البريد الإلكتروني القصيرة ورسائل البريد الإلكتروني الطويلة الملخصة

يتيح هذا النمط مسارات المعالجة المتوازية التي تتكيف مع خصائص المحتوى.

المفاهيم التي تمت تغطيتها

• حواف ذات توزيعات مروحة

نماذج البيانات للتحديد المتعدد

توسيع نماذج البيانات لدعم تحليل طول البريد الإلكتروني وتلخيصه:

/// <summary>
/// Represents the result of enhanced email analysis with additional metadata.
/// </summary>
public sealed class AnalysisResult
{
    [JsonPropertyName("spam_decision")]
    [JsonConverter(typeof(JsonStringEnumConverter))]
    public SpamDecision spamDecision { get; set; }

    [JsonPropertyName("reason")]
    public string Reason { get; set; } = string.Empty;

    // Additional properties for sophisticated routing
    [JsonIgnore]
    public int EmailLength { get; set; }

    [JsonIgnore]
    public string EmailSummary { get; set; } = string.Empty;

    [JsonIgnore]
    public string EmailId { get; set; } = string.Empty;
}

/// <summary>
/// Represents the response from the email assistant.
/// </summary>
public sealed class EmailResponse
{
    [JsonPropertyName("response")]
    public string Response { get; set; } = string.Empty;
}

/// <summary>
/// Represents the response from the email summary agent.
/// </summary>
public sealed class EmailSummary
{
    [JsonPropertyName("summary")]
    public string Summary { get; set; } = string.Empty;
}

/// <summary>
/// A custom workflow event for database operations.
/// </summary>
internal sealed class DatabaseEvent(string message) : WorkflowEvent(message) { }

/// <summary>
/// Constants for email processing thresholds.
/// </summary>
public static class EmailProcessingConstants
{
    public const int LongEmailThreshold = 100;
}

دالة المعين الهدف: قلب التحديد المتعدد

تحدد دالة المعين الهدف المنفذين الذين يجب أن يتلقوا كل رسالة:

/// <summary>
/// Creates a target assigner for routing messages based on the analysis result.
/// </summary>
/// <returns>A function that takes an analysis result and returns the target partitions.</returns>
private static Func<AnalysisResult?, int, IEnumerable<int>> GetTargetAssigner()
{
    return (analysisResult, targetCount) =>
    {
        if (analysisResult is not null)
        {
            if (analysisResult.spamDecision == SpamDecision.Spam)
            {
                return [0]; // Route only to spam handler (index 0)
            }
            else if (analysisResult.spamDecision == SpamDecision.NotSpam)
            {
                // Always route to email assistant (index 1)
                List<int> targets = [1];

                // Conditionally add summarizer for long emails (index 2)
                if (analysisResult.EmailLength > EmailProcessingConstants.LongEmailThreshold)
                {
                    targets.Add(2);
                }

                return targets;
            }
            else // Uncertain
            {
                return [3]; // Route only to uncertain handler (index 3)
            }
        }
        throw new ArgumentException("Invalid analysis result.");
    };
}

الميزات الرئيسية لدالة تعيين الهدف

1. تحديد الهدف الديناميكي: إرجاع قائمة بمؤشرات المنفذ لتنشيطها
2. التوجيه المدرك للمحتوى: اتخاذ القرارات استنادا إلى خصائص الرسالة مثل طول البريد الإلكتروني
3. المعالجة المتوازية: يمكن تنفيذ أهداف متعددة في وقت واحد
4. المنطق الشرطي: التفريع المعقد استنادا إلى معايير متعددة

منفذو سير العمل المحسنون

تنفيذ المنفذين الذين يتعاملون مع التحليل والتوجيه المتقدمين:

/// <summary>
/// Executor that analyzes emails using an AI agent with enhanced analysis.
/// </summary>
internal sealed partial class EmailAnalysisExecutor : Executor
{
    private readonly AIAgent _emailAnalysisAgent;

    public EmailAnalysisExecutor(AIAgent emailAnalysisAgent) : base("EmailAnalysisExecutor")
    {
        this._emailAnalysisAgent = emailAnalysisAgent;
    }

    [MessageHandler]
    private async ValueTask<AnalysisResult> HandleAsync(ChatMessage message, IWorkflowContext context, CancellationToken cancellationToken = default)
    {
        // Generate a random email ID and store the email content
        var newEmail = new Email
        {
            EmailId = Guid.NewGuid().ToString("N"),
            EmailContent = message.Text
        };
        await context.QueueStateUpdateAsync(newEmail.EmailId, newEmail, scopeName: EmailStateConstants.EmailStateScope);

        // Invoke the agent for enhanced analysis
        var response = await this._emailAnalysisAgent.RunAsync(message);
        var analysisResult = JsonSerializer.Deserialize<AnalysisResult>(response.Text);

        // Enrich with metadata for routing decisions
        analysisResult!.EmailId = newEmail.EmailId;
        analysisResult.EmailLength = newEmail.EmailContent.Length;

        return analysisResult;
    }
}

/// <summary>
/// Executor that assists with email responses using an AI agent.
/// </summary>
internal sealed partial class EmailAssistantExecutor : Executor
{
    private readonly AIAgent _emailAssistantAgent;

    public EmailAssistantExecutor(AIAgent emailAssistantAgent) : base("EmailAssistantExecutor")
    {
        this._emailAssistantAgent = emailAssistantAgent;
    }

    [MessageHandler]
    private async ValueTask<EmailResponse> HandleAsync(AnalysisResult message, IWorkflowContext context, CancellationToken cancellationToken = default)
    {
        if (message.spamDecision == SpamDecision.Spam)
        {
            throw new ArgumentException("This executor should only handle non-spam messages.");
        }

        // Retrieve the email content from shared state
        var email = await context.ReadStateAsync<Email>(message.EmailId, scopeName: EmailStateConstants.EmailStateScope);

        // Invoke the agent to draft a response
        var response = await this._emailAssistantAgent.RunAsync(email!.EmailContent);
        var emailResponse = JsonSerializer.Deserialize<EmailResponse>(response.Text);

        return emailResponse!;
    }
}

/// <summary>
/// Executor that summarizes emails using an AI agent for long emails.
/// </summary>
internal sealed partial class EmailSummaryExecutor : Executor
{
    private readonly AIAgent _emailSummaryAgent;

    public EmailSummaryExecutor(AIAgent emailSummaryAgent) : base("EmailSummaryExecutor")
    {
        this._emailSummaryAgent = emailSummaryAgent;
    }

    [MessageHandler]
    private async ValueTask<AnalysisResult> HandleAsync(AnalysisResult message, IWorkflowContext context, CancellationToken cancellationToken = default)
    {
        // Read the email content from shared state
        var email = await context.ReadStateAsync<Email>(message.EmailId, scopeName: EmailStateConstants.EmailStateScope);

        // Generate summary for long emails
        var response = await this._emailSummaryAgent.RunAsync(email!.EmailContent);
        var emailSummary = JsonSerializer.Deserialize<EmailSummary>(response.Text);

        // Enrich the analysis result with the summary
        message.EmailSummary = emailSummary!.Summary;

        return message;
    }
}

/// <summary>
/// Executor that sends emails.
/// </summary>
internal sealed partial class SendEmailExecutor : Executor
{
    public SendEmailExecutor() : base("SendEmailExecutor") { }

    [MessageHandler]
    private async ValueTask HandleAsync(EmailResponse message, IWorkflowContext context, CancellationToken cancellationToken = default) =>
        await context.YieldOutputAsync($"Email sent: {message.Response}");
}

/// <summary>
/// Executor that handles spam messages.
/// </summary>
internal sealed partial class HandleSpamExecutor : Executor
{
    public HandleSpamExecutor() : base("HandleSpamExecutor") { }

    [MessageHandler]
    private async ValueTask HandleAsync(AnalysisResult message, IWorkflowContext context, CancellationToken cancellationToken = default)
    {
        if (message.spamDecision == SpamDecision.Spam)
        {
            await context.YieldOutputAsync($"Email marked as spam: {message.Reason}");
        }
        else
        {
            throw new ArgumentException("This executor should only handle spam messages.");
        }
    }
}

/// <summary>
/// Executor that handles uncertain messages requiring manual review.
/// </summary>
internal sealed partial class HandleUncertainExecutor : Executor
{
    public HandleUncertainExecutor() : base("HandleUncertainExecutor") { }

    [MessageHandler]
    private async ValueTask HandleAsync(AnalysisResult message, IWorkflowContext context, CancellationToken cancellationToken = default)
    {
        if (message.spamDecision == SpamDecision.Uncertain)
        {
            var email = await context.ReadStateAsync<Email>(message.EmailId, scopeName: EmailStateConstants.EmailStateScope);
            await context.YieldOutputAsync($"Email marked as uncertain: {message.Reason}. Email content: {email?.EmailContent}");
        }
        else
        {
            throw new ArgumentException("This executor should only handle uncertain spam decisions.");
        }
    }
}

/// <summary>
/// Executor that handles database access with custom events.
/// </summary>
internal sealed partial class DatabaseAccessExecutor : Executor
{
    public DatabaseAccessExecutor() : base("DatabaseAccessExecutor") { }

    [MessageHandler]
    private async ValueTask HandleAsync(AnalysisResult message, IWorkflowContext context, CancellationToken cancellationToken = default)
    {
        // Simulate database operations
        await context.ReadStateAsync<Email>(message.EmailId, scopeName: EmailStateConstants.EmailStateScope);
        await Task.Delay(100); // Simulate database access delay

        // Emit custom database event for monitoring
        await context.AddEventAsync(new DatabaseEvent($"Email {message.EmailId} saved to database."));
    }
}

عوامل الذكاء الاصطناعي المحسنة

إنشاء عوامل للتحليل والمساعدة والتلخيص:

/// <summary>
/// Create an enhanced email analysis agent.
/// </summary>
/// <returns>A ChatClientAgent configured for comprehensive email analysis</returns>
private static ChatClientAgent GetEmailAnalysisAgent(IChatClient chatClient) =>
    new(chatClient, new ChatClientAgentOptions
    {
        ChatOptions = new()
        {
            Instructions = "You are a spam detection assistant that identifies spam emails.",
            ResponseFormat = ChatResponseFormat.ForJsonSchema<AnalysisResult>()
        }
    });

/// <summary>
/// Creates an email assistant agent.
/// </summary>
/// <returns>A ChatClientAgent configured for email assistance</returns>
private static ChatClientAgent GetEmailAssistantAgent(IChatClient chatClient) =>
    new(chatClient, new ChatClientAgentOptions
    {
        ChatOptions = new()
        {
            Instructions = "You are an email assistant that helps users draft responses to emails with professionalism.",
            ResponseFormat = ChatResponseFormat.ForJsonSchema<EmailResponse>()
        }
    });

/// <summary>
/// Creates an agent that summarizes emails.
/// </summary>
/// <returns>A ChatClientAgent configured for email summarization</returns>
private static ChatClientAgent GetEmailSummaryAgent(IChatClient chatClient) =>
    new(chatClient, new ChatClientAgentOptions
    {
        ChatOptions = new()
        {
            Instructions = "You are an assistant that helps users summarize emails.",
            ResponseFormat = ChatResponseFormat.ForJsonSchema<EmailSummary>()
        }
    });

بناء سير عمل متعدد التحديد

إنشاء سير العمل مع التوجيه المتطور والمعالجة المتوازية:

public static class Program
{
    private static async Task Main()
    {
        // Set up the Azure OpenAI client
        var endpoint = Environment.GetEnvironmentVariable("AZURE_OPENAI_ENDPOINT") ?? throw new Exception("AZURE_OPENAI_ENDPOINT is not set.");
        var deploymentName = Environment.GetEnvironmentVariable("AZURE_OPENAI_DEPLOYMENT_NAME") ?? "gpt-4o-mini";
        var chatClient = new AIProjectClient(new Uri(endpoint), new DefaultAzureCredential())
            .GetProjectOpenAIClient()
            .GetProjectResponsesClient()
            .AsIChatClient(deploymentName);

        // Create agents
        AIAgent emailAnalysisAgent = GetEmailAnalysisAgent(chatClient);
        AIAgent emailAssistantAgent = GetEmailAssistantAgent(chatClient);
        AIAgent emailSummaryAgent = GetEmailSummaryAgent(chatClient);

        // Create executors
        var emailAnalysisExecutor = new EmailAnalysisExecutor(emailAnalysisAgent);
        var emailAssistantExecutor = new EmailAssistantExecutor(emailAssistantAgent);
        var emailSummaryExecutor = new EmailSummaryExecutor(emailSummaryAgent);
        var sendEmailExecutor = new SendEmailExecutor();
        var handleSpamExecutor = new HandleSpamExecutor();
        var handleUncertainExecutor = new HandleUncertainExecutor();
        var databaseAccessExecutor = new DatabaseAccessExecutor();

        // Build the workflow with multi-selection fan-out
        WorkflowBuilder builder = new(emailAnalysisExecutor);
        builder.AddFanOutEdge(
            emailAnalysisExecutor,
            targets: [
                handleSpamExecutor,        // Index 0: Spam handler
                emailAssistantExecutor,    // Index 1: Email assistant (always for NotSpam)
                emailSummaryExecutor,      // Index 2: Summarizer (conditionally for long NotSpam)
                handleUncertainExecutor,   // Index 3: Uncertain handler
            ],
            targetSelector: GetTargetAssigner()
        )
        // Email assistant branch
        .AddEdge(emailAssistantExecutor, sendEmailExecutor)

        // Database persistence: conditional routing
        .AddEdge<AnalysisResult>(
            emailAnalysisExecutor,
            databaseAccessExecutor,
            condition: analysisResult => analysisResult?.EmailLength <= EmailProcessingConstants.LongEmailThreshold) // Short emails
        .AddEdge(emailSummaryExecutor, databaseAccessExecutor) // Long emails with summary

        .WithOutputFrom(handleUncertainExecutor, handleSpamExecutor, sendEmailExecutor);

        var workflow = builder.Build();

        // Read a moderately long email to trigger both assistant and summarizer
        string email = Resources.Read("email.txt");

        // Execute the workflow with custom event handling
        StreamingRun run = await InProcessExecution.RunStreamingAsync(workflow, new ChatMessage(ChatRole.User, email));
        await run.TrySendMessageAsync(new TurnToken(emitEvents: true));
        await foreach (WorkflowEvent evt in run.WatchStreamAsync().ConfigureAwait(false))
        {
            if (evt is WorkflowOutputEvent outputEvent)
            {
                Console.WriteLine($"Output: {outputEvent}");
            }

            if (evt is DatabaseEvent databaseEvent)
            {
                Console.WriteLine($"Database: {databaseEvent}");
            }
        }
    }
}

مقارنة الأنماط: التحديد المتعدد مقابل Switch-Case

نمطSwitch-Case (السابق):

// One input → exactly one output
builder.AddSwitch(spamDetectionExecutor, switchBuilder =>
    switchBuilder
    .AddCase(GetCondition(SpamDecision.NotSpam), emailAssistantExecutor)
    .AddCase(GetCondition(SpamDecision.Spam), handleSpamExecutor)
    .WithDefault(handleUncertainExecutor)
)

نمط التحديد المتعدد:

// One input → one or more outputs (dynamic fan-out)
builder.AddFanOutEdge(
    emailAnalysisExecutor,
    targets: [handleSpamExecutor, emailAssistantExecutor, emailSummaryExecutor, handleUncertainExecutor],
    targetSelector: GetTargetAssigner() // Returns list of target indices
)

المزايا الرئيسية لحواف التحديد المتعدد

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

تشغيل مثال التحديد المتعدد

عند تشغيل سير العمل هذا باستخدام بريد إلكتروني طويل:

Output: Email sent: [Professional response generated by AI]
Database: Email abc123 saved to database.

عند التشغيل باستخدام بريد إلكتروني قصير، يتم تخطي الملخص:

Output: Email sent: [Professional response generated by AI]
Database: Email def456 saved to database.

حالات استخدام Real-World

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

التنفيذ الكامل متعدد التحديدات

لتنفيذ العمل الكامل، راجع هذا النموذج في مستودع إطار عمل العامل.

ما وراء حالة التبديل: توجيه متعدد التحديد

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

سير عمل معالجة البريد الإلكتروني المتقدم

بناء على مثال حالة التبديل، ستقوم بإنشاء نظام معالجة بريد إلكتروني محسن يوضح منطق التوجيه المتطور:

• رسائل البريد الإلكتروني العشوائي → معالج بريد عشوائي واحد (مثل حالة التبديل)
• رسائل البريد الإلكتروني الشرعية → تشغيل مساعد البريد الإلكتروني دائما + ملخص المشغل الشرطي لرسائل البريد الإلكتروني الطويلة
• رسائل البريد الإلكتروني غير المؤكدة → معالج واحد غير مؤكد (مثل switch-case)
• → استمرارية قاعدة البيانات مشغلة لكل من رسائل البريد الإلكتروني القصيرة ورسائل البريد الإلكتروني الطويلة الملخصة

يتيح هذا النمط مسارات المعالجة المتوازية التي تتكيف مع خصائص المحتوى.

المفاهيم التي تمت تغطيتها

• Fan-Out الحواف

نماذج البيانات المحسنة للاختيار المتعدد

توسيع نماذج البيانات لدعم تحليل طول البريد الإلكتروني وتلخيصه:

class AnalysisResultAgent(BaseModel):
    """Enhanced structured output from email analysis agent."""

    spam_decision: Literal["NotSpam", "Spam", "Uncertain"]
    reason: str

class EmailResponse(BaseModel):
    """Response from email assistant."""

    response: str

class EmailSummaryModel(BaseModel):
    """Summary generated by email summary agent."""

    summary: str

@dataclass
class AnalysisResult:
    """Internal analysis result with email metadata for routing decisions."""

    spam_decision: str
    reason: str
    email_length: int  # Used for conditional routing
    email_summary: str  # Populated by summary agent
    email_id: str

@dataclass
class Email:
    """Email content stored in shared state."""

    email_id: str
    email_content: str

# Custom event data for database operations
class DatabaseEvent:
    """Custom event data for tracking database operations."""
    def __init__(self, message: str):
        self.message = message

    def __repr__(self) -> str:
        return f"DatabaseEvent({self.message})"

دالة التحديد: قلب التحديد المتعدد

تحدد دالة التحديد المنفذين الذين يجب أن يتلقوا كل رسالة:

LONG_EMAIL_THRESHOLD = 100

def select_targets(analysis: AnalysisResult, target_ids: list[str]) -> list[str]:
    """Intelligent routing based on spam decision and email characteristics."""

    # Target order: [handle_spam, submit_to_email_assistant, summarize_email, handle_uncertain]
    handle_spam_id, submit_to_email_assistant_id, summarize_email_id, handle_uncertain_id = target_ids

    if analysis.spam_decision == "Spam":
        # Route only to spam handler
        return [handle_spam_id]

    elif analysis.spam_decision == "NotSpam":
        # Always route to email assistant
        targets = [submit_to_email_assistant_id]

        # Conditionally add summarizer for long emails
        if analysis.email_length > LONG_EMAIL_THRESHOLD:
            targets.append(summarize_email_id)

        return targets

    else:  # Uncertain
        # Route only to uncertain handler
        return [handle_uncertain_id]

الميزات الرئيسية لدالات التحديد

1. تحديد الهدف الديناميكي: إرجاع قائمة بمعرفات المنفذ للتنشيط
2. التوجيه المدرك للمحتوى: اتخاذ القرارات استنادا إلى خصائص الرسالة
3. المعالجة المتوازية: يمكن تنفيذ أهداف متعددة في وقت واحد
4. المنطق الشرطي: التفريع المعقد استنادا إلى معايير متعددة

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

تنفيذ المنفذين الذين يتعاملون مع التحليل والتوجيه المحسنين:

EMAIL_STATE_PREFIX = "email:"
CURRENT_EMAIL_ID_KEY = "current_email_id"

@executor(id="store_email")
async def store_email(email_text: str, ctx: WorkflowContext[AgentExecutorRequest]) -> None:
    """Store email and initiate analysis."""

    new_email = Email(email_id=str(uuid4()), email_content=email_text)
    ctx.set_state(f"{EMAIL_STATE_PREFIX}{new_email.email_id}", new_email)
    ctx.set_state(CURRENT_EMAIL_ID_KEY, new_email.email_id)

    await ctx.send_message(
        AgentExecutorRequest(messages=[Message(role="user", contents=[new_email.email_content])], should_respond=True)
    )

@executor(id="to_analysis_result")
async def to_analysis_result(response: AgentExecutorResponse, ctx: WorkflowContext[AnalysisResult]) -> None:
    """Transform agent response into enriched analysis result."""

    parsed = AnalysisResultAgent.model_validate_json(response.agent_response.text)
    email_id: str = ctx.get_state(CURRENT_EMAIL_ID_KEY)
    email: Email = ctx.get_state(f"{EMAIL_STATE_PREFIX}{email_id}")

    # Create enriched analysis result with email length for routing decisions
    await ctx.send_message(
        AnalysisResult(
            spam_decision=parsed.spam_decision,
            reason=parsed.reason,
            email_length=len(email.email_content),  # Key for conditional routing
            email_summary="",
            email_id=email_id,
        )
    )

@executor(id="submit_to_email_assistant")
async def submit_to_email_assistant(analysis: AnalysisResult, ctx: WorkflowContext[AgentExecutorRequest]) -> None:
    """Handle legitimate emails by forwarding to email assistant."""

    if analysis.spam_decision != "NotSpam":
        raise RuntimeError("This executor should only handle NotSpam messages.")

    email: Email = ctx.get_state(f"{EMAIL_STATE_PREFIX}{analysis.email_id}")
    await ctx.send_message(
        AgentExecutorRequest(messages=[Message(role="user", contents=[email.email_content])], should_respond=True)
    )

@executor(id="finalize_and_send")
async def finalize_and_send(response: AgentExecutorResponse, ctx: WorkflowContext[Never, str]) -> None:
    """Final step for email assistant branch."""

    parsed = EmailResponse.model_validate_json(response.agent_response.text)
    await ctx.yield_output(f"Email sent: {parsed.response}")

@executor(id="summarize_email")
async def summarize_email(analysis: AnalysisResult, ctx: WorkflowContext[AgentExecutorRequest]) -> None:
    """Generate summary for long emails (parallel branch)."""

    # Only called for long NotSpam emails by selection function
    email: Email = ctx.get_state(f"{EMAIL_STATE_PREFIX}{analysis.email_id}")
    await ctx.send_message(
        AgentExecutorRequest(messages=[Message(role="user", contents=[email.email_content])], should_respond=True)
    )

@executor(id="merge_summary")
async def merge_summary(response: AgentExecutorResponse, ctx: WorkflowContext[AnalysisResult]) -> None:
    """Merge summary back into analysis result for database persistence."""

    summary = EmailSummaryModel.model_validate_json(response.agent_response.text)
    email_id: str = ctx.get_state(CURRENT_EMAIL_ID_KEY)
    email: Email = ctx.get_state(f"{EMAIL_STATE_PREFIX}{email_id}")

    # Create analysis result with summary for database storage
    await ctx.send_message(
        AnalysisResult(
            spam_decision="NotSpam",
            reason="",
            email_length=len(email.email_content),
            email_summary=summary.summary,  # Now includes summary
            email_id=email_id,
        )
    )

@executor(id="handle_spam")
async def handle_spam(analysis: AnalysisResult, ctx: WorkflowContext[Never, str]) -> None:
    """Handle spam emails (single target like switch-case)."""

    if analysis.spam_decision == "Spam":
        await ctx.yield_output(f"Email marked as spam: {analysis.reason}")
    else:
        raise RuntimeError("This executor should only handle Spam messages.")

@executor(id="handle_uncertain")
async def handle_uncertain(analysis: AnalysisResult, ctx: WorkflowContext[Never, str]) -> None:
    """Handle uncertain emails (single target like switch-case)."""

    if analysis.spam_decision == "Uncertain":
        email: Email | None = ctx.get_state(f"{EMAIL_STATE_PREFIX}{analysis.email_id}")
        await ctx.yield_output(
            f"Email marked as uncertain: {analysis.reason}. Email content: {getattr(email, 'email_content', '')}"
        )
    else:
        raise RuntimeError("This executor should only handle Uncertain messages.")

@executor(id="database_access")
async def database_access(analysis: AnalysisResult, ctx: WorkflowContext[Never, str]) -> None:
    """Simulate database persistence with custom events."""

    await asyncio.sleep(0.05)  # Simulate DB operation
    await ctx.add_event(WorkflowEvent("data", data=DatabaseEvent(f"Email {analysis.email_id} saved to database.")))

عوامل الذكاء الاصطناعي المحسنة

إنشاء عوامل للتحليل والمساعدة والتلخيص:

async def main() -> None:
    chat_client = OpenAIChatCompletionClient(
        model=os.environ["AZURE_OPENAI_CHAT_COMPLETION_MODEL"],
        azure_endpoint=os.environ["AZURE_OPENAI_ENDPOINT"],
        api_version=os.getenv("AZURE_OPENAI_API_VERSION"),
        credential=AzureCliCredential(),
    )

    # Enhanced analysis agent
    email_analysis_agent = AgentExecutor(
        chat_client.as_agent(
            instructions=(
                "You are a spam detection assistant that identifies spam emails. "
                "Always return JSON with fields 'spam_decision' (one of NotSpam, Spam, Uncertain) "
                "and 'reason' (string)."
            ),
            default_options={"response_format": AnalysisResultAgent},
        ),
        id="email_analysis_agent",
    )

    # Email assistant (same as before)
    email_assistant_agent = AgentExecutor(
        chat_client.as_agent(
            instructions=(
                "You are an email assistant that helps users draft responses to emails with professionalism."
            ),
            default_options={"response_format": EmailResponse},
        ),
        id="email_assistant_agent",
    )

    # New: Email summary agent for long emails
    email_summary_agent = AgentExecutor(
        chat_client.as_agent(
            instructions="You are an assistant that helps users summarize emails.",
            default_options={"response_format": EmailSummaryModel},
        ),
        id="email_summary_agent",
    )

إنشاء سير عمل متعدد التحديدات

إنشاء سير العمل مع التوجيه المتطور والمعالجة المتوازية:

    workflow = (
        WorkflowBuilder(start_executor=store_email)
        .add_edge(store_email, email_analysis_agent)
        .add_edge(email_analysis_agent, to_analysis_result)

        # Multi-selection edge group: intelligent fan-out based on content
        .add_multi_selection_edge_group(
            to_analysis_result,
            [handle_spam, submit_to_email_assistant, summarize_email, handle_uncertain],
            selection_func=select_targets,
        )

        # Email assistant branch (always for NotSpam)
        .add_edge(submit_to_email_assistant, email_assistant_agent)
        .add_edge(email_assistant_agent, finalize_and_send)

        # Summary branch (only for long NotSpam emails)
        .add_edge(summarize_email, email_summary_agent)
        .add_edge(email_summary_agent, merge_summary)

        # Database persistence: conditional routing
        .add_edge(to_analysis_result, database_access,
                 condition=lambda r: r.email_length <= LONG_EMAIL_THRESHOLD)  # Short emails
        .add_edge(merge_summary, database_access)  # Long emails with summary

        .build()
    )

التنفيذ باستخدام تدفق الأحداث

تشغيل سير العمل ومراقبة التنفيذ المتوازي من خلال الأحداث المخصصة:

    # Use a moderately long email to trigger both assistant and summarizer
    email = """
    Hello team, here are the updates for this week:

    1. Project Alpha is on track and we should have the first milestone completed by Friday.
    2. The client presentation has been scheduled for next Tuesday at 2 PM.
    3. Please review the Q4 budget allocation and provide feedback by Wednesday.

    Let me know if you have any questions or concerns.

    Best regards,
    Alex
    """

    # Stream events to see parallel execution
    async for event in workflow.run(email, stream=True):
        if isinstance(event.data, DatabaseEvent):
            print(f"Database: {event}")
        elif event.type == "output":
            print(f"Output: {event.data}")

تحديد متعدد مقابل مقارنة Switch-Case

نمطSwitch-Case (السابق):

# One input → exactly one output
.add_switch_case_edge_group(
    source,
    [
        Case(condition=lambda x: x.result == "A", target=handler_a),
        Case(condition=lambda x: x.result == "B", target=handler_b),
        Default(target=handler_c),
    ],
)

نمط التحديد المتعدد:

# One input → one or more outputs (dynamic fan-out)
.add_multi_selection_edge_group(
    source,
    [handler_a, handler_b, handler_c, handler_d],
    selection_func=intelligent_router,  # Returns list of target IDs
)

مزايا التحديد المتعدد

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

تطبيقات Real-World

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

نموذج التعليمات البرمجية متعدد التحديد

لتنفيذ العمل الكامل، راجع نموذج multi_selection_edge_group.py في مستودع إطار عمل العامل.

إنشاء سير عمل متعدد التحديدات

استخدم AddFanOutEdge مع workflow.WithEdgeAssigner عندما يجب توجيه رسالة واحدة إلى مجموعة فرعية من أهداف متعددة:

func routeAnalysis(_ int, msg any) iter.Seq[int] {
    return func(yield func(int) bool) {
        analysis, ok := msg.(AnalysisResult)
        if !ok {
            return
        }

        switch analysis.Decision {
        case Spam:
            yield(0) // spam handler
        case NotSpam:
            if !yield(1) { // email assistant
                return
            }
            if analysis.EmailLength > longEmailThreshold {
                yield(2) // summarizer
            }
        default:
            yield(3) // uncertain handler
        }
    }
}

wf, err := workflow.NewBuilder(analyzeEmail).
    AddFanOutEdge(
        analyzeEmail,
        []workflow.ExecutorBinding{spamHandler, emailAssistant, summarizer, uncertainHandler},
        workflow.WithEdgeAssigner(routeAnalysis),
    ).
    AddEdge(emailAssistant, sendEmail).
    AddEdge(summarizer, databaseAccess).
    WithOutputFrom(spamHandler, sendEmail, uncertainHandler, databaseAccess).
    Build()

نموذج التعليمات البرمجية متعدد التحديد

لتنفيذ العمل الكامل، راجع نموذج التحديد المتعدد في مستودع Agent Framework Go.

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

الأحداث