지속성 확장

.NET 프로젝트에서 호스팅 모델에 대한 패키지 집합을 선택합니다.

Azure Functions 호스팅의 경우 Azure Functions 통합 패키지 및 Functions 작업자 패키지를 추가합니다.

dotnet add package Azure.AI.Projects --prerelease
dotnet add package Azure.Identity
dotnet add package Microsoft.Agents.AI.Foundry --prerelease
dotnet add package Microsoft.Agents.AI.Hosting.AzureFunctions --prerelease

bring-your-own-compute 호스팅의 경우 기본 지속성 작업 통합 패키지와 호스트에서 사용하는 지속성 작업 스케줄러 작업자/클라이언트 패키지를 추가합니다.

dotnet add package Microsoft.Agents.AI.DurableTask --prerelease
dotnet add package Microsoft.DurableTask.Client.AzureManaged
dotnet add package Microsoft.DurableTask.Worker.AzureManaged
dotnet add package Microsoft.Extensions.Hosting

Python 프로젝트에서 호스팅 모델에 대한 패키지를 선택합니다.

Azure Functions 호스팅의 경우 Azure Functions 통합 패키지를 설치합니다.

pip install azure-identity
pip install agent-framework-azurefunctions --pre

Bring-your-own-compute 호스팅의 경우 지속성 작업 통합 패키지를 설치합니다.

pip install azure-identity
pip install agent-framework-durabletask --pre

using System;
using Azure.AI.Projects;
using Azure.Identity;
using Microsoft.Agents.AI;
using Microsoft.Agents.AI.Hosting.AzureFunctions;
using Microsoft.Azure.Functions.Worker.Builder;
using Microsoft.Extensions.Hosting;

var endpoint = Environment.GetEnvironmentVariable("AZURE_OPENAI_ENDPOINT");
var deploymentName = Environment.GetEnvironmentVariable("AZURE_OPENAI_DEPLOYMENT") ?? "gpt-4o-mini";

// Create an AI agent following the standard Microsoft Agent Framework pattern
AIAgent agent = new AIProjectClient(new Uri(endpoint), new DefaultAzureCredential())
    .AsAIAgent(
        model: deploymentName,
        instructions: "You are good at telling jokes.",
        name: "Joker");

// Configure the function app to host the agent with durable thread management
// This automatically creates HTTP endpoints and manages state persistence
using IHost app = FunctionsApplication
    .CreateBuilder(args)
    .ConfigureFunctionsWebApplication()
    .ConfigureDurableAgents(options =>
        options.AddAIAgent(agent)
    )
    .Build();
app.Run();

import os
from agent_framework.azure import AgentFunctionApp
from agent_framework.openai import OpenAIChatCompletionClient
from azure.identity import DefaultAzureCredential

endpoint = os.getenv("AZURE_OPENAI_ENDPOINT")
deployment_name = os.getenv("AZURE_OPENAI_CHAT_COMPLETION_MODEL", "gpt-4o-mini")
api_version = os.getenv("AZURE_OPENAI_API_VERSION")

# Create an AI agent following the standard Microsoft Agent Framework pattern
agent = OpenAIChatCompletionClient(
    azure_endpoint=endpoint,
    model=deployment_name,
    api_version=api_version,
    credential=DefaultAzureCredential()
).as_agent(
    instructions="You are good at telling jokes.",
    name="Joker"
)

# Configure the function app to host the agent with durable thread management
# This automatically creates HTTP endpoints and manages state persistence
app = AgentFunctionApp(agents=[agent])

기본 지속성 작업 통합 패키지를 사용하여 호스트를 구성합니다. 지속성 에이전트에는 ConfigureDurableAgents, 그래프 기반 Microsoft Agent Framework 워크플로에는 ConfigureDurableWorkflows 사용합니다.

string connectionString = Environment.GetEnvironmentVariable("DURABLE_TASK_SCHEDULER_CONNECTION_STRING")
    ?? "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();

