امتداد المهمة الدائمة لإطار عمل Microsoft Agent

يجلب امتداد Durable Task لإطار عمل الوكيل Microsoftالتنفيذ الدائم مباشرة إلى إطار عمل الوكيل Microsoft. يمكنك تسجيل الوكلاء باستخدام الامتداد لجعلهم دائما تلقائيا مع الجلسات الدائمة، ونقاط نهاية API مدمجة، وتوسيع موزع — دون تغييرات في منطق الوكيل الخاص بك.

متى تستخدم العوامل المتينة

اختر امتداد المهمة الدائمة عندما تحتاج:

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

إذا لم تكن بحاجة إلى تنسيق حالة دائم أو تنسيق متعدد الوكلاء، فقد يكون وكيل Microsoft Agent Framework القياسي بدون الامتداد كافيا.

Tip

للحصول على دليل خطوة بخطوة يشمل الإعداد المحلي، والمتطلبات المسبقة، والنشر، راجع tutorial على Microsoft Learn.

الهندسة

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

يدعم الامتداد نهجين للاستضافة:

• دالات Azure باستخدام حزمة التكامل دالات Azure.
• أحضر حسابك الخاص باستخدام الحزمة الأساسية.

استضافة الوكلاء

حدد وكيلك باستخدام نمط إطار عمل Microsoft Agent القياسي، ثم قم بتحسينه بإضافة Durable Task. تتعامل الإضافة تلقائيا مع استمرارية الجلسة، وإنشاء نقطة النهاية، وإدارة الحالة.

C#‎

var endpoint = Environment.GetEnvironmentVariable("AZURE_OPENAI_ENDPOINT")
    ?? throw new InvalidOperationException("AZURE_OPENAI_ENDPOINT is not set.");
var deploymentName = Environment.GetEnvironmentVariable("AZURE_OPENAI_DEPLOYMENT")
    ?? "gpt-4o-mini";

// Create an AI agent using the Microsoft Agent Framework pattern
AIAgent agent = new AIProjectClient(new Uri(endpoint), new DefaultAzureCredential())
    .AsAIAgent(
        model: deploymentName,
        instructions: "You are a professional content writer who creates engaging, "
                    + "well-structured documents for any given topic.",
        name: "DocumentPublisher");

// One line to make the agent durable with serverless hosting
using IHost app = FunctionsApplication
    .CreateBuilder(args)
    .ConfigureFunctionsWebApplication()
    .ConfigureDurableAgents(options => options.AddAIAgent(agent))
    .Build();
app.Run();

Python

import os
from agent_framework.azure import FoundryChatClient, AgentFunctionApp
from azure.identity import DefaultAzureCredential

endpoint = os.getenv("AZURE_OPENAI_ENDPOINT")
deployment_name = os.getenv("AZURE_OPENAI_DEPLOYMENT") or "gpt-4o-mini"

client = FoundryChatClient(
    endpoint=endpoint,
    credential=DefaultAzureCredential()
)

agent = client.as_agent(
    model=deployment_name,
    instructions="You are a professional content writer who creates engaging, "
                 "well-structured documents for any given topic.",
    name="DocumentPublisher"
)

# One line to make the agent durable with serverless hosting
app = AgentFunctionApp(agents=[agent])

C#‎

var endpoint = Environment.GetEnvironmentVariable("AZURE_OPENAI_ENDPOINT")
    ?? throw new InvalidOperationException("AZURE_OPENAI_ENDPOINT is not set.");
var deploymentName = Environment.GetEnvironmentVariable("AZURE_OPENAI_DEPLOYMENT")
    ?? "gpt-4o-mini";

// Create an AI agent using the Microsoft Agent Framework pattern
AIAgent agent = new AIProjectClient(new Uri(endpoint), new DefaultAzureCredential())
    .AsAIAgent(
        model: deploymentName,
        instructions: "You are a professional content writer who creates engaging, "
                    + "well-structured documents for any given topic.",
        name: "DocumentPublisher");

// Host the agent with Durable Task Scheduler
string connectionString = "Endpoint=http://localhost:8080;TaskHub=default;Authentication=None";

IHost host = Host.CreateDefaultBuilder(args)
    .ConfigureServices(services =>
    {
        services.ConfigureDurableAgents(
            options => options.AddAIAgent(agent),
            workerBuilder: builder => builder.UseDurableTaskScheduler(connectionString),
            clientBuilder: builder => builder.UseDurableTaskScheduler(connectionString));
    })
    .Build();

await host.StartAsync();

Python

from agent_framework.azure import FoundryChatClient, DurableAIAgentWorker
from azure.identity import AzureCliCredential
from durabletask.azuremanaged.worker import DurableTaskSchedulerWorker

