Tartós feladatbővítmény a Microsoft Agent Frameworkhöz

A Microsoft Agent Framework tartó feladatkiterjesztése tartós végrehajtást hoz közvetlenül a Microsoft Agent Framework-be. Az ügynököket a bővítményrel regisztrálva automatikusan tartóssá teheti őket állandó munkamenetekkel, beépített API-végpontokkal és elosztott skálázással – az ügynöklogika módosítása nélkül.

Mikor érdemes tartós ügynököket használni?

Szükség esetén válassza a Durable Task bővítményt:

Állandó beszélgetési állapot – Az ügynök munkamenetei a környezet elvesztése nélkül túlélik a folyamat összeomlását, újraindítását és skálázását.
Többügynöki koordináció – Specializált ügynökök koordinálása determinisztikus munkafolyamatokban, automatikus ellenőrzőpont-készítéssel és hibajavítással.
Hosszú ideig futó munkafolyamatok – Támogatja a cikluson belüli emberi jóváhagyásokat vagy az időzített várakozásokat, amelyek számítási erőforrások felhasználása nélkül tarthatnak órákig, napokig vagy hetekig.
Scalable, kiszolgáló nélküli üzemeltetés – Skálázás több ezer egyidejű ügynöki munkamenetre (vagy nullára) a Azure Functions Flex Consumption csomagban.

Ha nincs szüksége tartós állapotra vagy többügynök-koordinációra, elegendő lehet egy szabványos Microsoft Ügynök-keretrendszer ügynök a bővítmény nélkül.

Jótanács

A helyi beállításokkal, előfeltételekkel és üzembe helyezéssel kapcsolatos részletes útmutatóért tekintse meg a Microsoft Learn tutorial című témakört.

Architecture

A bővítmény belsőleg entitásalapú ügynökhurkokat implementál, ahol minden ügynök-munkamenet egy tartós entitás, amely automatikusan kezeli a beszélgetés állapotát és az ellenőrzőpontozást.

A bővítmény két üzemeltetési módszert támogat:

Azure Functions a Azure Functions integrációs csomag használatával.
Hozza el a saját számítási erőforrásait az alapcsomag használatával.

Ügynök üzemeltetése

Definiálja az ügynököt a standard Microsoft Agent Framework-mintával, majd javítsa ki a Durable Task kiterjesztéssel. A bővítmény automatikusan kezeli a munkamenetek megőrzését, a végpontok létrehozását és az állapotkezelést.

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

Megjegyzés:

A C#-példák AIProjectClient a Azure.AI.Projects csomagból származnak, ami az új projektek ajánlott mintája. A Python példák a FoundryChatClient-t használják agent_framework.azure-ből. Mindkét ügyfél támogatja a .AsAIAgent() / .as_agent() Durable Task kiterjesztéssel integrálható bővítményt. A legújabb csomaghivatkozásokért tekintse meg a mintákat a GitHubon.

Többügynök-vezénylés tartós ellenőrzőpontokkal

A tartós vezénylés lépéseiként több specializált ügynököt is koordinálhat. Minden ügynökhívást ellenőrzőpontokkal látunk el, és az automatizált folyamat automatikusan helyreáll, ha egy lépés kudarcot vall. A befejezett ügynökhívások nem lesznek újra végrehajtva a helyreállítás során.

