Ekstensi Tugas Tahan Lama untuk Microsoft Agent Framework

Ekstensi Tugas Durable untuk Microsoft Agent Framework membawa eksekusi yang tahan lama langsung ke dalam Microsoft Agent Framework. Anda dapat mendaftarkan agen dengan ekstensi untuk membuatnya secara otomatis tahan lama dengan sesi persisten, titik akhir API bawaan, dan penskalaan terdistribusi — tanpa perubahan pada logika agen Anda.

Kapan menggunakan agen tahan lama

Pilih ekstensi Tugas Tahan Lama saat Anda membutuhkan:

Status percakapan persisten — Sesi agen bertahan dari kerusakan proses, restart, dan peristiwa penskalaan tanpa kehilangan konteks.
Orkestrasi multi-agen — Koordinasikan agen khusus dalam alur kerja deterministik dengan titik pemeriksaan otomatis dan pemulihan kegagalan.
Alur kerja yang berjalan lama — Mendukung persetujuan yang melibatkan manusia atau penundaan yang terjadwal yang dapat berlangsung berjam-jam, berhari-hari, atau berminggu-minggu tanpa mengonsumsi sumber daya komputasi.
Skalabilitas, hosting tanpa server — Memperluas ke ribuan sesi agen bersamaan (atau ke nol) pada paket Flex Consumption Azure Functions.

Jika Anda tidak memerlukan koordinasi status tahan lama atau multi-agen, agen Microsoft Agent Framework standar tanpa ekstensi mungkin cukup.

Petunjuk / Saran

Untuk panduan langkah demi langkah termasuk penyiapan, prasyarat, dan penyebaran lokal, lihat tutorial di Microsoft Learn.

Architecture

Ekstensi ini secara internal mengimplementasikan perulangan agen berbasis entitas, di mana setiap sesi agen adalah entitas tahan lama yang secara otomatis mengelola status percakapan dan titik pemeriksaan.

Ekstensi ini mendukung dua pendekatan hosting:

Azure Functions menggunakan paket integrasi Azure Functions.
Bawa perangkat komputasi Anda sendiri menggunakan paket dasar.

Hosting agen

Tentukan agen Anda menggunakan pola Microsoft Agent Framework standar, lalu tingkatkan dengan ekstensi Tugas Tahan Lama. Ekstensi menangani persistensi sesi, pembuatan titik akhir, dan manajemen status secara otomatis.

C#
Python

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

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#
Python

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

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

Nota

Contoh C# menggunakan AIProjectClient dari paket Azure.AI.Projects, yang merupakan pola yang direkomendasikan untuk proyek baru. Contoh Python menggunakan FoundryChatClient dari agent_framework.azure. Kedua klien mendukung ekstensi .AsAIAgent() / .as_agent() yang diintegrasikan dengan ekstensi Tugas Tahan Lama. Lihat sampel pada GitHub untuk referensi paket terbaru.

Orkestrasi multi-agen dengan titik pemeriksaan tahan lama

Anda dapat mengoordinasikan beberapa agen khusus sebagai langkah-langkah dalam orkestrasi yang andal. Setiap panggilan agen di-checkpoint, dan orkestrasi pulih secara otomatis jika ada langkah yang gagal. Panggilan agen yang selesai tidak dieksekusi ulang pada proses pemulihan.