endpoint = os.getenv("AZURE_OPENAI_ENDPOINT")
deployment_name = os.getenv("AZURE_OPENAI_DEPLOYMENT") or "gpt-4o-mini"

client = FoundryChatClient(endpoint=endpoint, credential=AzureCliCredential())

agent = client.as_agent(
    model=deployment_name,
    name="DocumentPublisher",
    instructions="You are a professional content writer who creates engaging, "
                 "well-structured documents for any given topic.",
)

# Create a worker connected to the Durable Task Scheduler
worker = DurableTaskSchedulerWorker(
    host_address="http://localhost:8080",
    secure_channel=False,
    taskhub="default",
)

# Register the agent and start processing
agent_worker = DurableAIAgentWorker(worker)
agent_worker.add_agent(agent)
worker.start()

ملحوظة

تستخدم أمثلة C#‎ AIProjectClient من حزمة Azure.AI.Projects، وهي النمط الموصى به للمشاريع الجديدة. تستخدم الأمثلة Python FoundryChatClient من agent_framework.azure. كلا العميلين يدعمان الامتداد .AsAIAgent() / .as_agent() الذي يتكامل مع امتداد المهمة الدائمة. راجع samples على GitHub للحصول على أحدث مراجع الحزمة.

تنسيق الوكلاء المتعدد مع نقاط تفتيش متينة

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