A context.GetAgent() (C#) vagy app.get_agent() (Python) használatával kérheti le az orchestrationön belüli regisztrált ügynököket. A visszaadott DurableAIAgent burkoló biztosítja a hívások nyomon követését és ellenőrzését a keretrendszerben.

Az alábbi példa egy egymást követő többügynökös munkafolyamatot mutat be, ahol egy kutató ügynök információkat gyűjt, és egy író ügynök létrehoz egy dokumentumot.

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

Gráfalapú munkafolyamatok Microsoft Ügynök-keretrendszerrel

A Durable Task bővítmény támogatja a Microsoft Agent Framework munkafolyamatait is, amelyek deklaratív, gráfalapú programozási modellt (WorkflowBuilder) használnak a végrehajtók és ügynökök többlépéses folyamatainak definiálásához. A bővítmény automatikusan ellenőrzi a gráf minden lépését, és a munkafolyamat-definíció módosítása nélkül helyreállítja a hibákat.

A munkafolyamatok kiegészítik a vezényléseket: vezényléseket használnak az imperatív ügynökök feltételes logikával való összehangolásához, valamint rögzített gráftopológiákhoz tartozó munkafolyamatokat a típus által ellenőrzött üzenet-útválasztással.

Szekvenciális munkafolyamat

Az alábbi példa három végrehajtót láncolt egy megrendelés-lemondási munkafolyamatba: keresse meg a rendelést, mondja le, majd küldjön egy visszaigazoló e-mailt.

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

A OrderLookup, OrderCancel és SendEmail végrehajtók standard Microsoft Agent Framework-végrehajtók tartós kód nélkül. A teljes implementációkért tekintse meg a samples 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>();

A OrderLookup, OrderCancel és SendEmail végrehajtók standard Microsoft Agent Framework-végrehajtók tartós kód nélkül. A teljes implementációkért tekintse meg a samples GitHub.

Ki- és beillő (egyidejű) munkafolyamat

Több végrehajtóra vagy ügynökre szétoszthatja a feladatokat, amelyek párhuzamosan futnak, majd összegyűjtheti az eredményeket. Az alábbi példa egy tudományos kérdést küld egy fizikusnak és vegyészügynöknek párhuzamosan, majd összesíti a válaszaikat.

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

A ParseQuestionExecutor és AggregatorExecutor standard Microsoft Agent Framework-végrehajtók, amelyek nem tartalmaznak Durable-specifikus kódot. A teljes implementációkért tekintse meg a samples 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);
}

Feltételes útválasztási munkafolyamat

Futtatási eredmények alapján különböző ágakhoz irányíthatja a végrehajtást. Az alábbi példa egy levélszemétészlelő ügynököt használ a bejövő e-mailek besorolásához, majd egy levélszemétkezelőhöz vagy egy e-mail-segédügynökhöz irányít.

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

A SpamHandlerExecutor és EmailSenderExecutor standard Microsoft Agent Framework-végrehajtók, amelyek nem tartalmaznak Durable-specifikus kódot. A teljes implementációkért tekintse meg a samples 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>();

Human-in-the-loop (HITL) munkafolyamat

A munkafolyamat-végrehajtást a kijelölt pontokon szüneteltetheti, hogy várjon a külső bemenetre a folytatás előtt. Az Microsoft Agent Framework munkafolyamat-modellje RequestPort csomópontokat (.NET) vagy ctx.request_info() (Python) használ a szüneteltetési pontok meghatározásához. Az alábbi példa egy költségtérítési munkafolyamatot implementál egy felettes jóváhagyásával, majd párhuzamos költségvetést és megfelelőségi jóváhagyásokat.

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

A keretrendszer automatikusan létrehoz három HTTP-végpontot a HITL-interakcióhoz.

POST /api/workflows/{name}/run : A munkafolyamat indítása
GET /api/workflows/{name}/status/{id} : Állapot és függőben lévő jóváhagyások ellenőrzése
POST /api/workflows/{name}/respond/{id} : Jóváhagyási válasz küldése a folytatáshoz

A munkafolyamaton áthaladó adatokat a következő rekordtípusok határozzák meg:

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

A CreateApprovalRequest, PrepareFinanceReview és ExpenseReimburse végrehajtók standard Microsoft Agent Framework-végrehajtók tartós kód nélkül. A teljes implementációkért tekintse meg a samples 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;
    }
}

A munkafolyamaton áthaladó adatokat a következő rekordtípusok határozzák meg:

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

Durable Task Scheduler irányítópult

A Durable Task Scheduler irányítópult használatával teljes körűen áttekintheti a tartós ügynököket, vezényléseket és gráfalapú munkafolyamatokat:

Az egyes ügynök-munkamenetek beszélgetési előzményeinek megtekintése
Eszközhívások és strukturált kimenetek vizsgálata
Vezénylési és munkafolyamat-végrehajtási folyamatok nyomon követése
Teljesítménymetrikák monitorozása