실행 가능한 자체 호스팅 예제는 .NET 지속성 에이전트 콘솔 샘플 및 .NET 지속성 워크플로 콘솔 샘플을 참조하세요.

지속성 작업 통합 패키지를 사용하여 에이전트를 등록하고 요청을 수신 대기하는 작업자 프로세스를 실행합니다. 클라이언트 코드는 다른 프로세스에서 동일한 지속성 작업 스케줄러 작업 허브에 연결할 수 있습니다.

from agent_framework.azure import DurableAIAgentWorker
from durabletask.azuremanaged.worker import DurableTaskSchedulerWorker

worker = DurableTaskSchedulerWorker(
    host_address="http://localhost:8080",
    secure_channel=False,
    taskhub="default",
)

agent_worker = DurableAIAgentWorker(worker)
agent_worker.add_agent(agent)

worker.start()

단일 에이전트 호스팅, 다중 에이전트 라우팅, 신뢰할 수 있는 스트리밍, 오케스트레이션 체인, 동시성, 조건부 및 휴먼 인 더 루프 패턴을 포함한 작업자 클라이언트 예제는 Python 지속성 작업 샘플을 참조하세요.

자체 호스팅 앱에 대한 ConfigureDurableWorkflows 또는 Azure Functions 호스팅을 위해 Functions 앱 작성기에서 ConfigureDurableWorkflows 그래프 기반 워크플로를 등록합니다.

.NET 지속성 워크플로 Azure Functions 샘플 및 .NET 지속성 워크플로 콘솔 샘플 참조하세요.

지속성 워크플로 샘플은 공유 상태, 공유 상태 없음, 병렬 워크플로 실행 및 휴먼 인 더 루프 워크플로를 포함하여 Azure Functions 호스팅에 사용할 수 있습니다.

지속성 에이전트, 오케스트레이션, MCP 서버 및 워크플로 예제는 Python Azure Functions 샘플을 참조하세요.

오케스트레이션에서 에이전트를 사용할 때는 context.GetAgent() API를 사용하여 DurableAIAgent 인스턴스를 얻어야 합니다. 이 인스턴스는 등록된 에이전트 중 하나를 래핑하는 표준 AIAgent 타입의 특수 하위 클래스입니다. 래퍼는 DurableAIAgent 에이전트 호출이 내구성 오케스트레이션 프레임워크에 의해 올바르게 추적되고 검사점으로 기록되도록 합니다.

using Microsoft.Azure.Functions.Worker;
using Microsoft.DurableTask;
using Microsoft.Agents.AI.DurableTask;

[Function(nameof(SpamDetectionOrchestration))]
public static async Task<string> SpamDetectionOrchestration(
    [OrchestrationTrigger] TaskOrchestrationContext context)
{
    Email email = context.GetInput<Email>();

    // Check if the email is spam
    DurableAIAgent spamDetectionAgent = context.GetAgent("SpamDetectionAgent");
    AgentSession spamSession = await spamDetectionAgent.CreateSessionAsync();

    AgentResponse<DetectionResult> spamDetectionResponse = await spamDetectionAgent.RunAsync<DetectionResult>(
        message: $"Analyze this email for spam: {email.EmailContent}",
        session: spamSession);
    DetectionResult result = spamDetectionResponse.Result;

    if (result.IsSpam)
    {
        return await context.CallActivityAsync<string>(nameof(HandleSpamEmail), result.Reason);
    }

    // Generate response for legitimate email
    DurableAIAgent emailAssistantAgent = context.GetAgent("EmailAssistantAgent");
    AgentSession emailSession = await emailAssistantAgent.CreateSessionAsync();

    AgentResponse<EmailResponse> emailAssistantResponse = await emailAssistantAgent.RunAsync<EmailResponse>(
        message: $"Draft a professional response to: {email.EmailContent}",
        session: emailSession);

    return await context.CallActivityAsync<string>(nameof(SendEmail), emailAssistantResponse.Result.Response);
}