استخدم context.GetAgent() (C#) أو app.get_agent() (Python) لاسترجاع الوكلاء المسجلين داخل التوزيع الأوركسترالي. يضمن الغلاف المسترجع DurableAIAgent تتبع المكالمات والتحقق منها من قبل إطار العمل.

المثال التالي يوضح سير عمل متعدد الوكلاء متسلسلا حيث يجمع وكيل البحث المعلومات وينتج وكيل كاتب وثيقة.

C#‎

[Function(nameof(DocumentPublishingOrchestration))]
public async Task<string> DocumentPublishingOrchestration(
    [OrchestrationTrigger] TaskOrchestrationContext context)
{
    var docRequest = context.GetInput<DocumentRequest>();

    DurableAIAgent researchAgent = context.GetAgent("ResearchAgent");
    DurableAIAgent writerAgent = context.GetAgent("DocumentPublisherAgent");

    // Step 1: Research the topic
    AgentResponse<ResearchResult> researchResult = await researchAgent
        .RunAsync<ResearchResult>(
            $"Research the following topic: {docRequest.Topic}");

    // Step 2: Write the document using the research findings
    AgentResponse<DocumentResponse> document = await writerAgent
        .RunAsync<DocumentResponse>(
            $"""Create a document about {docRequest.Topic}.
            Research findings: {researchResult.Result.Findings}""");

    // Step 3: Publish
    return await context.CallActivityAsync<string>(
        nameof(PublishDocument),
        new { docRequest.Topic, document.Result.Text });
}

Python

@app.orchestration_trigger(context_name="context")
def document_publishing_orchestration(context: DurableOrchestrationContext):
    doc_request = context.get_input()

    research_agent = app.get_agent(context, "ResearchAgent")
    writer_agent = app.get_agent(context, "DocumentPublisherAgent")

    research_session = research_agent.create_session()
    writer_session = writer_agent.create_session()

    # Step 1: Research the topic
    research_result = yield research_agent.run(
        messages=f"Research the following topic: {doc_request['topic']}",
        session=research_session,
    )

    # Step 2: Write the document using the research findings
    document = yield writer_agent.run(
        messages=f"""Create a document about {doc_request['topic']}.
        Research findings: {research_result.text}""",
        session=writer_session,
    )

    # Step 3: Publish
    return (yield context.call_activity("publish_document", {
        "title": doc_request["topic"],
        "content": document.text
    }))

C#‎

static async Task<string> DocumentPublishingOrchestration(
    TaskOrchestrationContext context, DocumentRequest docRequest)
{
    DurableAIAgent researchAgent = context.GetAgent("ResearchAgent");
    DurableAIAgent writerAgent = context.GetAgent("DocumentPublisherAgent");

    // Step 1: Research the topic
    AgentResponse<ResearchResult> researchResult = await researchAgent
        .RunAsync<ResearchResult>(
            $"Research the following topic: {docRequest.Topic}");

    // Step 2: Write the document using the research findings
    AgentResponse<DocumentResponse> document = await writerAgent
        .RunAsync<DocumentResponse>(
            $"""Create a document about {docRequest.Topic}.
            Research findings: {researchResult.Result.Findings}""");

    // Step 3: Publish
    return await context.CallActivityAsync<string>(
        nameof(PublishDocument),
        new { docRequest.Topic, document.Result.Text });
}

Python

from agent_framework.azure import DurableAIAgentOrchestrationContext

def document_publishing_orchestration(ctx, doc_request: dict):
    agent_context = DurableAIAgentOrchestrationContext(ctx)

    research = agent_context.get_agent("ResearchAgent")
    writer = agent_context.get_agent("DocumentPublisherAgent")

    research_session = research.create_session()
    writer_session = writer.create_session()

    # Step 1: Research the topic
    research_result = yield research.run(
        messages=f"Research the following topic: {doc_request['topic']}",
        session=research_session,
    )

    # Step 2: Write the document using the research findings
    document = yield writer.run(
        messages=f"""Create a document about {doc_request['topic']}.
        Research findings: {research_result.text}""",
        session=writer_session,
    )

    # Step 3: Publish
    return (yield ctx.call_activity(publish_document, input={
        "title": doc_request["topic"],
        "content": document.text
    }))

سير العمل المعتمد على الرسوم البيانية مع إطار عمل Microsoft Agent

يدعم امتداد Durable Task أيضا سير عمل إطار عمل الوكلاء Microsoft، والتي تستخدم نموذج برمجة بياني تصريحي (WorkflowBuilder) لتعريف خطوط أنابيب متعددة الخطوات للمنفذين والوكلاء. يقوم الامتداد تلقائيا بالتحقق من كل خطوة في الرسم البياني ويتعافى من الأعطال دون تغييرات في تعريف سير العمل.

تكمل سير العمل التنسيقات: تستخدم التنسيقات لتنسيق الوكلاء الإلزامي مع المنطق الشرطي، وسير العمل لطوبولوجيات الرسوم البيانية الثابتة مع توجيه رسائل معتمد من النوع.

سير العمل التسلسلي

المثال التالي يربط ثلاثة منفذين في سير عمل لإلغاء الطلب: ابحث عن الطلب، ألغ الطلب، ثم أرسل رسالة تأكيد.

C#‎

OrderLookup orderLookup = new();
OrderCancel orderCancel = new();
SendEmail sendEmail = new();

Workflow cancelOrder = new WorkflowBuilder(orderLookup)
    .WithName("CancelOrder")
    .WithDescription("Cancel an order and notify the customer")
    .AddEdge(orderLookup, orderCancel)
    .AddEdge(orderCancel, sendEmail)
    .Build();

using IHost app = FunctionsApplication
    .CreateBuilder(args)
    .ConfigureFunctionsWebApplication()
    .ConfigureDurableWorkflows(workflows => workflows.AddWorkflows(cancelOrder))
    .Build();
app.Run();

منفذو OrderLookup، OrderCancel، و SendEmail هي منفذات Agent Framework القياسية Microsoft القياسية بدون كود خاص ب Durable. للاطلاع على التنفيذات الكاملة، راجع samples على GitHub.

Python

from agent_framework import Workflow, WorkflowBuilder, WorkflowContext, executor
from agent_framework.azure import AgentFunctionApp

@executor(id="store_email")
async def store_email(email_text: str, ctx: WorkflowContext) -> None:
    ctx.set_state("current_email", email_text)
    await ctx.send_message(email_text)

@executor(id="process_email")
async def process_email(email_text: str, ctx: WorkflowContext) -> None:
    result = f"Processed: {email_text[:50]}..."
    await ctx.send_message(result)

@executor(id="finalize")
async def finalize(result: str, ctx: WorkflowContext[None, str]) -> None:
    await ctx.yield_output(f"Complete: {result}")

workflow = (
    WorkflowBuilder(start_executor=store_email)
    .add_edge(store_email, process_email)
    .add_edge(process_email, finalize)
    .build()
)

app = AgentFunctionApp(workflow=workflow)

string dtsConnectionString = Environment.GetEnvironmentVariable("DURABLE_TASK_SCHEDULER_CONNECTION_STRING")
    ?? "Endpoint=http://localhost:8080;TaskHub=default;Authentication=None";

OrderLookup orderLookup = new();
OrderCancel orderCancel = new();
SendEmail sendEmail = new();

Workflow cancelOrder = new WorkflowBuilder(orderLookup)
    .WithName("CancelOrder")
    .WithDescription("Cancel an order and notify the customer")
    .AddEdge(orderLookup, orderCancel)
    .AddEdge(orderCancel, sendEmail)
    .Build();

IHost host = Host.CreateDefaultBuilder(args)
    .ConfigureServices(services =>
    {
        services.ConfigureDurableWorkflows(
            workflowOptions => workflowOptions.AddWorkflow(cancelOrder),
            workerBuilder: builder => builder.UseDurableTaskScheduler(dtsConnectionString),
            clientBuilder: builder => builder.UseDurableTaskScheduler(dtsConnectionString));
    })
    .Build();

await host.StartAsync();

IWorkflowClient workflowClient = host.Services.GetRequiredService<IWorkflowClient>();
IAwaitableWorkflowRun run = (IAwaitableWorkflowRun)await workflowClient.RunAsync(cancelOrder, "ORD-12345");
string? result = await run.WaitForCompletionAsync<string>();

منفذو OrderLookup، OrderCancel، و SendEmail هي منفذات Agent Framework القياسية Microsoft القياسية بدون كود خاص ب Durable. للاطلاع على التنفيذات الكاملة، راجع samples على GitHub.

سير عمل المراحل/الدخول (المتزامن)

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

C#‎

ChatClient chatClient = new AzureOpenAIClient(
    new Uri(endpoint), new DefaultAzureCredential()).GetChatClient(deploymentName);

AIAgent physicist = chatClient.AsAIAgent(
    "You are a physics expert. Be concise (2-3 sentences).", "Physicist");
AIAgent chemist = chatClient.AsAIAgent(
    "You are a chemistry expert. Be concise (2-3 sentences).", "Chemist");

ParseQuestionExecutor parseQuestion = new();
AggregatorExecutor aggregator = new();

Workflow workflow = new WorkflowBuilder(parseQuestion)
    .WithName("ExpertReview")
    .AddFanOutEdge(parseQuestion, [physicist, chemist])
    .AddFanInBarrierEdge([physicist, chemist], aggregator)
    .Build();

using IHost app = FunctionsApplication
    .CreateBuilder(args)
    .ConfigureFunctionsWebApplication()
    .ConfigureDurableWorkflows(workflows => workflows.AddWorkflows(workflow))
    .Build();
app.Run();

ParseQuestionExecutor و AggregatorExecutor هما منفذو Microsoft Agent Framework القياسي بدون كود خاص ب Durable. للاطلاع على التنفيذات الكاملة، راجع samples على GitHub.

Python

from agent_framework import (
    Agent, AgentExecutorResponse, Workflow,
    WorkflowBuilder, WorkflowContext, executor,
)
from agent_framework.azure import AgentFunctionApp

chat_client = ...  # OpenAIChatCompletionClient or FoundryChatClient

sentiment_agent = Agent(
    client=chat_client,
    name="SentimentAnalysisAgent",
    instructions="You are a sentiment analysis expert. Analyze the sentiment of the given text.",
)

keyword_agent = Agent(
    client=chat_client,
    name="KeywordExtractionAgent",
    instructions="You are a keyword extraction expert. Extract important keywords from the given text.",
)

@executor(id="input_router")
async def input_router(doc: str, ctx: WorkflowContext) -> None:
    await ctx.send_message(doc)

@executor(id="prepare_for_output")
async def prepare_for_output(
    analyses: list[AgentExecutorResponse], ctx: WorkflowContext[None, str]
) -> None:
    parts = [f"[{a.executor_id}]: {a.agent_response.text}" for a in analyses]
    await ctx.yield_output("\n\n".join(parts))

workflow = (
    WorkflowBuilder(start_executor=input_router)
    .add_fan_out_edges(source=input_router, targets=[sentiment_agent, keyword_agent])
    .add_fan_in_edges(sources=[sentiment_agent, keyword_agent], target=prepare_for_output)
    .build()
)

app = AgentFunctionApp(workflow=workflow)

string dtsConnectionString = Environment.GetEnvironmentVariable("DURABLE_TASK_SCHEDULER_CONNECTION_STRING")
    ?? "Endpoint=http://localhost:8080;TaskHub=default;Authentication=None";

ChatClient chatClient = new AzureOpenAIClient(
    new Uri(endpoint), new DefaultAzureCredential()).GetChatClient(deploymentName);

ParseQuestionExecutor parseQuestion = new();
AIAgent physicist = chatClient.AsAIAgent(
    "You are a physics expert. Be concise (2-3 sentences).", "Physicist");
AIAgent chemist = chatClient.AsAIAgent(
    "You are a chemistry expert. Be concise (2-3 sentences).", "Chemist");
AggregatorExecutor aggregator = new();

Workflow workflow = new WorkflowBuilder(parseQuestion)
    .WithName("ExpertReview")
    .AddFanOutEdge(parseQuestion, [physicist, chemist])
    .AddFanInBarrierEdge([physicist, chemist], aggregator)
    .Build();

IHost host = Host.CreateDefaultBuilder(args)
    .ConfigureServices(services =>
    {
        services.ConfigureDurableOptions(
            options => options.Workflows.AddWorkflow(workflow),
            workerBuilder: builder => builder.UseDurableTaskScheduler(dtsConnectionString),
            clientBuilder: builder => builder.UseDurableTaskScheduler(dtsConnectionString));
    })
    .Build();

await host.StartAsync();

IWorkflowClient workflowClient = host.Services.GetRequiredService<IWorkflowClient>();
IWorkflowRun run = await workflowClient.RunAsync(workflow, "Why is the sky blue?");

if (run is IAwaitableWorkflowRun awaitableRun)
{
    string? result = await awaitableRun.WaitForCompletionAsync<string>();
    Console.WriteLine(result);
}

ParseQuestionExecutor و AggregatorExecutor هما منفذو Microsoft Agent Framework القياسي بدون كود خاص ب Durable. للاطلاع على التنفيذات الكاملة، راجع samples على GitHub.

سير عمل التوجيه الشرطي

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

C#‎

AIAgent spamDetector = chatClient.AsAIAgent(
    "You are a spam detection assistant. Return JSON with is_spam (bool) and reason (string).",
    "SpamDetectionAgent");
AIAgent emailAssistant = chatClient.AsAIAgent(
    "You are an email assistant. Draft a professional response.",
    "EmailAssistantAgent");

SpamHandlerExecutor spamHandler = new();
EmailSenderExecutor emailSender = new();

Workflow workflow = new WorkflowBuilder(spamDetector)
    .WithName("EmailClassification")
    .AddSwitchCaseEdgeGroup(spamDetector, [
        new Case(condition: IsSpamDetected, target: spamHandler),
        new Default(target: emailAssistant),
    ])
    .AddEdge(emailAssistant, emailSender)
    .Build();

using IHost app = FunctionsApplication
    .CreateBuilder(args)
    .ConfigureFunctionsWebApplication()
    .ConfigureDurableWorkflows(workflows => workflows.AddWorkflows(workflow))
    .Build();
app.Run();

SpamHandlerExecutor و EmailSenderExecutor هما منفذو Microsoft Agent Framework القياسي بدون كود خاص ب Durable. للاطلاع على التنفيذات الكاملة، راجع samples على GitHub.

Python

from agent_framework import (
    Agent, AgentExecutorResponse, Case, Default, Executor,
    Workflow, WorkflowBuilder, WorkflowContext, handler,
)
from agent_framework.azure import AgentFunctionApp
from pydantic import BaseModel

class SpamDetectionResult(BaseModel):
    is_spam: bool
    reason: str

chat_client = ...  # FoundryChatClient or OpenAIChatCompletionClient

spam_agent = Agent(
    client=chat_client,
    name="SpamDetectionAgent",
    instructions="You are a spam detection assistant. Return JSON with is_spam and reason.",
    default_options={"response_format": SpamDetectionResult},
)

email_agent = Agent(
    client=chat_client,
    name="EmailAssistantAgent",
    instructions="You are an email assistant that drafts professional responses.",
)

class SpamHandlerExecutor(Executor):
    @handler
    async def handle(self, response: AgentExecutorResponse, ctx: WorkflowContext[None, str]) -> None:
        text = response.agent_response.text
        result = SpamDetectionResult.model_validate_json(text)
        await ctx.yield_output(f"Email marked as spam: {result.reason}")

class EmailSenderExecutor(Executor):
    @handler
    async def handle(self, response: AgentExecutorResponse, ctx: WorkflowContext[None, str]) -> None:
        await ctx.yield_output(f"Email sent: {response.agent_response.text}")

def is_spam_detected(message) -> bool:
    if not isinstance(message, AgentExecutorResponse):
        return False
    result = SpamDetectionResult.model_validate_json(message.agent_response.text)
    return result.is_spam

spam_handler = SpamHandlerExecutor(id="spam_handler")
email_sender = EmailSenderExecutor(id="email_sender")

workflow = (
    WorkflowBuilder(start_executor=spam_agent)
    .add_switch_case_edge_group(spam_agent, [
        Case(condition=is_spam_detected, target=spam_handler),
        Default(target=email_agent),
    ])
    .add_edge(email_agent, email_sender)
    .build()
)

app = AgentFunctionApp(workflow=workflow)

string dtsConnectionString = Environment.GetEnvironmentVariable("DURABLE_TASK_SCHEDULER_CONNECTION_STRING")
    ?? "Endpoint=http://localhost:8080;TaskHub=default;Authentication=None";

ChatClient chatClient = new AzureOpenAIClient(
    new Uri(endpoint), new DefaultAzureCredential()).GetChatClient(deploymentName);

AIAgent spamDetector = chatClient.AsAIAgent(
    "You are a spam detection assistant. Return JSON with is_spam (bool) and reason (string).",
    "SpamDetectionAgent");
AIAgent emailAssistant = chatClient.AsAIAgent(
    "You are an email assistant. Draft a professional response.",
    "EmailAssistantAgent");

SpamHandlerExecutor spamHandler = new();
EmailSenderExecutor emailSender = new();

Workflow workflow = new WorkflowBuilder(spamDetector)
    .WithName("EmailClassification")
    .AddSwitchCaseEdgeGroup(spamDetector, [
        new Case(condition: IsSpamDetected, target: spamHandler),
        new Default(target: emailAssistant),
    ])
    .AddEdge(emailAssistant, emailSender)
    .Build();

IHost host = Host.CreateDefaultBuilder(args)
    .ConfigureServices(services =>
    {
        services.ConfigureDurableWorkflows(
            workflowOptions => workflowOptions.AddWorkflow(workflow),
            workerBuilder: builder => builder.UseDurableTaskScheduler(dtsConnectionString),
            clientBuilder: builder => builder.UseDurableTaskScheduler(dtsConnectionString));
    })
    .Build();

await host.StartAsync();

IWorkflowClient workflowClient = host.Services.GetRequiredService<IWorkflowClient>();
IAwaitableWorkflowRun run = (IAwaitableWorkflowRun)await workflowClient.RunAsync(workflow, "Check this email for spam");
string? result = await run.WaitForCompletionAsync<string>();

SpamHandlerExecutor و EmailSenderExecutor هما منفذو Microsoft Agent Framework القياسي بدون كود خاص ب Durable. للاطلاع على التنفيذات الكاملة، راجع samples على GitHub.

سير عمل الإنسان في الحلقة (HITL)

يمكنك إيقاف تنفيذ سير العمل مؤقتا عند نقاط محددة لانتظار المدخلات الخارجية قبل الاستمرار. يستخدم نموذج سير عمل إطار عمل الوكلاء Microsoft <عقد c0 /> (باللغة .NET) أو (باللغة Python) لتعريف نقاط التوقف. المثال التالي ينفذ سير عمل لتعويض المصاريف مع موافقة مدير يتبعها موافقات ميزانية وتوافق متوازي.

C#‎

CreateApprovalRequest createRequest = new();
RequestPort<ApprovalRequest, ApprovalResponse> managerApproval =
    RequestPort.Create<ApprovalRequest, ApprovalResponse>("ManagerApproval");
PrepareFinanceReview prepareFinanceReview = new();
RequestPort<ApprovalRequest, ApprovalResponse> budgetApproval =
    RequestPort.Create<ApprovalRequest, ApprovalResponse>("BudgetApproval");
RequestPort<ApprovalRequest, ApprovalResponse> complianceApproval =
    RequestPort.Create<ApprovalRequest, ApprovalResponse>("ComplianceApproval");
ExpenseReimburse reimburse = new();

Workflow expenseApproval = new WorkflowBuilder(createRequest)
    .WithName("ExpenseReimbursement")
    .WithDescription("Expense reimbursement with manager and parallel finance approvals")
    .AddEdge(createRequest, managerApproval)
    .AddEdge(managerApproval, prepareFinanceReview)
    .AddFanOutEdge(prepareFinanceReview, [budgetApproval, complianceApproval])
    .AddFanInBarrierEdge([budgetApproval, complianceApproval], reimburse)
    .Build();

using IHost app = FunctionsApplication
    .CreateBuilder(args)
    .ConfigureFunctionsWebApplication()
    .ConfigureDurableWorkflows(workflows =>
        workflows.AddWorkflow(expenseApproval, exposeStatusEndpoint: true))
    .Build();
app.Run();

يقوم الإطار بتوليد ثلاث نقاط نهاية HTTP تلقائيا لتفاعل HITL.

• POST /api/workflows/{name}/run : ابدأ سير العمل
• GET /api/workflows/{name}/status/{id} حالة التحقق والموافقات المعلقة :
• POST /api/workflows/{name}/respond/{id} : أرسل رد الموافقة إلى السيرة الذاتية

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

public record ApprovalRequest(string ExpenseId, decimal Amount, string EmployeeName);
public record ApprovalResponse(bool Approved, string? Comments);

منفذو CreateApprovalRequest، PrepareFinanceReview، و ExpenseReimburse هي منفذات Agent Framework القياسية Microsoft القياسية بدون كود خاص ب Durable. للاطلاع على التنفيذات الكاملة، راجع samples على GitHub.

Python

from agent_framework import (
    Agent, Executor, Workflow, WorkflowBuilder,
    WorkflowContext, handler, response_handler,
)
from agent_framework.azure import AgentFunctionApp
from pydantic import BaseModel

class HumanApprovalResponse(BaseModel):
    approved: bool
    reviewer_notes: str = ""

chat_client = ...  # FoundryChatClient or OpenAIChatCompletionClient

content_analyzer_agent = Agent(
    client=chat_client,
    name="ContentAnalyzerAgent",
    instructions="You are a content moderation assistant. Analyze content for policy compliance.",
)

class InputRouterExecutor(Executor):
    def __init__(self):
        super().__init__(id="input_router")

    @handler
    async def route_input(self, input_text: str, ctx: WorkflowContext) -> None:
        await ctx.send_message(input_text)

class HumanReviewExecutor(Executor):
    def __init__(self):
        super().__init__(id="human_review_executor")

    @handler
    async def request_review(self, data, ctx: WorkflowContext) -> None:
        approval_request = {
            "content": data,
            "prompt": "Please approve or reject this content.",
        }
        await ctx.request_info(
            request_data=approval_request,
            response_type=HumanApprovalResponse,
        )

    @response_handler
    async def handle_approval_response(self, original_request, response, ctx: WorkflowContext) -> None:
        status = "approved" if response.approved else "rejected"
        await ctx.send_message(f"Content {status}: {response.reviewer_notes}")

class PublishExecutor(Executor):
    def __init__(self):
        super().__init__(id="publish_executor")

    @handler
    async def handle_result(self, result: str, ctx: WorkflowContext[None, str]) -> None:
        await ctx.yield_output(result)

input_router = InputRouterExecutor()
human_review = HumanReviewExecutor()
publish = PublishExecutor()

workflow = (
    WorkflowBuilder(start_executor=input_router)
    .add_edge(input_router, content_analyzer_agent)
    .add_edge(content_analyzer_agent, human_review)
    .add_edge(human_review, publish)
    .build()
)

app = AgentFunctionApp(workflow=workflow)

string dtsConnectionString = Environment.GetEnvironmentVariable("DURABLE_TASK_SCHEDULER_CONNECTION_STRING")
    ?? "Endpoint=http://localhost:8080;TaskHub=default;Authentication=None";

CreateApprovalRequest createRequest = new();
RequestPort<ApprovalRequest, ApprovalResponse> managerApproval =
    RequestPort.Create<ApprovalRequest, ApprovalResponse>("ManagerApproval");
PrepareFinanceReview prepareFinanceReview = new();
RequestPort<ApprovalRequest, ApprovalResponse> budgetApproval =
    RequestPort.Create<ApprovalRequest, ApprovalResponse>("BudgetApproval");
RequestPort<ApprovalRequest, ApprovalResponse> complianceApproval =
    RequestPort.Create<ApprovalRequest, ApprovalResponse>("ComplianceApproval");
ExpenseReimburse reimburse = new();

Workflow expenseApproval = new WorkflowBuilder(createRequest)
    .WithName("ExpenseReimbursement")
    .AddEdge(createRequest, managerApproval)
    .AddEdge(managerApproval, prepareFinanceReview)
    .AddFanOutEdge(prepareFinanceReview, [budgetApproval, complianceApproval])
    .AddFanInBarrierEdge([budgetApproval, complianceApproval], reimburse)
    .Build();

IHost host = Host.CreateDefaultBuilder(args)
    .ConfigureServices(services =>
    {
        services.ConfigureDurableWorkflows(
            options => options.AddWorkflow(expenseApproval),
            workerBuilder: builder => builder.UseDurableTaskScheduler(dtsConnectionString),
            clientBuilder: builder => builder.UseDurableTaskScheduler(dtsConnectionString));
    })
    .Build();

await host.StartAsync();

IWorkflowClient workflowClient = host.Services.GetRequiredService<IWorkflowClient>();
IStreamingWorkflowRun run = await workflowClient.StreamAsync(expenseApproval, "EXP-2025-001");

await foreach (WorkflowEvent evt in run.WatchStreamAsync())
{
    switch (evt)
    {
        case DurableWorkflowWaitingForInputEvent requestEvent:
            Console.WriteLine($"Workflow paused at: {requestEvent.RequestPort.Id}");
            ApprovalResponse approval = new(Approved: true, Comments: "Approved.");
            await run.SendResponseAsync(requestEvent, approval);
            break;

        case DurableWorkflowCompletedEvent completedEvent:
            Console.WriteLine($"Workflow completed: {completedEvent.Result}");
            break;
    }
}

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

public record ApprovalRequest(string ExpenseId, decimal Amount, string EmployeeName);
public record ApprovalResponse(bool Approved, string? Comments);

منفذو CreateApprovalRequest، PrepareFinanceReview، و ExpenseReimburse هي منفذات Agent Framework القياسية Microsoft القياسية بدون كود خاص ب Durable. للاطلاع على التنفيذات الكاملة، راجع samples على GitHub.

لوحة معلومات Durable Task Scheduler

استخدم لوحة تحكم جدولة المهام الدائمة للحصول على رؤية كاملة لوكلائك المتينين، والتنسيقات، وسير العمل المعتمد على الرسوم البيانية:

• عرض سجل المحادثات لكل جلسة وكيل
• فحص استدعاءات الأدوات والمخرجات المنظمة
• تتبع التنسيق وتدفقات تنفيذ سير العمل
• مقاييس أداء المراقبة

كل من التطوير المحلي (عبر المحاكي) وعمليات الإنتاج يقدمان نفس تجربة لوحة التحكم.

تظهر لقطة الشاشة التالية جلسة وكيل مع تاريخ المحادثة وتفاصيل الجلسة:

تظهر لقطة الشاشة التالية تنسيقا حتميا مع تفاصيل تنفيذ النشاط:

مدة الجلسة حتى العيش (TTL) للوكلاء الدائمين

تحافظ جلسات الوكلاء الدائمة تلقائيا على سجل المحادثة وحالته، والتي قد تتراكم إلى أجل غير مسمى. توفر ميزة الوقت للتشغيل (TTL) تنظيفا تلقائيا للجلسات الخمولة، مما يمنع استهلاك موارد التخزين وزيادة التكاليف.

عندما تبقى جلسة الوكيل في وضع الخمول لفترة أطول من فترة TTL المكونة، يتم حذف حالة الجلسة تلقائيا. كل تفاعل جديد يعيد ضبط مؤقت TTL، مما يطيل عمر الجلسة.

القيم الافتراضية

• مدة TTL الافتراضية: 14 يوما
• الحد الأدنى لتأخير حذف TTL: 5 دقائق

الإعداد

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

ملحوظة

تكوين TTL متوفر حاليا بصيغة .NET فقط.

services.ConfigureDurableAgents(
    options =>
    {
        // Set global default TTL to 7 days
        options.DefaultTimeToLive = TimeSpan.FromDays(7);

        // Agent with custom TTL of 1 day
        options.AddAIAgent(shortLivedAgent, timeToLive: TimeSpan.FromDays(1));

        // Agent with custom TTL of 90 days
        options.AddAIAgent(longLivedAgent, timeToLive: TimeSpan.FromDays(90));

        // Agent using global default (7 days)
        options.AddAIAgent(defaultAgent);

        // Agent with no TTL (never expires)
        options.AddAIAgent(permanentAgent, timeToLive: null);
    });

القيود المعروفة

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

• الكمون.
  يتم توجيه جميع تفاعلات الوكلاء عبر جدولة المهام الدائمة، التي تضيف زمن استجابة مقارنة بتنفيذ الوكيل في الذاكرة. هذا المقابل يوفر متانة وتكبير موزع.

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

• انتهاء صلاحية TTL.
  يعتمد مؤقت TTL على وقت ساعة الجدار منذ آخر رسالة، وليس على وقت النشاط التراكمي. بمجرد حذف الجلسة (عن طريق انتهاء صلاحية TTL أو الحذف اليدوي)، لا يمكن استعادة سجل المحادثات.

قدرات إضافية

الأنماط المتقدمة التالية متوفرة في <عينات c0> على GitHub لكنها غير مغطاة بالتفصيل في هذا المقال:

النمط	الوصف
الأدوات طويلة الأمد	يمكن للوكلاء بدء عمليات تنسيق متينة من خلال استدعاءات الأدوات، مما يتيح أدوات تعمل لفترات طويلة مع تتبع التقدم.
الوكيل كأداة MCP	كشف الوكلاء الدائمين كأدوات بروتوكول السياق النموذجي (MCP)، مما يسمح لوكلاء أو عملاء آخرين باستدعاؤها عبر واجهة قياسية.
البث الموثوق	استخدم Redis Streams (أو وسائل النقل المشابهة) لبث الرموز القابلة للاستئناف التي تصمد بعد انقطاعات العملاء وإعادة الاتصال.

الارتباطات ذات الصلة

لعينات الكود الكاملة:

• .NET دالات Azure
• .NET سير العمل الدائم
• Python دالات Azure (الوكلاء وسير العمل)

لعينات الكود الكاملة:

• .NET أي مضيف
• .NET سير العمل الدائم
• Python أي مضيف (الوكلاء وسير العمل)

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

• نظرة عامة على المهام الدائمة لوكلاء الذكاء الاصطناعي
• أنماط التطبيقات الوكالية
• دالات Azure (متين)
• نظرة عامة على Durable Task Scheduler