Microsoft Agent Framework için Dayanıklı Görev Uzantısı

Microsoft Agent Framework için Dayanıklı Görev uzantısı, dayanıklı yürütmeyi doğrudan Microsoft Agent Framework içine getirir. Aracıları, aracı mantığınızda değişiklik yapmadan kalıcı oturumlar, yerleşik API uç noktaları ve dağıtılmış ölçeklendirme ile otomatik olarak dayanıklı hale getirmek için uzantıya kaydedebilirsiniz.

Dayanıklı ajanlar ne zaman kullanılmalıdır?

İhtiyacınız olduğunda Dayanıklı Görev uzantısını seçin:

• Kalıcı konuşma durumu — Aracı oturumları bağlamı kaybetmeden işlem kilitlenmelerinden, yeniden başlatmalardan ve ölçeklendirme olaylarından kurtulur.
• Çok aracılı düzenleme — Otomatik denetim noktası oluşturma ve hata kurtarma ile belirlenimci iş akışlarında özel aracıları koordine edin.
• Uzun süre çalışan iş akışları : İşlem kaynaklarını kullanmadan saatler, günler veya haftalar sürebilen döngüdeki insan onaylarını veya zamanlanmış beklemeleri destekler.
• Ölçeklenebilir, sunucusuz barındırma — Azure İşlevleri Esnek Tüketim planında binlerce eşzamanlı aracı oturumuna (veya sıfıra kadar) ölçeklendirin.

Dayanıklı duruma veya çok aracılı koordinasyona ihtiyacınız yoksa, uzantı olmadan standart bir Microsoft Agent Framework aracısı yeterli olabilir.

Tip

Yerel kurulum, önkoşullar ve dağıtım gibi adım adım izlenecek yol için Microsoft Learn üzerindeki tutorial bölümüne bakın.

Architecture

Uzantı, her aracı oturumunun konuşma durumunu ve denetim noktasını otomatik olarak yöneten dayanıklı bir varlık olduğu varlık tabanlı aracı döngülerini dahili olarak uygular.

Uzantı iki barındırma yaklaşımını destekler:

• Azure İşlevleri tümleştirme paketini kullanarak Azure İşlevleri çalıştırma.
• Kendi hesaplama kaynaklarınızı getirin temel paketi kullanarak.

Aracı barındırma

Aracınızı standart Microsoft Agent Framework desenini kullanarak tanımlayın, ardından Dayanıklı Görev uzantısıyla geliştirin. Uzantı oturum kalıcılığını, uç nokta oluşturmayı ve durum yönetimini otomatik olarak işler.

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()

Uyarı

C# örnekleri, yeni projeler için önerilen desen olan AIProjectClient paketinden Azure.AI.Projects kullanır. Python örnekleri agent_framework.azure'dan FoundryChatClient kullanır. Her iki istemci de Dayanıklı Görev uzantısıyla tümleşen .AsAIAgent() / .as_agent() uzantısını destekler. En son paket referansları için GitHub üzerindeki örneklere bakın.

Kalıcı kontrol noktaları ile çok aracılı orkestrasyon

Birden çok özel aracı dayanıklı bir orkestrasyon içinde adım olarak koordine edebilirsiniz. Her aracı çağrısı bir kontrol noktası oluşturur ve herhangi bir adım başarısız olduğunda orkestrasyon otomatik olarak kendini kurtarır. Tamamlanan aracı çağrıları, kurtarma sırasında yeniden yürütülemez.