Gunakan context.GetAgent() (C#) atau app.get_agent() (Python) untuk mengambil agen terdaftar di dalam orkestrasi. Pembungkus yang dikembalikan DurableAIAgent memastikan panggilan dilacak dan diperiksa oleh kerangka kerja.

Contoh berikut menunjukkan alur kerja multi-agen berurutan di mana agen penelitian mengumpulkan informasi dan agen penulis menghasilkan dokumen.

C#
Python

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

@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#
Python

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

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

Alur kerja berbasis grafik dengan kerangka kerja agen Microsoft

Ekstensi Durable Task juga mendukung alur kerja Microsoft Agent Framework, yang menggunakan model pemrograman deklaratif berbasis grafik () untuk menentukan alur multi-langkah pelaksana dan agen. Ekstensi secara otomatis memeriksa setiap langkah dalam grafik dan pulih dari kegagalan tanpa perubahan pada definisi alur kerja.

Alur kerja melengkapi orkestrasi: gunakan orkestrasi untuk koordinasi agen imperatif dengan logika kondisional, dan alur kerja untuk topologi grafik tetap dengan perutean pesan yang divalidasi jenis.

Alur kerja berurutan

Contoh berikut menautkan tiga pelaksana ke dalam alur kerja pembatalan pesanan: mencari pesanan, membatalkannya, lalu mengirim email konfirmasi.

C#
Python

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

Pelaksana OrderLookup, OrderCancel, dan SendEmail adalah eksekutor standar dari Microsoft Agent Framework tanpa kode khusus untuk Durable. Untuk implementasi lengkap, lihat sampel pada GitHub.

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

Pelaksana OrderLookup, OrderCancel, dan SendEmail adalah eksekutor standar dari Microsoft Agent Framework tanpa kode khusus untuk Durable. Untuk implementasi lengkap, lihat sampel pada GitHub.

Alur kerja fan-out/fan-in (bersamaan)

Anda dapat menyebar ke beberapa eksekutor atau agen yang berjalan secara paralel, lalu menghimpun untuk mengagregasi hasilnya. Contoh berikut mengirimkan pertanyaan sains ke agen fisikawan dan kimia secara paralel, lalu menggabungkan respons mereka.

C#
Python

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 dan AggregatorExecutor adalah eksekutor Microsoft Agent Framework standar tanpa kode khusus Durable. Untuk implementasi lengkap, lihat sampel pada GitHub.

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 dan AggregatorExecutor adalah eksekutor Microsoft Agent Framework standar tanpa kode khusus Durable. Untuk implementasi lengkap, lihat sampel pada GitHub.

Alur kerja perutean bersyarat

Anda dapat merutekan eksekusi ke cabang yang berbeda berdasarkan hasil runtime. Contoh berikut menggunakan agen deteksi spam untuk mengklasifikasikan email masuk, lalu merutekan ke penangan spam atau agen asisten email.

C#
Python

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 dan EmailSenderExecutor adalah eksekutor Microsoft Agent Framework standar tanpa kode khusus Durable. Untuk implementasi lengkap, lihat sampel pada GitHub.

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 dan EmailSenderExecutor adalah eksekutor Microsoft Agent Framework standar tanpa kode khusus Durable. Untuk implementasi lengkap, lihat sampel pada GitHub.

Alur kerja human-in-the-loop (HITL)

Anda dapat menjeda eksekusi alur kerja pada titik yang ditentukan untuk menunggu input eksternal sebelum melanjutkan. Model alur kerja Microsoft Agent Framework menggunakan simpul RequestPort (dalam .NET) atau ctx.request_info() (dalam Python) untuk menentukan titik jeda. Contoh berikut menerapkan alur kerja penggantian biaya dengan persetujuan manajer diikuti oleh persetujuan anggaran dan kepatuhan paralel.

C#
Python

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

Kerangka kerja ini secara otomatis menghasilkan tiga titik akhir HTTP untuk interaksi HITL.

POST /api/workflows/{name}/run : Memulai alur kerja
GET /api/workflows/{name}/status/{id} : Periksa status dan persetujuan yang tertunda
POST /api/workflows/{name}/respond/{id} : Kirim respons persetujuan untuk dilanjutkan

Jenis catatan berikut menentukan data yang mengalir melalui alur kerja:

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

Pelaksana CreateApprovalRequest, PrepareFinanceReview, dan ExpenseReimburse adalah eksekutor standar dari Microsoft Agent Framework tanpa kode khusus untuk Durable. Untuk implementasi lengkap, lihat sampel pada GitHub.

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

Jenis catatan berikut menentukan data yang mengalir melalui alur kerja:

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

Dasbor Penjadwal Tugas Berkelanjutan

Gunakan dasbor Durable Task Scheduler untuk visibilitas penuh ke agen, orkestrasi, dan alur kerja berbasis grafik Anda yang terjamin daya tahannya:

Menampilkan riwayat percakapan untuk setiap sesi agen
Periksa panggilan alat dan keluaran terstruktur
Melacak orkestrasi dan alur eksekusi kerja
Pemantauan metrik performa

Pengembangan lokal (melalui emulator) dan penyebaran produksi menampilkan pengalaman dasbor yang sama.

Cuplikan layar berikut menunjukkan sesi agen dengan riwayat percakapan dan detail sesinya:

Cuplikan layar berikut menunjukkan orkestrasi deterministik dengan detail eksekusi aktivitas:

Waktu hidup sesi (TTL) untuk agen tahan lama

Sesi agen persisten secara otomatis mempertahankan riwayat dan status percakapan, yang dapat terus bertambah tanpa batas. Fitur time-to-live (TTL) menyediakan pembersihan otomatis sesi diam, mencegah konsumsi sumber daya penyimpanan dan peningkatan biaya.

Ketika sesi agen menganggur lebih lama dari periode TTL yang dikonfigurasi, status sesi akan dihapus secara otomatis. Setiap interaksi baru mengatur ulang timer TTL, memperpanjang masa pakai sesi.

Nilai baku

TTL default: 14 hari
Penundaan penghapusan TTL minimum: 5 menit

Konfigurasi

TTL dapat dikonfigurasi secara global atau per agen. Saat sesi agen kedaluwarsa, seluruh statusnya dihapus, termasuk riwayat percakapan dan data status kustom apa pun. Jika pesan dikirim ke sesi yang sama setelah penghapusan, sesi baru dibuat dengan riwayat percakapan baru.

Nota

Konfigurasi TTL saat ini hanya tersedia di .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);
    });