오케스트레이션에서 에이전트를 사용할 때는, 등록된 에이전트 중 하나를 감싸는 특수 래퍼인 지속 가능한 에이전트 인스턴스를 얻기 위해 app.get_agent() 메서드를 반드시 사용해야 합니다. 지속성 에이전트 래퍼는 에이전트 호출이 지속성 오케스트레이션 프레임워크에 의해 올바르게 추적되고 검사점이 설정되도록 보장합니다.

import azure.durable_functions as df
from typing import cast
from agent_framework.azure import AgentFunctionApp
from pydantic import BaseModel

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

class EmailResponse(BaseModel):
    response: str

app = AgentFunctionApp(agents=[spam_detection_agent, email_assistant_agent])

@app.orchestration_trigger(context_name="context")
def spam_detection_orchestration(context: df.DurableOrchestrationContext):
    email = context.get_input()

    # Check if the email is spam
    spam_agent = app.get_agent(context, "SpamDetectionAgent")
    spam_thread = spam_agent.create_session()

    spam_result_raw = yield spam_agent.run(
        messages=f"Analyze this email for spam: {email['content']}",
        session=spam_thread,
        response_format=SpamDetectionResult
    )
    spam_result = cast(SpamDetectionResult, spam_result_raw.get("structured_response"))

    if spam_result.is_spam:
        result = yield context.call_activity("handle_spam_email", spam_result.reason)
        return result

    # Generate response for legitimate email
    email_agent = app.get_agent(context, "EmailAssistantAgent")
    email_thread = email_agent.create_session()

    email_response_raw = yield email_agent.run(
        messages=f"Draft a professional response to: {email['content']}",
        session=email_thread,
        response_format=EmailResponse
    )
    email_response = cast(EmailResponse, email_response_raw.get("structured_response"))

    result = yield context.call_activity("send_email", email_response.response)
    return result

using Microsoft.Azure.Functions.Worker;
using Microsoft.DurableTask;
using Microsoft.Agents.AI.DurableTask;

[Function(nameof(ResearchOrchestration))]
public static async Task<string> ResearchOrchestration(
    [OrchestrationTrigger] TaskOrchestrationContext context)
{
    string topic = context.GetInput<string>();

    // Execute multiple research agents in parallel
    DurableAIAgent technicalAgent = context.GetAgent("TechnicalResearchAgent");
    DurableAIAgent marketAgent = context.GetAgent("MarketResearchAgent");
    DurableAIAgent competitorAgent = context.GetAgent("CompetitorResearchAgent");

    // Start all agent runs concurrently
    Task<AgentResponse<TextResponse>> technicalTask = 
        technicalAgent.RunAsync<TextResponse>($"Research technical aspects of {topic}");
    Task<AgentResponse<TextResponse>> marketTask = 
        marketAgent.RunAsync<TextResponse>($"Research market trends for {topic}");
    Task<AgentResponse<TextResponse>> competitorTask = 
        competitorAgent.RunAsync<TextResponse>($"Research competitors in {topic}");

    // Wait for all tasks to complete
    await Task.WhenAll(technicalTask, marketTask, competitorTask);

    // Aggregate results
    string allResearch = string.Join("\n\n", 
        technicalTask.Result.Result.Text,
        marketTask.Result.Result.Text,
        competitorTask.Result.Result.Text);

    DurableAIAgent summaryAgent = context.GetAgent("SummaryAgent");
    AgentResponse<TextResponse> summaryResponse = 
        await summaryAgent.RunAsync<TextResponse>($"Summarize this research:\n{allResearch}");

    return summaryResponse.Result.Text;
}

import azure.durable_functions as df
from agent_framework.azure import AgentFunctionApp

app = AgentFunctionApp(agents=[technical_agent, market_agent, competitor_agent, summary_agent])