Kayıtlı aracıları orkestrasyon içinde almak için context.GetAgent() (C#) veya app.get_agent() (Python) kullanın. Döndürülen DurableAIAgent sarmalayıcı, çağrıların çerçeve tarafından izlenmesini ve denetlenmesini sağlar.

Aşağıdaki örnekte, bir araştırma aracısının bilgi topladığı ve yazar aracısının belge ürettiği sıralı çok aracılı bir iş akışı gösterilmektedir.

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 Framework ile graf tabanlı iş akışları

Dayanıklı Görev uzantısı, yürütücülerin ve aracıların çok adımlı işlem hatlarını tanımlamak için bildirim temelli, graf tabanlı programlama modeli () kullanan WorkflowBuilder. Uzantı, grafikteki her adımı otomatik olarak kontrol eder ve iş akışı tanımında değişiklik yapmadan hatalardan kurtarır.

İş akışları, orkestrasyonları tamamlar: Imperatif ajan koordinasyonu için koşullu mantıkla orkestrasyonları ve tip doğrulamasına sahip sabit grafik topolojileri ile ileti yönlendirmesi için iş akışlarını kullanın.

Sıralı iş akışı

Aşağıdaki örnek, sipariş iptali iş akışına üç işlemciyi zincirler: siparişi bulun, iptal edin ve onay e-postası gönderin.

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 ve SendEmail yürütücüleri dayanıklı kod içermeyen standart Microsoft Agent Framework yürütücüleridir. Tam uygulamalar için GitHub üzerindeki örneklerine bakın.

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 ve SendEmail yürütücüleri dayanıklı kod içermeyen standart Microsoft Agent Framework yürütücüleridir. Tam uygulamalar için GitHub üzerindeki örneklerine bakın.

Fan-out/fan-in (eşzamanlı) çalışma akışı

Birden çok yürütücüye veya aracıya görev dağıtımı yapabilir, ardından sonuçları toplamak için birleştirebilirsiniz. Aşağıdaki örnek, bir fen sorusunu paralel olarak bir fizikçi ve kimyager aracısına gönderir ve ardından yanıtlarını toplar.

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 ve AggregatorExecutor, dayanıklı kod içermeyen standart Microsoft Agent Framework yürütücüleridir. Tam uygulamalar için GitHub üzerindeki örneklerine bakın.

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 ve AggregatorExecutor, dayanıklı kod içermeyen standart Microsoft Agent Framework yürütücüleridir. Tam uygulamalar için GitHub üzerindeki örneklerine bakın.

Koşullu yönlendirme iş akışı

Çalışma zamanı sonuçlarına göre yürütmeyi farklı dallara yönlendirebilirsiniz. Aşağıdaki örnek, gelen e-postayı sınıflandırmak için bir istenmeyen posta algılama aracısı kullanır ve ardından istenmeyen posta işleyicisine veya e-posta yardımcısı aracısına yönlendirir.

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 ve EmailSenderExecutor, dayanıklı kod içermeyen standart Microsoft Agent Framework yürütücüleridir. Tam uygulamalar için GitHub üzerindeki örneklerine bakın.

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 ve EmailSenderExecutor, dayanıklı kod içermeyen standart Microsoft Agent Framework yürütücüleridir. Tam uygulamalar için GitHub üzerindeki örneklerine bakın.

İnsan faktörlü (HITL) iş akışı

Devam etmeden önce dış girişi beklemek için iş akışı yürütmeyi belirlenen noktalarda duraklatabilirsiniz. Microsoft Agent Framework iş akışı modeli, duraklatma noktalarını tanımlamak için RequestPort düğümlerini (.NET) veya ctx.request_info() (Python) kullanır. Aşağıdaki örnek, yönetici onayı ve ardından paralel bütçe ve uyumluluk onayları içeren bir gider geri ödeme iş akışı uygular.

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();

Çerçeve, HITL etkileşimi için üç HTTP uç noktasını otomatik olarak oluşturur.

• POST /api/workflows/{name}/run : İş akışını başlatma
• GET /api/workflows/{name}/status/{id} : Durumu ve bekleyen onayları denetleyin
• POST /api/workflows/{name}/respond/{id} : Devam etmek için onay yanıtı gönder

Aşağıdaki kayıt türleri, iş akışı üzerinden akan verileri tanımlar:

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

CreateApprovalRequest, PrepareFinanceReview ve ExpenseReimburse yürütücüleri dayanıklı kod içermeyen standart Microsoft Agent Framework yürütücüleridir. Tam uygulamalar için GitHub üzerindeki örneklerine bakın.

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;
    }
}

Aşağıdaki kayıt türleri, iş akışı üzerinden akan verileri tanımlar:

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

CreateApprovalRequest, PrepareFinanceReview ve ExpenseReimburse yürütücüleri dayanıklı kod içermeyen standart Microsoft Agent Framework yürütücüleridir. Tam uygulamalar için GitHub üzerindeki örneklerine bakın.

Dayanıklı Görev Zamanlayıcı panosu

Dayanıklı aracılarınıza, düzenlemelerinize ve graf tabanlı iş akışlarınıza tam görünürlük için Dayanıklı Görev Zamanlayıcı panosunu kullanın:

• Her aracı oturumu için konuşma geçmişini görüntüleme
• Araç çağrılarını ve yapılandırılmış çıkışları inceleme
• Düzenleme ve iş akışı yürütme akışlarını izleme
• Performans ölçümlerini izleme

Hem yerel geliştirme (öykünücü aracılığıyla) hem de üretim dağıtımları aynı pano deneyimini ortaya çıkartır.

Aşağıdaki ekran görüntüsü, konuşma geçmişi ve oturum ayrıntılarıyla birlikte bir ajan oturumunu gösterir.