A helyi fejlesztés (az emulátoron keresztül) és az éles üzembe helyezés egyaránt ugyanazt az irányítópultot használja.

Az alábbi képernyőképen egy ügynök-munkamenet látható a beszélgetési előzményekkel és a munkamenet részleteivel:

Az alábbi képernyőképen egy determinisztikus vezénylés látható a tevékenységek végrehajtásának részleteivel:

Munkamenet élettartam (TTL) tartós ügynökökhöz

A tartós ügynök-munkamenetek automatikusan megőrzik a beszélgetések előzményeit és állapotát, ami határozatlan ideig halmozódhat fel. Az élettartam (TTL) funkció automatikusan törli az inaktív munkameneteket, így megakadályozza a tárerőforrás-felhasználást és a megnövekedett költségeket.

Ha egy ügynök munkamenete hosszabb ideig tétlen, mint a konfigurált TTL-időszak, a munkamenet állapota automatikusan törlődik. Minden új interakció alaphelyzetbe állítja a TTL időzítőt, meghosszabbítva a munkamenet élettartamát.

Alapértelmezett értékek

Alapértelmezett TTL: 14 nap
Minimális TTL-törlési késleltetés: 5 perc

Konfiguráció

A TTL globálisan vagy ügynökönként konfigurálható. Amikor egy ügynök munkamenete lejár, a teljes állapota törlődik, beleértve a beszélgetési előzményeket és az egyéni állapotadatokat is. Ha a törlés után egy üzenet érkezik ugyanahhoz a munkamenethez, a rendszer új munkamenetet hoz létre egy friss beszélgetési előzményekkel.

Megjegyzés:

A TTL-konfiguráció jelenleg csak .NET érhető el.

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

Ismert korlátozások

Beszélgetés maximális mérete.
Az ügynöki munkamenet állapotára, beleértve a teljes beszélgetési előzményt, a tartós háttérrendszer állapotméret korlátai vonatkoznak. A Durable Task Scheduler használatakor az entitásállapot maximális mérete 1 MB. Azok a beszélgetések, amelyek hosszú ideig futnak nagy eszközhívási válaszokkal, elérhetik ezt a korlátot. A beszélgetési előzmények tömörítését manuálisan kell elvégezni, például egy új ügynök-munkamenet elindításával és a korábbi környezet összegzésével.
Késleltetés.
Az ügynök minden interakciója a Durable Task Scheduler-en keresztül van irányítva, ami nagyobb késleltetést okoz az ügynök közvetlen memória-alapú végrehajtásával összehasonlítva. Ez a kompromisszum tartósságot és elosztott skálázást biztosít.
Streaming.
Mivel a tartós ügynökök a tartós entitásokon felül vannak implementálva, a mögöttes kommunikációs modell a kérés/válasz. A streamelést válaszvisszahívások támogatják (például tokenek leküldése egy Redis Streambe ügyfél-felhasználásra), míg az entitás a teljes választ adja vissza a stream befejezése után.
TTL-lejárat.
A TTL időzítő az utolsó üzenet óta eltelt valós idő alapján kerül meghatározásra, nem pedig az összesített tevékenységi idő alapján. A munkamenet törlése után (TTL-lejárat vagy manuális törlés) a beszélgetési előzmények nem állíthatók helyre.

További képességek

GitHub mintatárai között érhetők el a következő fejlett minták, de a jelen cikkben nem foglalkozunk velük részletesen.

Minta	Leírás
Hosszú ideig futó eszközök	Az ügynökök tartós vezényléseket indíthatnak el az eszközhívásokból, így a hosszabb ideig futó eszközök állapotkövetéssel is használhatók.
Agent MCP-eszközként	A tartós ügynököket modellkörnyezeti protokoll (MCP) eszközökként teheti elérhetővé, így más ügynökök vagy ügyfelek szabványos felületen hívhatják meg őket.
Megbízható streamelés	Redis Streams (vagy hasonló átvitelek) használatával folytatható token streamelés valósítható meg, amely túléli az ügyfél kapcsolatbontását és újracsatlakozását.

Teljes kódminták esetén:

Következő lépések

Visszajelzés

Hasznosnak találta ezt az oldalt?

Last updated on 2026-05-06