@app.orchestration_trigger(context_name="context")
def research_orchestration(context: df.DurableOrchestrationContext):
    topic = context.get_input()

    # Execute multiple research agents in parallel
    technical_agent = app.get_agent(context, "TechnicalResearchAgent")
    market_agent = app.get_agent(context, "MarketResearchAgent")
    competitor_agent = app.get_agent(context, "CompetitorResearchAgent")

    technical_task = technical_agent.run(messages=f"Research technical aspects of {topic}")
    market_task = market_agent.run(messages=f"Research market trends for {topic}")
    competitor_task = competitor_agent.run(messages=f"Research competitors in {topic}")

    # Wait for all tasks to complete
    results = yield context.task_all([technical_task, market_task, competitor_task])

    # Aggregate results
    all_research = "\n\n".join([r.get('response', '') for r in results])

    summary_agent = app.get_agent(context, "SummaryAgent")
    summary = yield summary_agent.run(messages=f"Summarize this research:\n{all_research}")

    return summary.get('response', '')

using Microsoft.Azure.Functions.Worker;
using Microsoft.DurableTask;
using Microsoft.Agents.AI.DurableTask;

[Function(nameof(ContentApprovalWorkflow))]
public static async Task<string> ContentApprovalWorkflow(
    [OrchestrationTrigger] TaskOrchestrationContext context)
{
    string topic = context.GetInput<string>();

    // Generate content using an agent
    DurableAIAgent contentAgent = context.GetAgent("ContentGenerationAgent");
    AgentResponse<GeneratedContent> contentResponse = 
        await contentAgent.RunAsync<GeneratedContent>($"Write an article about {topic}");
    GeneratedContent draftContent = contentResponse.Result;

    // Send for human review
    await context.CallActivityAsync(nameof(NotifyReviewer), draftContent);

    // Wait for approval with timeout
    HumanApprovalResponse approvalResponse;
    try
    {
        approvalResponse = await context.WaitForExternalEvent<HumanApprovalResponse>(
            eventName: "ApprovalDecision",
            timeout: TimeSpan.FromHours(24));
    }
    catch (OperationCanceledException)
    {
        // Timeout occurred - escalate for review
        return await context.CallActivityAsync<string>(nameof(EscalateForReview), draftContent);
    }

    if (approvalResponse.Approved)
    {
        return await context.CallActivityAsync<string>(nameof(PublishContent), draftContent);
    }

    return "Content rejected";
}

import azure.durable_functions as df
from datetime import timedelta
from agent_framework.azure import AgentFunctionApp

app = AgentFunctionApp(agents=[content_agent])

@app.orchestration_trigger(context_name="context")
def content_approval_workflow(context: df.DurableOrchestrationContext):
    topic = context.get_input()

    # Generate content using an agent
    content_agent = app.get_agent(context, "ContentGenerationAgent")
    draft_content = yield content_agent.run(
        messages=f"Write an article about {topic}"
    )

    # Send for human review
    yield context.call_activity("notify_reviewer", draft_content)

    # Wait for approval with timeout
    approval_task = context.wait_for_external_event("ApprovalDecision")
    timeout_task = context.create_timer(
        context.current_utc_datetime + timedelta(hours=24)
    )

    winner = yield context.task_any([approval_task, timeout_task])

    if winner == approval_task:
        timeout_task.cancel()
        approval_data = approval_task.result
        if approval_data.get("approved"):
            result = yield context.call_activity("publish_content", draft_content)
            return result
        return "Content rejected"

    # Timeout occurred - escalate for review
    result = yield context.call_activity("escalate_for_review", draft_content)
    return result

await client.RaiseEventAsync(instanceId, "ApprovalDecision", new HumanApprovalResponse 
{ 
    Approved = true,
    Feedback = "Looks great!"
});

approval_data = {
    "approved": True,
    "feedback": "Looks great!"
}
await client.raise_event(instance_id, "ApprovalDecision", approval_data)

• .NET 9.0 SDK 이상
• Azure Functions Core Tools v4.x
• Azure 개발자 명령줄 도구(azd)
• Azure CLI 설치 및 인증
• Docker Desktop 설치 및 실행(Azurite 및 지속성 작업 스케줄러 에뮬레이터를 사용한 로컬 개발용)
• 리소스를 만들 수 있는 권한이 있는 Azure 구독