Aşağıdaki ekran görüntüsünde etkinlik yürütme ayrıntılarıyla belirleyici bir düzenleme gösterilmektedir:

Dayanıklı ajanlar için oturum yaşam süresi (TTL)

Dayanıklı aracı oturumları konuşma geçmişini ve durumunu otomatik olarak korur ve bu da süresiz olarak birikebilir. Yaşam süresi (TTL) özelliği, boş oturumların otomatik olarak temizlenmesini sağlayarak depolama kaynağı tüketimini ve artan maliyetleri önler.

Aracı oturumu yapılandırılan TTL süresinden daha uzun süre boşta kaldığında, oturum durumu otomatik olarak silinir. Her yeni etkileşim TTL zamanlayıcısını sıfırlayarak oturumun ömrünü uzatır.

Varsayılan değerler

• Varsayılan TTL: 14 gün
• En düşük TTL silme gecikmesi: 5 dakika

Konfigürasyon

TTL genel olarak veya aracı başına yapılandırılabilir. Bir aracı oturum süresi dolduğunda konuşma geçmişi ve özel durum verileri de dahil olmak üzere tüm durumu silinir. Silme işleminden sonra aynı oturuma bir ileti gönderilirse, yeni bir konuşma geçmişiyle yeni bir oturum oluşturulur.

Uyarı

TTL yapılandırması şu anda yalnızca .NET kullanılabilir.

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);
    });

Bilinen sınırlamalar

• Konuşma boyutu üst sınırı.
  Konuşma geçmişinin tamamı da dahil olmak üzere aracı oturum durumu, dayanıklı arka ucun durum boyutu sınırlarına tabidir. Dayanıklı Görev Zamanlayıcı kullanılırken, varlık durumu boyutu üst sınırı 1 MB'tır. Büyük araç çağrısı yanıtlarıyla uzun süre çalışan konuşmalar bu sınıra ulaşabilir. Konuşma geçmişinin sıkıştırılması, örneğin yeni bir aracı oturumu başlatılarak ve önceki bağlam özetlenerek el ile yapılmalıdır.

• Gecikme.
  Tüm aracı etkileşimleri Dayanıklı Görev Zamanlayıcı aracılığıyla yönlendirilir ve bu da bellek içi aracı yürütmeye kıyasla gecikme süresi ekler. Bu takas dayanıklılık ve dağıtılmış ölçeklendirme sağlar.

• Akış.
  Dayanıklı aracılar dayanıklı varlıkların üzerine uygulandığından, temel alınan iletişim modeli istek/yanıttır. Akış, yanıt geri çağırmaları aracılığıyla desteklenir (örneğin, istemci tüketimi için belirteçleri Redis Stream'e gönderme), varlık ise akış tamamlandıktan sonra yanıtın tamamını döndürür.

• TTL süre sonu.
  TTL zamanlayıcı, kümülatif etkinlik süresini değil, son iletiden bu yana duvar saati süresini temel alır. Bir oturum silindikten sonra (TTL süre sonu veya el ile silme yoluyla), konuşma geçmişi kurtarılamaz.

Ek olanaklar

Aşağıdaki gelişmiş desenler, GitHub üzerindeki örneklerinde bulunur ancak bu makalede ayrıntılı olarak ele alınamaz:

Desen	Açıklama
Uzun süre çalışan araçlar	Aracılar, araç çağrılarından dayanıklı orkestrasyonlar başlatarak, ilerleme takibi ile uzun süre çalışan araçları etkinleştirebilir.
MCP aracı olarak Agent	Dayanıklı aracıları Model Bağlam Protokolü (MCP) araçları olarak kullanıma seçerek diğer aracıların veya istemcilerin bunları standart bir arabirim aracılığıyla çağırmasını sağlar.
Güvenilir akış	İstemci bağlantısının kesilmesini ve yeniden bağlanmasını sürdüren devam ettirilebilir belirteç akışı için Redis Akışlarını (veya benzer aktarımları) kullanın.

İlgili bağlantılar

Tam kod örnekleri için:

• .NET Azure İşlevleri
• .NET Dayanıklı İş Akışları
• Python Azure İşlevleri (aracılar ve iş akışları)

Tam kod örnekleri için:

• .NET herhangi bir platform
• .NET Dayanıklı İş Akışları
• Python çoklu konak (acenteler ve iş akışları)

Sonraki Adımlar

• Yapay zeka aracıları için Dayanıklı Göreve genel bakış
• Aracı uygulama desenleri
• Azure İşlevleri (Dayanıklı)
• Dayanıklı Görev Zamanlayıcı'ya genel bakış