Batasan yang diketahui

Ukuran percakapan maksimum.
Status sesi agen, termasuk riwayat percakapan lengkap, tunduk pada batas ukuran status dari backend persisten. Saat menggunakan Durable Task Scheduler, ukuran status entitas maksimum adalah 1 MB. Percakapan yang berlangsung lama dengan respons panggilan alat yang besar dapat mencapai batas ini. Pemadatan riwayat percakapan harus dilakukan secara manual, misalnya, dengan memulai sesi agen baru dan meringkas konteks sebelumnya.
Latency.
Semua interaksi agen dirutekan melalui Durable Task Scheduler, yang menyebabkan penambahan latensi dibandingkan dengan eksekusi agen dalam memori. Pertukaran ini memberikan ketahanan dan skala yang terdistribusi.
Streaming.
Karena agen tahan lama diimplementasikan di atas entitas tahan lama, model komunikasi yang mendasar adalah permintaan/respons. Streaming didukung melalui panggilan balik respons (contohnya, mengirimkan token ke Redis Stream untuk konsumsi klien), sedangkan entitas mengembalikan respons lengkap setelah streaming selesai.
Kedaluwarsa TTL.
Timer TTL didasarkan pada waktu jam dinding sejak pesan terakhir, bukan waktu aktivitas kumulatif. Setelah sesi dihapus (melalui kedaluwarsa TTL atau penghapusan manual), riwayat percakapannya tidak dapat dipulihkan.

Kemampuan tambahan

Pola tingkat lanjut berikut tersedia dalam sampel pada GitHub tetapi tidak dibahas secara rinci dalam artikel ini:

Pola	Deskripsi
Alat yang berjalan lama	Agen dapat memulai orkestrasi yang tahan lama melalui panggilan alat, memungkinkan alat yang beroperasi dalam jangka waktu lama dengan pelacakan kemajuan.
Agen sebagai alat MCP	Mengekspos agen tahan lama sebagai alat Protokol Konteks Model (MCP), memungkinkan agen atau klien lain untuk memanggilnya melalui antarmuka standar.
Streaming yang dapat diandalkan	Gunakan Redis Streams (atau media pengiriman serupa) untuk streaming token yang bersifat dapat dilanjutkan yang bertahan dari pemutusan dan koneksi ulang klien.

Untuk sampel kode lengkap:

Langkah berikutnya

Saran dan Komentar

Apakah halaman ini membantu?

Last updated on 2026-05-06