• Python 3.10 이상
• Azure Functions Core Tools v4.x
• Azure 개발자 명령줄 도구(azd)
• Azure CLI 설치 및 인증
• Docker Desktop 설치 및 실행(Azurite 및 지속성 작업 스케줄러 에뮬레이터를 사용한 로컬 개발용)
• 리소스를 만들 수 있는 권한이 있는 Azure 구독

1. 프로젝트에 대한 새 디렉터리를 만들고 프로젝트로 이동합니다.

   • Bash
   • PowerShell

   mkdir MyDurableAgent
   cd MyDurableAgent
   

1. 템플릿에서 프로젝트를 초기화합니다.

   azd init --template durable-agents-quickstart-dotnet
   

   환경 이름을 묻는 메시지가 표시되면 다음과 같은 my-durable-agent이름을 입력합니다.

Azure Functions 구성, 에이전트 코드 및 인프라를 코드 템플릿으로 포함하여 필요한 모든 파일이 포함된 빠른 시작 프로젝트를 다운로드합니다.

1. 프로젝트에 대한 새 디렉터리를 만들고 프로젝트로 이동합니다.

   • Bash
   • PowerShell

   mkdir MyDurableAgent
   cd MyDurableAgent
   

1. 템플릿에서 프로젝트를 초기화합니다.

   azd init --template durable-agents-quickstart-python
   

   환경 이름을 묻는 메시지가 표시되면 다음과 같은 my-durable-agent이름을 입력합니다.

2. 가상 환경을 만들고 활성화합니다.

   • Bash
   • PowerShell

   uv venv .venv
   source .venv/bin/activate
   

1. 필요한 패키지를 설치합니다.

   python -m pip install -r requirements.txt
   

Azure Functions 구성, 에이전트 코드 및 인프라를 코드 템플릿으로 포함하여 필요한 모든 파일이 포함된 빠른 시작 프로젝트를 다운로드합니다. 또한 필요한 종속성을 사용하여 가상 환경을 준비합니다.

에이전트 구성을 보려면 다음을 엽니다 Program.cs .

using Azure.AI.Projects;
using Azure.Identity;
using Microsoft.Agents.AI;
using Microsoft.Agents.AI.Hosting.AzureFunctions;
using Microsoft.Azure.Functions.Worker.Builder;
using Microsoft.Extensions.Hosting;

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

// Create an AI agent following the standard Microsoft Agent Framework pattern
AIAgent agent = new AIProjectClient(new Uri(endpoint), new DefaultAzureCredential())
    .AsAIAgent(
        model: deploymentName,
        instructions: "You are a helpful assistant that can answer questions and provide information.",
        name: "MyDurableAgent");

using IHost app = FunctionsApplication
    .CreateBuilder(args)
    .ConfigureFunctionsWebApplication()
    .ConfigureDurableAgents(options => options.AddAIAgent(agent))
    .Build();
app.Run();

이 코드:

1. 환경 변수에서 Azure OpenAI 구성을 검색합니다.
2. Azure 자격 증명을 사용하여 Azure OpenAI 클라이언트를 만듭니다.
3. 지침 및 이름을 사용하여 AI 에이전트를 만듭니다.
4. 지속성 스레드 관리를 사용하여 에이전트를 호스트하도록 Azure Functions 앱을 구성합니다.

에이전트 구성을 보려면 다음을 엽니다 function_app.py .

import os
from agent_framework.azure import AgentFunctionApp
from agent_framework.openai import OpenAIChatCompletionClient
from azure.identity import DefaultAzureCredential

endpoint = os.getenv("AZURE_OPENAI_ENDPOINT")
if not endpoint:
    raise ValueError("AZURE_OPENAI_ENDPOINT is not set.")
deployment_name = os.getenv("AZURE_OPENAI_CHAT_COMPLETION_MODEL", "gpt-4o-mini")
api_version = os.getenv("AZURE_OPENAI_API_VERSION")

# Create an AI agent following the standard Microsoft Agent Framework pattern
agent = OpenAIChatCompletionClient(
    azure_endpoint=endpoint,
    model=deployment_name,
    api_version=api_version,
    credential=DefaultAzureCredential()
).as_agent(
    instructions="You are a helpful assistant that can answer questions and provide information.",
    name="MyDurableAgent"
)

# Configure the function app to host the agent with durable thread management
app = AgentFunctionApp(agents=[agent])

이 코드:

• 환경 변수에서 Azure OpenAI 구성을 검색합니다.
• Azure 자격 증명을 사용하여 Azure OpenAI 클라이언트를 만듭니다.
• 지침 및 이름을 사용하여 AI 에이전트를 만듭니다.
• 지속성 스레드 관리를 사용하여 에이전트를 호스트하도록 Azure Functions 앱을 구성합니다.

이전 Program.cs과 함께 번역 에이전트를 기존 MyDurableAgent에 등록하도록 업데이트하십시오:

using System;
using Azure.AI.Projects;
using Azure.Identity;
using Microsoft.Agents.AI;
using Microsoft.Agents.AI.Hosting.AzureFunctions;
using Microsoft.Azure.Functions.Worker.Builder;
using Microsoft.Extensions.Hosting;

// Get the Azure OpenAI configuration
string endpoint = Environment.GetEnvironmentVariable("AZURE_OPENAI_ENDPOINT")
    ?? throw new InvalidOperationException("AZURE_OPENAI_ENDPOINT is not set.");
string deploymentName = Environment.GetEnvironmentVariable("AZURE_OPENAI_DEPLOYMENT")
    ?? "gpt-4o-mini";

// Create the Microsoft Foundry client
AIProjectClient client = new(new Uri(endpoint), new DefaultAzureCredential());

// Create the main agent from the first tutorial
AIAgent mainAgent = client.AsAIAgent(
    model: deploymentName,
    instructions: "You are a helpful assistant that can answer questions and provide information.",
    name: "MyDurableAgent");

// Create translation agents
AIAgent frenchAgent = client.AsAIAgent(
    model: deploymentName,
    instructions: "You are a translator. Translate the following text to French. Return only the translation, no explanations.",
    name: "FrenchTranslator");

AIAgent spanishAgent = client.AsAIAgent(
    model: deploymentName,
    instructions: "You are a translator. Translate the following text to Spanish. Return only the translation, no explanations.",
    name: "SpanishTranslator");

// Build and configure the Functions host
using IHost app = FunctionsApplication
    .CreateBuilder(args)
    .ConfigureFunctionsWebApplication()
    .ConfigureDurableAgents(options =>
    {
        // Register all agents for use in orchestrations and HTTP endpoints
        options.AddAIAgent(mainAgent);
        options.AddAIAgent(frenchAgent);
        options.AddAIAgent(spanishAgent);
    })
    .Build();

app.Run();

이전 function_app.py과 함께 번역 에이전트를 기존 MyDurableAgent에 등록하도록 업데이트하십시오:

import os
from azure.identity import DefaultAzureCredential
from agent_framework.azure import AgentFunctionApp
from agent_framework.openai import OpenAIChatCompletionClient

# Get the Azure OpenAI configuration
endpoint = os.getenv("AZURE_OPENAI_ENDPOINT")
if not endpoint:
    raise ValueError("AZURE_OPENAI_ENDPOINT is not set.")
deployment_name = os.getenv("AZURE_OPENAI_CHAT_COMPLETION_MODEL", "gpt-4o-mini")
api_version = os.getenv("AZURE_OPENAI_API_VERSION")

# Create the Azure OpenAI client
chat_client = OpenAIChatCompletionClient(
    azure_endpoint=endpoint,
    model=deployment_name,
    api_version=api_version,
    credential=DefaultAzureCredential()
)

# Create the main agent from the first tutorial
main_agent = chat_client.as_agent(
    instructions="You are a helpful assistant that can answer questions and provide information.",
    name="MyDurableAgent"
)

# Create translation agents
french_agent = chat_client.as_agent(
    instructions="You are a translator. Translate the following text to French. Return only the translation, no explanations.",
    name="FrenchTranslator"
)

spanish_agent = chat_client.as_agent(
    instructions="You are a translator. Translate the following text to Spanish. Return only the translation, no explanations.",
    name="SpanishTranslator"
)

# Create the function app and register all agents
app = AgentFunctionApp(agents=[main_agent, french_agent, spanish_agent])

프로젝트 디렉터리에 이름이 지정된 AgentOrchestration.cs 새 파일을 만듭니다.

using System.Collections.Generic;
using System.Threading.Tasks;
using Microsoft.Agents.AI;
using Microsoft.Agents.AI.DurableTask;
using Microsoft.Azure.Functions.Worker;
using Microsoft.DurableTask;

namespace MyDurableAgent;

public static class AgentOrchestration
{
    // Define a strongly-typed response structure for agent outputs
    public sealed record TextResponse(string Text);

    [Function("agent_orchestration_workflow")]
    public static async Task<Dictionary<string, string>> AgentOrchestrationWorkflow(
        [OrchestrationTrigger] TaskOrchestrationContext context)
    {
        var input = context.GetInput<string>() ?? throw new ArgumentNullException(nameof(context), "Input cannot be null");

        // Step 1: Get the main agent's response
        DurableAIAgent mainAgent = context.GetAgent("MyDurableAgent");
        AgentResponse<TextResponse> mainResponse = await mainAgent.RunAsync<TextResponse>(input);
        string agentResponse = mainResponse.Result.Text;

        // Step 2: Fan out - get the translation agents and run them concurrently
        DurableAIAgent frenchAgent = context.GetAgent("FrenchTranslator");
        DurableAIAgent spanishAgent = context.GetAgent("SpanishTranslator");

        Task<AgentResponse<TextResponse>> frenchTask = frenchAgent.RunAsync<TextResponse>(agentResponse);
        Task<AgentResponse<TextResponse>> spanishTask = spanishAgent.RunAsync<TextResponse>(agentResponse);

        // Step 3: Wait for both translation tasks to complete (fan-in)
        await Task.WhenAll(frenchTask, spanishTask);

        // Get the translation results
        TextResponse frenchResponse = (await frenchTask).Result;
        TextResponse spanishResponse = (await spanishTask).Result;

        // Step 4: Combine results into a dictionary
        var result = new Dictionary<string, string>
        {
            ["original"] = agentResponse,
            ["french"] = frenchResponse.Text,
            ["spanish"] = spanishResponse.Text
        };

        return result;
    }
}

오케스트레이션 함수를 파일에 추가합니다 function_app.py .

import azure.durable_functions as df

@app.orchestration_trigger(context_name="context")
def agent_orchestration_workflow(context: df.DurableOrchestrationContext):
    """
    Orchestration function that coordinates multiple agents.
    Returns a dictionary with the original response and translations.
    """
    input_text = context.get_input()

    # Step 1: Get the main agent's response
    main_agent = app.get_agent(context, "MyDurableAgent")
    main_response = yield main_agent.run(input_text)
    agent_response = main_response.text

    # Step 2: Fan out - get the translation agents and run them concurrently
    french_agent = app.get_agent(context, "FrenchTranslator")
    spanish_agent = app.get_agent(context, "SpanishTranslator")

    parallel_tasks = [
        french_agent.run(agent_response),
        spanish_agent.run(agent_response)
    ]

    # Step 3: Wait for both translation tasks to complete (fan-in)
    translations = yield context.task_all(parallel_tasks) # type: ignore

    # Step 4: Combine results into a dictionary
    result = {
        "original": agent_response,
        "french": translations[0].text,
        "spanish": translations[1].text
    }

    return result