Estensione durevole

Durable Extension for Microsoft Agent Framework offre un'esecuzione durevole agli agenti, alle orchestrazioni multi-agente e ai flussi di lavoro di Agent Framework Microsoft. È possibile usarlo per rendere persistenti le sessioni dell'agente, l'orchestrazione dei checkpoint e lo stato del flusso di lavoro, il ripristino da errori e il ridimensionamento del lavoro tra gli host distribuiti senza modificare la logica dell'agente principale.

L'estensione supporta due modelli di hosting in C# e Python:

  • Funzioni di Azure per l'hosting serverless gestito con il modello di programmazione Funzioni di Azure.
  • Bring your own compute/self-hosted per l'esecuzione di agenti durevoli e flussi di lavoro nel processo di lavoro, nel servizio, nel contenitore, nell'ambiente Kubernetes o nell'infrastruttura di app esistente.

Overview

Gli agenti permanenti combinano il modello di programmazione di Agent Framework con l'infrastruttura Durable Task, ad esempio l'Utilità di pianificazione attività durevole, per creare agenti che:

  • Rendere persistente lo stato automaticamente tra le richieste e le esecuzioni del ruolo di lavoro
  • Riprendere dopo gli errori senza perdere il contesto della conversazione o ripetere il lavoro completato
  • Scalabilità tra i ruoli di lavoro distribuiti e senza stato in base alla richiesta
  • Orchestrare flussi di lavoro multi-agente con garanzie di esecuzione affidabili
  • Flussi di lavoro di Checkpoint Agent Framework creati con il modello di flusso di lavoro basato su grafo
  • Sospendere l'input umano o gli eventi esterni senza utilizzare token di calcolo o di modello durante l'attesa
  • Trasmettere le risposte in modo affidabile quando configurate con un broker di flusso affidabile, ad esempio Redis
  • Gestire il ciclo di vita della sessione con la pulizia TTL (Session Time to Live) e il monitoraggio basato su dashboard

Quando usare agenti durevoli

Scegliere agenti durevoli quando necessario:

  • Stato conversazione persistente: le sessioni dell'agente sopravvivono agli arresti anomali, ai riavvii e agli eventi di scalabilità orizzontale
  • Orchestrazioni complesse: coordinare più agenti con flussi di lavoro deterministici e affidabili che possono essere eseguiti per giorni o settimane
  • Orchestrazione guidata dagli eventi: integrazione con trigger, code, webhook, timer o eventi dell'applicazione esistenti
  • Stato conversazione automatico: la cronologia delle conversazioni dell'agente viene gestita automaticamente e resa persistente senza richiedere la gestione esplicita dello stato nel codice
  • flussi di lavoro di Durable Agent Framework: rendere durevoli i flussi di lavoro basati su grafo Microsoft Agent Framework in modo che ogni passaggio possa essere sottoposto a checkpoint e ripreso
  • Sessioni di lunga durata: mantenere disponibili conversazioni utili durante l'uso della pulizia TTL (Session Time to Live) per rimuovere automaticamente le sessioni inattive
  • Risposte in tempo reale affidabili: output del token di flusso in modo durevole per le applicazioni che necessitano di esperienza utente in tempo reale con garanzie di recapito

Questo approccio di hosting è diverso dall'hosting dell'agente gestito basato sul servizio ,ad esempio il servizio foundry Agent, che fornisce un'infrastruttura completamente gestita senza richiedere la distribuzione o la gestione degli host di lavoro. Gli agenti durevoli sono ideali quando è necessaria la flessibilità della distribuzione code-first combinata con la gestione dello stato durevole.

Scegliere un modello di hosting

Modello di hosting Scegliere l'opzione quando è necessario
Funzioni di Azure Modello di hosting serverless gestito; scale-out e scalabilità orizzontale predefiniti; Funzioni di Azure trigger e associazioni; Endpoint HTTP generati dal modello di programmazione funzioni; trigger del server MCP; e una gestione minima dell'infrastruttura host.
Bring Your Own Compute/self-hosted Maggiore controllo sul processo host, l'ambiente di distribuzione, il ciclo di vita del runtime, l'infrastruttura, la rete, l'autenticazione o l'integrazione con un'app o un servizio esistente. Usare questo modello per contenitori, Kubernetes, ruoli di lavoro a esecuzione prolungata, app console, servizi personalizzati o ambienti di hosting non funzioni.

Se ospitato nel piano di hosting Funzioni di Azure Flex Consumption, gli agenti possono scalare fino a migliaia di istanze o a zero istanze quando non sono in uso, consentendoti di pagare solo per il calcolo necessario. Negli scenari self-hosted, il proprio host controlla la durata del processo, il ridimensionamento, la rete e la distribuzione.

Come iniziare

In un progetto di .NET scegliere il set di pacchetti per il modello di hosting.

Per Funzioni di Azure hosting, aggiungere il pacchetto di integrazione Funzioni di Azure e i pacchetti del ruolo di lavoro di Funzioni.

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

Annotazioni

Oltre a questi pacchetti, assicurarsi che il progetto usi la versione 2.2.0 o successiva del pacchetto Microsoft.Azure.Functions.Worker .

Per l'hosting bring-your-own-compute, aggiungere il pacchetto di integrazione di durable task di base e i pacchetti di lavoro/client durable task scheduler usati dall'host:

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

In un progetto di Python scegliere il pacchetto per il modello di hosting.

Per Funzioni di Azure hosting, installare il pacchetto di integrazione Funzioni di Azure.

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

Per l'hosting bring-your-own-compute, installare il pacchetto di integrazione Durable Task.

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

hosting Funzioni di Azure

Con Durable Extension è possibile distribuire e ospitare agenti Microsoft Agent Framework in Funzioni di Azure con endpoint HTTP predefiniti e chiamate basate sull'orchestrazione. Funzioni di Azure offre prezzi basati su eventi e con pagamento in base alla chiamata con scalabilità automatica e gestione minima dell'infrastruttura.

Quando si configura un agente durevole in Funzioni di Azure, l'estensione crea automaticamente endpoint HTTP per l'agente e gestisce l'infrastruttura sottostante per l'archiviazione dello stato della conversazione, la gestione delle richieste simultanee e il coordinamento dei flussi di lavoro multi-agente. L'integrazione dell'hosting Funzioni di Azure offre anche praticità specifiche di Funzioni, ad esempio le API REST generate per l'invio di messaggi, il controllo dello stato e la gestione delle sessioni, oltre a trigger come il trigger del server MCP per l'hosting di agenti come server MCP senza scrivere trigger glue.

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

Avvertimento

DefaultAzureCredential è utile per lo sviluppo, ma richiede un'attenta considerazione nell'ambiente di produzione. Nell'ambiente di produzione prendere in considerazione l'uso di credenziali specifiche ,ad esempio ManagedIdentityCredential, per evitare problemi di latenza, probe di credenziali indesiderate e potenziali rischi per la sicurezza dai meccanismi di fallback.

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

Hosting bring-your-own-compute/self-hosted

Usare l'hosting bring-your-own-compute quando si vogliono usare le funzionalità durable extension senza usare il modello di programmazione Funzioni di Azure. In questo modello il processo avvia un ruolo di lavoro Durable Task, registra agenti durevoli o flussi di lavoro e si connette a un back-end dell'utilità di pianificazione durevole. Il codice client può essere eseguito nello stesso processo o in un servizio separato.

I ruoli di lavoro self-hosted usano le stesse funzionalità principali di Durable Extension di Funzioni di Azure hosting: checkpoint e ripresa, orchestrazione dell'agente deterministico, flussi di lavoro durevoli di Agent Framework, attese umane nel ciclo, streaming affidabile, pulizia della sessione inattiva, visibilità del dashboard ed esecuzione distribuita tra istanze di lavoro senza stato. L'host è responsabile dell'esposizione delle proprie API, della gestione del ciclo di vita, della rete, dell'autenticazione e del modello di distribuzione.

Configurare l'host con il pacchetto di integrazione di Durable Task di base. Usare ConfigureDurableAgents per gli agenti durevoli e ConfigureDurableWorkflows per i flussi di lavoro Microsoft di Agent Framework basati su grafo.

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

Vedere gli esempi della console .NET Durable Agents e .NET esempi di console Durable Workflow per esempi auto-ospitati eseguibili.

Usare il pacchetto di integrazione di Durable Task per eseguire un processo di lavoro che registra gli agenti e rimane in ascolto delle richieste. Il codice client può connettersi allo stesso hub attività utilità di pianificazione durevole da un altro processo.

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

Vedere gli esempi di Python Durable Task per esempi di client di lavoro, tra cui hosting con agente singolo, routing multi-agente, flusso affidabile, concatenamento dell'orchestrazione, concorrenza, condizionali e modelli di ciclo umano-in-the-loop.

Flussi di lavoro di Durable Agent Framework

La durabilità non è limitata alle orchestrazioni durevoli. Microsoft flussi di lavoro di Agent Framework creati con il modello di flusso di lavoro basato su grafo possono anche essere resi durevoli. L'esecuzione del flusso di lavoro dei checkpoint dell'estensione durevole in modo che l'executor e i passaggi dell'agente completati non vengano ripetuti dopo un riavvio o un errore del processo.

Usare orchestrazioni durevoli quando si desidera un coordinamento imperativo con diramazioni, timer, attività ed eventi esterni basati su codice. Usare flussi di lavoro durable Agent Framework quando si vuole un grafico dichiarativo di executor e agenti con routing tipizzato, fan-out/fan-in, archi condizionali, eventi del flusso di lavoro, stato condiviso, flussi di lavoro secondari o porte di richiesta human-in-the-loop.

Annotazioni

I flussi di lavoro di Durable Agent Framework sono diversi dall'archiviazione dei checkpoint nei flussi di lavoro standard. L'archiviazione del checkpoint consente di riprendere un'esecuzione del flusso di lavoro nel runtime di Agent Framework. Durable Extension esegue il flusso di lavoro nell'infrastruttura Durable Task in modo che lo stato del flusso di lavoro venga sottoposto a checkpoint e ripristinato tra i ruoli di lavoro durevoli distribuiti. Per il checkpoint del flusso di lavoro standard, vedere Checkpoint e ripresa.

Registrare flussi di lavoro basati su grafo con ConfigureDurableWorkflows per le app self-hosted o ConfigureDurableWorkflows nel generatore di app per le funzioni per Funzioni di Azure hosting.

Vedere gli esempi di .NET Durable Workflow Funzioni di Azure e .NET esempi di console Durable Workflow.

Gli esempi di flussi di lavoro durevoli sono disponibili per Funzioni di Azure hosting, inclusi lo stato condiviso, lo stato condiviso, l'esecuzione parallela del flusso di lavoro e i flussi di lavoro del ciclo umano.Durable workflow samples are available for Funzioni di Azure hosting, including shared state, no shared state, parallel workflow execution, and human-in-the-loop workflow.

Vedere gli esempi Python Funzioni di Azure per esempi di agente durevole, orchestrazione, server MCP e flusso di lavoro.

Samples

Language Modello di hosting Samples
C# Funzioni di Azure .NET Durable Agents - Funzioni di Azure, .NET Flussi di lavoro durevoli - Funzioni di Azure
C# Bring Your Own Compute/self-hosted .NET Durable Agents - App console, .NET Flussi di lavoro durevoli - App console
Python Funzioni di Azure esempi Python Funzioni di Azure
Python Bring Your Own Compute/self-hosted esempi di Python Durable Task

Thread dell'agente con stato con cronologia conversazioni

Gli agenti mantengono thread persistenti che sopravvivono a più interazioni. Ogni thread viene identificato da un ID thread univoco e archivia la cronologia di conversazione completa nell'archiviazione durevole gestita dall'infrastruttura di Durable Task, ad esempio l'Utilità di pianificazione attività durevole.

Questo modello consente la continuità della conversazione in cui lo stato dell'agente viene mantenuto tramite crash e riavvii del processo, permettendo la conservazione completa della cronologia delle conversazioni tra thread utente. L'archiviazione durevole garantisce che anche se un processo host riavvia o riprende il lavoro in un'istanza di lavoro diversa, la conversazione continua senza problemi da dove è stata interrotta.

Usare la pulizia TTL (Session Time to Live) per i carichi di lavoro che necessitano di continuità durevole durante l'uso attivo, ma devono pulire automaticamente le conversazioni inattive. La pulizia basata su TTL impedisce alle sessioni inutilizzate e alla cronologia delle conversazioni di accumularsi a tempo indeterminato mantenendo lo stato attivo della sessione.

L'esempio di Funzioni di Azure seguente illustra più richieste HTTP allo stesso thread, che illustra come il contesto della conversazione persiste. Nelle app self-hosted usare le API client Durable Task del proprio processo o servizio.

# First interaction - start a new thread
curl -X POST https://your-function-app.azurewebsites.net/api/agents/Joker/run \
  -H "Content-Type: text/plain" \
  -d "Tell me a joke about pirates"

# Response includes thread ID in x-ms-thread-id header and joke as plain text
# HTTP/1.1 200 OK
# Content-Type: text/plain
# x-ms-thread-id: @dafx-joker@263fa373-fa01-4705-abf2-5a114c2bb87d
#
# Why don't pirates shower before they walk the plank? Because they'll just wash up on shore later!

# Second interaction - continue the same thread with context
curl -X POST "https://your-function-app.azurewebsites.net/api/agents/Joker/run?thread_id=@dafx-joker@263fa373-fa01-4705-abf2-5a114c2bb87d" \
  -H "Content-Type: text/plain" \
  -d "Tell me another one about the same topic"

# Agent remembers the pirate context from the first message and responds with plain text
# What's a pirate's favorite letter? You'd think it's R, but it's actually the C!

Lo stato dell'agente viene mantenuto nell'archiviazione durevole, abilitando l'esecuzione distribuita tra più istanze. Qualsiasi istanza può riprendere l'esecuzione di un agente dopo interruzioni o errori, garantendo un funzionamento continuo.

Streaming affidabile

Durable Extension supporta lo streaming affidabile per le applicazioni che richiedono il recapito dei token in tempo reale con garanzie di recapito durevoli. Lo streaming può essere usato con l'estensione principale in entrambi i modelli di hosting, ma gli host distribuiti necessitano di un broker di flusso affidabile, ad esempio Redis, in modo che i flussi di token possano essere distribuiti in modo coerente tra riavvii dei processi, riconnessioni o modifiche del ruolo di lavoro.

Usare lo streaming affidabile quando l'esperienza utente dipende dalle risposte incrementali, ma il carico di lavoro richiede ancora una semantica di esecuzione durevole. Per esempi eseguibili, vedere gli esempi di Python Durable Task, che includono modelli di streaming affidabili.

Orchestrazioni multi-agente deterministiche

Durable Extension supporta la creazione di flussi di lavoro deterministici che coordinano più agenti usando orchestrazioni di Durable Task. In Funzioni di Azure, questi usano Durable Functions orchestrazioni; negli host bring-your-own-compute vengono eseguiti tramite durable task worker e client configurati.

Le orchestrazioni sono flussi di lavoro basati su codice che coordinano più operazioni , ad esempio chiamate di agenti, chiamate API esterne o timer, in modo affidabile. Deterministico significa che il codice di orchestrazione viene eseguito allo stesso modo quando viene riprodotto dopo un errore, rendendo i flussi di lavoro affidabili e di cui è possibile eseguire il debug. Quando si riproduce la cronologia di un'orchestrazione, è possibile vedere esattamente cosa è successo in ogni passaggio.

Le orchestrazioni vengono eseguite in modo affidabile, sopravvivendo ai guasti tra le chiamate di agenti e offrono processi in modo prevedibile e ripetibile. Ciò li rende ideali per scenari multi-agente complessi in cui è necessario un ordine di esecuzione e una tolleranza di errore garantiti.

Orchestrazioni sequenziali

Nel modello sequenziale multi-agente, gli agenti specializzati vengono eseguiti in un ordine specifico, in cui l'output di ogni agente può influenzare l'esecuzione dell'agente successivo. Questo modello supporta la logica condizionale e la diramazione secondo le risposte dell'agente.

Quando si usano gli agenti nelle orchestrazioni, è necessario usare l'API context.GetAgent() per ottenere un'istanza DurableAIAgent, che è una sottoclasse speciale del tipo standard AIAgent che include uno degli agenti registrati. Il DurableAIAgent wrapper garantisce che le chiamate dell'agente vengano tracciate correttamente e che siano sottoposte a checkpoint dal framework di orchestrazione durevole.

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

Quando si utilizzano agenti nelle orchestrazioni, si deve usare il metodo app.get_agent() per ottenere un'istanza durevole dell'agente, ossia un involucro speciale attorno a uno degli agenti registrati. Il wrapper dell'agente durevole garantisce che le chiamate dell'agente vengano rilevate correttamente e monitorate correttamente dal framework di orchestrazione durevole.

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

Le orchestrazioni coordinano il lavoro fra più agenti, riuscendo a sopravvivere ai fallimenti tra le chiamate degli agenti. Il contesto di orchestrazione fornisce metodi per recuperare e interagire con gli agenti ospitati all'interno delle orchestrazioni.

Orchestrazioni parallele

Nel modello multi-agente parallelo si eseguono più agenti contemporaneamente e quindi si aggregano i risultati. Questo modello è utile per raccogliere diverse prospettive o elaborare contemporaneamente sottoattività indipendenti.

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', '')

L'esecuzione parallela viene rilevata usando un elenco di attività. Il checkpoint automatico garantisce che le esecuzioni completate dell'agente non vengano ripetute o perse se si verifica un errore durante l'aggregazione.

Orchestrazioni human-in-the-loop

Le orchestrazioni degli agenti deterministici possono sospendere l'input umano, l'approvazione o la revisione senza usare risorse di calcolo. L'esecuzione durevole consente alle orchestrazioni di attendere giorni o persino settimane durante l'attesa delle risposte umane. In combinazione con l'hosting serverless, tutte le risorse di calcolo vengono disattivate durante il periodo di attesa, eliminando i costi di calcolo fino a quando l'utente non fornisce l'input.

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

Le orchestrazioni degli agenti deterministici possono attendere eventi esterni, conservando in modo durevole il loro stato in attesa di feedback umano, sopravvivendo a errori, riavvii e periodi di attesa prolungati. Quando arriva la risposta umana, l'orchestrazione riprende automaticamente con il contesto di conversazione completo e lo stato di esecuzione intatti.

Fornire input umano

Per inviare l'approvazione o l'input a un'orchestrazione in attesa, generare un evento esterno all'istanza di orchestrazione usando Durable Task Client SDK o gli endpoint dell'estensione durable Funzioni di Azure. Ad esempio, un revisore potrebbe approvare il contenuto tramite un modulo Web che chiama:

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)

Efficienza dei costi

I flussi di lavoro con intervento umano con agenti durevoli sono estremamente convenienti quando sono ospitati nel piano Flex Consumption di Funzioni di Azure. Per un flusso di lavoro in attesa di 24 ore per l'approvazione, si paga solo per pochi secondi di tempo di esecuzione (il tempo per generare contenuto, inviare una notifica ed elaborare la risposta), non le 24 ore di attesa. Durante il periodo di attesa, non vengono utilizzate risorse di calcolo.

Osservabilità con Durable Task Scheduler

Il Durable Task Scheduler (DTS) è il back-end durevole consigliato per i tuoi agenti durevoli, offrendo le migliori prestazioni, infrastruttura interamente gestita e osservabilità integrata tramite una dashboard dell'interfaccia utente. Funzioni di Azure le app possono usare altri back-end di archiviazione (ad esempio Archiviazione di Azure), ma DTS è ottimizzato in modo specifico per carichi di lavoro durevoli e offre funzionalità di monitoraggio e prestazioni superiori. I ruoli di lavoro self-hosted usano anche DTS per la pianificazione durevole, lo stato e la visibilità del dashboard.

Informazioni dettagliate sulla sessione dell'agente

  • Cronologia conversazioni: visualizzare la cronologia di chat completa per ogni sessione dell'agente, inclusi tutti i messaggi, le chiamate agli strumenti e il contesto della conversazione in qualsiasi momento
  • Tempistica delle attività: Monitorare il tempo necessario per completare attività specifiche e interazioni degli agenti.

Screenshot del dashboard di Durable Task Scheduler che mostra la cronologia della chat dell'agente con thread e messaggi di conversazione.

Informazioni dettagliate sull'orchestrazione

  • Visualizzazione multi-agente: Vedere il flusso di esecuzione quando si chiamano più agenti specializzati con rappresentazione visiva delle esecuzioni parallele e della diramazione condizionale
  • Cronologia di esecuzione: accedere ai log di esecuzione dettagliati
  • Monitoraggio in tempo reale: Tieni traccia delle orchestrazioni attive, degli elementi di lavoro in coda e degli stati degli agenti nella vostra implementazione
  • Metriche delle prestazioni: monitorare i tempi di risposta dell'agente, l'utilizzo dei token e la durata dell'orchestrazione

Screenshot del dashboard del Durable Task Scheduler che mostra la visualizzazione dell'orchestrazione con diverse interazioni tra agenti ed esecuzione del flusso di lavoro.

Funzionalità di debug

  • Visualizzare gli output dell'agente strutturato e i risultati delle chiamate degli strumenti
  • Chiamate allo strumento di traccia e relativi risultati
  • Monitorare la gestione degli eventi esterni per scenari umani nel ciclo

Il dashboard consente di comprendere esattamente le operazioni eseguite dagli agenti, diagnosticare rapidamente i problemi e ottimizzare le prestazioni in base ai dati di esecuzione reali.

Esercitazione: Creare ed eseguire un agente durevole con Funzioni di Azure

Questa esercitazione illustra come creare ed eseguire un agente di intelligenza artificiale durevole usando il modello di hosting Funzioni di Azure per Durable Extension. Creerai un'app Funzioni di Azure che ospita un agente con stato con endpoint HTTP integrati e imparerai come monitorarla usando il dashboard del Scheduler attività durevole. Per gli agenti self-hosted, vedere gli esempi.

Prerequisiti

Prima di iniziare, assicurarsi di avere i prerequisiti seguenti:

Annotazioni

Microsoft Agent Framework è supportato con tutte le versioni supportate attivamente di .NET. Ai fini di questo esempio, è consigliabile usare .NET 9 SDK o una versione successiva.

Scaricare il progetto di avvio rapido

Usare l'interfaccia della riga di comando per sviluppatori di Azure per inizializzare un nuovo progetto dal modello di avvio rapido degli agenti durevoli.

  1. Crea una nuova directory per il tuo progetto e naviga al suo interno.

    mkdir MyDurableAgent
cd MyDurableAgent

  1. Inizializzare il progetto dal modello:

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

    Quando viene richiesto un nome di ambiente, immettere un nome come my-durable-agent.

Verrà scaricato il progetto di avvio rapido con tutti i file necessari, tra cui la configurazione di Funzioni di Azure, il codice dell'agente e l'infrastruttura come modelli di codice.

  1. Crea una nuova directory per il tuo progetto e naviga al suo interno.

    mkdir MyDurableAgent
cd MyDurableAgent

  1. Inizializzare il progetto dal modello:

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

    Quando viene richiesto un nome di ambiente, immettere un nome come my-durable-agent.

  2. Creare e attivare un ambiente virtuale:

    uv venv .venv
source .venv/bin/activate

Annotazioni

python3 -m venv .venv funziona anche, ma può bloccarsi a tempo indeterminato per Windows con Microsoft Store Python a causa di un problema di ensurepip noto. Usare uv venv .venv per evitare questo problema.

  1. Installare i pacchetti necessari:

    python -m pip install -r requirements.txt

Verrà scaricato il progetto di avvio rapido con tutti i file necessari, tra cui la configurazione di Funzioni di Azure, il codice dell'agente e l'infrastruttura come modelli di codice. Prepara anche un ambiente virtuale con le dipendenze necessarie.

Effettuare il provisioning delle risorse di Azure

Usare la CLI per sviluppatori di Azure per creare le risorse Azure necessarie per il tuo agente durevole.

  1. Eseguire la configurazione dell'infrastruttura

    azd provision

    Questo comando crea:

    • Un servizio OpenAI di Azure con una distribuzione gpt-4o-mini
    • Un'app di Funzioni di Azure con il piano di hosting Flex Consumption
    • Un account di archiviazione di Azure per il runtime delle Funzioni di Azure e l'archiviazione persistente
    • Istanza di Durable Task Scheduler (piano Consumption) per gestire lo stato dell’agente
    • Configurazioni di rete e identità necessarie

  2. Quando richiesto, selezionare la sottoscrizione di Azure e scegliere una posizione per le risorse.

Il processo di provisioning richiede alcuni minuti. Al termine, azd salva le informazioni sulle risorse create nell'ambiente.

Esaminare il codice dell'agente

Esaminiamo ora il codice che definisce l'agente durevole.

Aprire Program.cs per visualizzare la configurazione dell'agente:

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

Questo codice:

  1. Recupera la configurazione di Azure OpenAI dalle variabili di ambiente.
  2. Crea un client OpenAI di Azure usando le credenziali di Azure.
  3. Crea un agente di intelligenza artificiale con istruzioni e un nome.
  4. Configura l'app di Funzioni Azure per ospitare l'agente con la gestione durevole dei thread.

Aprire function_app.py per visualizzare la configurazione dell'agente:

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

Questo codice:

  • Recupera la configurazione di Azure OpenAI dalle variabili di ambiente.
  • Crea un client OpenAI di Azure usando le credenziali di Azure.
  • Crea un agente di intelligenza artificiale con istruzioni e un nome.
  • Configura l'app di Funzioni Azure per ospitare l'agente con la gestione durevole dei thread.

L'agente è ora pronto per essere ospitato in Funzioni di Azure. L'estensione Durable Task crea automaticamente gli endpoint HTTP per interagire con il tuo agente e gestisce lo stato della conversazione tra più richieste.

Configurare le impostazioni locali

Creare un local.settings.json file per lo sviluppo locale in base al file di esempio incluso nel progetto.

  1. Copiare il file di impostazioni di esempio:

    cp local.settings.sample.json local.settings.json

  1. Ottieni l'endpoint OpenAI di Azure dalle risorse fornite.

    azd env get-value AZURE_OPENAI_ENDPOINT

  2. Aprire local.settings.json e sostituire <your-resource-name> del valore AZURE_OPENAI_ENDPOINT con l'endpoint del comando eseguito precedentemente.

L'aspetto local.settings.json dovrebbe essere simile al seguente:

{
  "IsEncrypted": false,
  "Values": {
    // ... other settings ...
    "AZURE_OPENAI_ENDPOINT": "https://your-openai-resource.openai.azure.com",
    "AZURE_OPENAI_DEPLOYMENT": "gpt-4o-mini",
    "TASKHUB_NAME": "default"
  }
}

Annotazioni

Il local.settings.json file viene usato solo per lo sviluppo locale e non viene distribuito in Azure. Per le distribuzioni di produzione, queste impostazioni vengono configurate automaticamente nell'app Funzioni di Azure dai modelli di infrastruttura.

Avvia le dipendenze di sviluppo locale

Per eseguire agenti software durevoli in locale, è necessario attivare due servizi:

  • Azurite: emula i servizi di Archiviazione di Azure (usati da Funzioni di Azure per la gestione di trigger e stato interno).
  • Emulatore DTS (Durable Task Scheduler): gestisce lo stato durevole (cronologia delle conversazioni, stato di orchestrazione) e la pianificazione per gli agenti

Avvia Azurite

Azurite emula i servizi di Archiviazione di Azure in locale. Funzioni di Azure la usa per la gestione dello stato interno. Sarà necessario eseguirlo in una nuova finestra del terminale e mantenerlo in esecuzione mentre si sviluppa e si testa l'agente persistente.

  1. Aprire una nuova finestra del terminale ed eseguire il pull dell'immagine Azurite Docker:

    docker pull mcr.microsoft.com/azure-storage/azurite

  2. Avvia Azurite in una finestra del terminale:

    docker run -p 10000:10000 -p 10001:10001 -p 10002:10002 mcr.microsoft.com/azure-storage/azurite

    Azurite avvierà e rimarrà in ascolto sulle porte predefinite per i servizi BLOB (10000), Queue (10001) e Table (10002).

Mantenere aperta questa finestra del terminale durante lo sviluppo e il test dell'agente durevole.

Tip

Per altre informazioni su Azurite, inclusi i metodi di installazione alternativi, vedere Usare l'emulatore Azurite per lo sviluppo locale di Archiviazione di Azure.

Avviare l'emulatore di Durable Task Scheduler

L'emulatore DTS fornisce un back-end persistente per la gestione dello stato degli agenti e delle orchestrazioni. Archivia la cronologia delle conversazioni e assicura la persistenza dello stato dell’agente tra i riavvii. Attiva anche orchestrazioni e agenti durevoli. Dovrai eseguire questo in una nuova finestra del terminale separata e lasciarlo in esecuzione mentre sviluppi e testi il tuo agente persistente.

  1. Aprire un'altra finestra del terminale ed eseguire il pull dell'immagine Docker dell'emulatore DTS:

    docker pull mcr.microsoft.com/dts/dts-emulator:latest

  2. Eseguire l'emulatore DTS:

    docker run -p 8080:8080 -p 8082:8082 mcr.microsoft.com/dts/dts-emulator:latest

    Questo comando avvia l'emulatore ed espone:

    • Porta 8080: endpoint gRPC per lo scheduler di attività durevoli (utilizzato dall'applicazione Funzioni)
    • Porta 8082: Il cruscotto amministrativo

  3. Il dashboard sarà disponibile all'indirizzo http://localhost:8082.

Mantenere aperta questa finestra del terminale durante lo sviluppo e il test dell'agente durevole.

Tip

Per altre informazioni sull'emulatore DTS, tra cui come configurare più hub attività e accedere al dashboard, vedere Sviluppare con Durable Task Scheduler.

Eseguire l'applicazione di funzioni

Ora sei pronto/a per eseguire l'app di Funzioni di Azure con l'agente durevole.

  1. In una nuova finestra del terminale (mantenendo sia Azurite che l'emulatore DTS in esecuzione in finestre separate), naviga nella directory del progetto.

  2. Avviare il runtime di Funzioni di Azure:

    func start

  3. Dovresti vedere l'output che indica che la tua applicazione di funzioni è in esecuzione, inclusi gli endpoint HTTP per l'agente.

    Functions:
     http-MyDurableAgent: [POST] http://localhost:7071/api/agents/MyDurableAgent/run
     dafx-MyDurableAgent: entityTrigger

Questi endpoint gestiscono automaticamente lo stato della conversazione: non è necessario creare o gestire gli oggetti thread manualmente.

Testare l'agente localmente

È ora possibile interagire con l'agente durevole usando le richieste HTTP. L'agente mantiene lo stato della conversazione tra più richieste, abilitando conversazioni a più turni.

Inizia una nuova conversazione

Creare un nuovo thread e inviare il primo messaggio:

curl -i -X POST http://localhost:7071/api/agents/MyDurableAgent/run \
  -H "Content-Type: text/plain" \
  -d "What are three popular programming languages?"

Risposta di esempio (si noti che l'intestazione x-ms-thread-id contiene l'ID thread):

HTTP/1.1 200 OK
Content-Type: text/plain
x-ms-thread-id: @dafx-mydurableagent@263fa373-fa01-4705-abf2-5a114c2bb87d
Content-Length: 189

Three popular programming languages are Python, JavaScript, and Java. Python is known for its simplicity and readability, JavaScript powers web interactivity, and Java is widely used in enterprise applications.

Salvare l'ID del thread dall'intestazione x-ms-thread-id ,ad esempio @dafx-mydurableagent@263fa373-fa01-4705-abf2-5a114c2bb87d, per la richiesta successiva.

Continuare la conversazione

Inviare un messaggio di completamento allo stesso thread includendo l'ID del thread come parametro di query:

curl -X POST "http://localhost:7071/api/agents/MyDurableAgent/run?thread_id=@dafx-mydurableagent@263fa373-fa01-4705-abf2-5a114c2bb87d" \
  -H "Content-Type: text/plain" \
  -d "Which one is best for beginners?"

Sostituire @dafx-mydurableagent@263fa373-fa01-4705-abf2-5a114c2bb87d con l'ID thread effettivo dell'intestazione della x-ms-thread-id risposta precedente.

Risposta di esempio:

Python is often considered the best choice for beginners among those three. Its clean syntax reads almost like English, making it easier to learn programming concepts without getting overwhelmed by complex syntax. It's also versatile and widely used in education.

Si noti che l'agente memorizza il contesto del messaggio precedente (i tre linguaggi di programmazione) senza doverli specificare di nuovo. Poiché lo stato della conversazione viene archiviato in modo permanente dall'Utilità di pianificazione attività durevole, questa cronologia viene mantenuta anche se si riavvia l'app per le funzioni o la conversazione viene ripresa da un'istanza diversa.

Monitorare con il dashboard di Durable Task Scheduler

Durable Task Scheduler fornisce un dashboard predefinito per il monitoraggio e il debug degli agenti durevoli. Il dashboard offre una visibilità approfondita sulle operazioni dell'agente, sulla cronologia delle conversazioni e sul flusso di esecuzione.

Accedere al dashboard

  1. Aprire il dashboard per l'emulatore DTS locale all'indirizzo http://localhost:8082 nel Web browser.

  2. Selezionare l'hub attività predefinito dall'elenco per visualizzarne i dettagli.

  3. Selezionare l'icona a forma di ingranaggio nell'angolo in alto a destra per aprire le impostazioni e assicurarsi che l'opzione Abilita pagine agente in Funzionalità di anteprima sia selezionata.

Esplorare le conversazioni degli agenti

  1. Nel dashboard passare alla scheda Agenti .

  2. Seleziona il thread dell'agente durevole, ad esempio mydurableagent - 263fa373-fa01-4705-abf2-5a114c2bb87d, dall'elenco.

    Verrà visualizzata una visualizzazione dettagliata del thread dell'agente, inclusa la cronologia completa della conversazione con tutti i messaggi e le risposte.

    Screenshot del dashboard Utilità di pianificazione attività permanenti che mostra la cronologia delle conversazioni di un thread dell'agente.

Il dashboard offre una visualizzazione sequenza temporale che consente di comprendere il flusso della conversazione. Le informazioni chiave includono:

  • Timestamp e durata per ogni interazione
  • Contenuto della richiesta e della risposta
  • Numero di token usati

Tip

Il dashboard DTS fornisce aggiornamenti in tempo reale, in modo da poter controllare il comportamento dell'agente durante l'interazione con esso tramite gli endpoint HTTP.

Distribuzione su Azure

Dopo aver testato l'agente durevole in locale, distribuirlo in Azure.

  1. Distribuire l'applicazione:

    azd deploy

    Questo comando pacchettizza la tua applicazione e la distribuisce nell'app Funzioni di Azure creata durante il provisioning.

  2. Attendere il completamento della distribuzione. L'output confermerà quando l'agente sta funzionando in Azure.

Testare l'agente distribuito

Dopo la distribuzione, testare l'agente in esecuzione in Azure.

Ottenere il tasto funzione

Funzioni di Azure richiede una chiave API per le funzioni attivate da HTTP nell'ambiente di produzione:

API_KEY=`az functionapp function keys list --name $(azd env get-value AZURE_FUNCTION_NAME) --resource-group $(azd env get-value AZURE_RESOURCE_GROUP) --function-name http-MyDurableAgent --query default -o tsv`

Avviare una nuova conversazione in Azure

Creare un nuovo thread e inviare il primo messaggio all'agente distribuito:

curl -i -X POST "https://$(azd env get-value AZURE_FUNCTION_NAME).azurewebsites.net/api/agents/MyDurableAgent/run?code=$API_KEY" \
  -H "Content-Type: text/plain" \
  -d "What are three popular programming languages?"

Prendere nota dell'ID del thread restituito nell'intestazione della x-ms-thread-id risposta.

Continuare la conversazione in Azure

Inviare un messaggio di completamento nello stesso thread. Sostituire <thread-id> con l'ID del thread della risposta precedente:

THREAD_ID="<thread-id>"
curl -X POST "https://$(azd env get-value AZURE_FUNCTION_NAME).azurewebsites.net/api/agents/MyDurableAgent/run?code=$API_KEY&thread_id=$THREAD_ID" \
  -H "Content-Type: text/plain" \
  -d "Which is easiest to learn?"

L'agente gestisce il contesto della conversazione in Azure esattamente come in locale, dimostrando la durabilità dello stato dell'agente.

Monitorare l'agente distribuito

È possibile monitorare l'agente distribuito usando il dashboard dell'Utilità di pianificazione attività permanenti in Azure.

  1. Ottenendo il nome di un'istanza dello Scheduler delle attività durevoli:

    azd env get-value DTS_NAME

  2. Aprire il portale di Azure e cercare il nome del Pianificatore di attività durevoli dal passaggio precedente.

  3. Nel pannello di panoramica della risorsa Utilità di pianificazione attività permanenti, selezionare l'hub attività predefinito dall'elenco.

  4. Selezionare Apri dashboard nella parte superiore della pagina dell'hub attività per aprire il dashboard di monitoraggio.

  5. Visualizzare le conversazioni dell'agente esattamente come è stato fatto con l'emulatore locale.

Il dashboard ospitato in Azure offre le stesse funzionalità di debug e monitoraggio dell'emulatore locale, consentendo di esaminare la cronologia delle conversazioni, le chiamate agli strumenti di traccia e analizzare le prestazioni nell'ambiente di produzione.

Esercitazione: Orchestrare agenti durevoli con Funzioni di Azure

Questa esercitazione illustra come orchestrare più agenti di intelligenza artificiale durevoli usando il modello di hosting Funzioni di Azure e il modello fan-out/fan-in. Si estenderà l'agente durevole dell'esercitazione precedente per creare un sistema multi-agente che elabora la domanda di un utente, quindi converte la risposta in più lingue contemporaneamente. Per esempi di orchestrazione self-hosted, vedere gli esempi.

Informazioni sul modello di orchestrazione

L'orchestrazione che verrà compilata segue questo flusso:

  1. Input dell'utente : domanda o messaggio dell'utente
  2. Agente principale : il MyDurableAgent dalla prima esercitazione elabora la domanda
  3. Fan-out : la risposta dell'agente principale viene inviata simultaneamente a entrambi gli agenti di traduzione
  4. Agenti di traduzione - Due agenti specializzati traducono la risposta (francese e spagnolo)
  5. Fan-in : i risultati vengono aggregati in una singola risposta JSON con la risposta e le traduzioni originali

Questo modello consente l'elaborazione simultanea, riducendo il tempo di risposta totale rispetto alla conversione sequenziale.

Registrare gli agenti all'avvio

Per usare correttamente gli agenti nelle orchestrazioni durevoli, registrarli all'avvio dell'applicazione. Possono essere usati nelle esecuzioni di orchestrazione.

Aggiorna il tuo Program.cs per registrare gli agenti di traduzione insieme agli esistenti 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();

Aggiorna il tuo function_app.py per registrare gli agenti di traduzione insieme agli esistenti 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])

Creare una funzione di orchestrazione

Una funzione di orchestrazione coordina il flusso di lavoro tra più agenti. Recupera gli agenti registrati dal contesto persistente e ne orchestra l'esecuzione, prima chiamando l'agente principale e poi coinvolgendo gli agenti di traduzione in parallelo.

Creare un nuovo file denominato AgentOrchestration.cs nella directory del progetto:

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

Aggiungere la funzione di orchestrazione al function_app.py file:

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

Testare l'orchestrazione

Verificare che le dipendenze di sviluppo locale della prima esercitazione siano ancora in esecuzione:

  • Azurite in una finestra terminale
  • Emulatore di Durable Task Scheduler in un'altra finestra del terminale

Con le dipendenze di sviluppo locale attive:

  1. Avviare l'app Funzioni di Azure in una nuova finestra del terminale:

    func start

  2. L'estensione Durable Functions crea automaticamente endpoint HTTP predefiniti per la gestione delle orchestrazioni. Avviare l'orchestrazione usando l'API predefinita:

    curl -X POST http://localhost:7071/runtime/webhooks/durabletask/orchestrators/agent_orchestration_workflow \
  -H "Content-Type: application/json" \
  -d '"\"What are three popular programming languages?\""'

  1. La risposta include gli URL per la gestione dell'istanza di orchestrazione:

    {
  "id": "abc123def456",
  "statusQueryGetUri": "http://localhost:7071/runtime/webhooks/durabletask/instances/abc123def456",
  "sendEventPostUri": "http://localhost:7071/runtime/webhooks/durabletask/instances/abc123def456/raiseEvent/{eventName}",
  "terminatePostUri": "http://localhost:7071/runtime/webhooks/durabletask/instances/abc123def456/terminate",
  "purgeHistoryDeleteUri": "http://localhost:7071/runtime/webhooks/durabletask/instances/abc123def456"
}

  2. Interrogare lo stato dell'orchestrazione usando statusQueryGetUri, sostituendo abc123def456 con l'ID effettivo dell'istanza.

    curl http://localhost:7071/runtime/webhooks/durabletask/instances/abc123def456

  1. Eseguire il polling dell'endpoint di stato fino a quando runtimeStatus diventa Completed. Al termine, verrà visualizzato l'output dell'orchestrazione con la risposta dell'agente principale e le relative traduzioni:

    {
  "name": "agent_orchestration_workflow",
  "instanceId": "abc123def456",
  "runtimeStatus": "Completed",
  "output": {
    "original": "Three popular programming languages are Python, JavaScript, and Java. Python is known for its simplicity...",
    "french": "Trois langages de programmation populaires sont Python, JavaScript et Java. Python est connu pour sa simplicité...",
    "spanish": "Tres lenguajes de programación populares son Python, JavaScript y Java. Python es conocido por su simplicidad..."
  }
}

Monitorare l'orchestrazione nel dashboard

Il dashboard di Durable Task Scheduler offre una visibilità sull'orchestrazione:

  1. Aprire http://localhost:8082 nel browser.

  2. Selezionare l'hub attività "predefinito".

  3. Selezionare la scheda "Orchestrazioni".

  4. Trova l'istanza di orchestrazione nell'elenco.

  5. Selezionare l'istanza da visualizzare:

    • Sequenza temporale dell'orchestrazione
    • Esecuzione dell'agente principale seguita da agenti di conversione simultanei
    • Ogni esecuzione dell'agente (MyDurableAgent, seguita dai traduttori francesi e spagnoli)
    • I modelli fan-out e fan-in visualizzati
    • Intervallo e durata per ogni passaggio

Distribuire l'orchestrazione in Azure

Distribuire l'applicazione aggiornata usando l'interfaccia della riga di comando per sviluppatori di Azure:

azd deploy

In questo modo il codice aggiornato viene distribuito con la nuova funzione di orchestrazione e gli agenti aggiuntivi nell'app Funzioni di Azure creata nella prima esercitazione.

Testare l'orchestrazione implementata

Dopo la distribuzione, testa la tua orchestrazione in esecuzione in Azure.

  1. Ottenere la chiave di sistema per l'estensione durevole:

    SYSTEM_KEY=$(az functionapp keys list --name $(azd env get-value AZURE_FUNCTION_NAME) --resource-group $(azd env get-value AZURE_RESOURCE_GROUP) --query "systemKeys.durabletask_extension" -o tsv)

  1. Avviare l'orchestrazione usando l'API predefinita:

    curl -X POST "https://$(azd env get-value AZURE_FUNCTION_NAME).azurewebsites.net/runtime/webhooks/durabletask/orchestrators/agent_orchestration_workflow?code=$SYSTEM_KEY" \
  -H "Content-Type: application/json" \
  -d '"\"What are three popular programming languages?\""'
  1. Usare il statusQueryGetUri dalla risposta per effettuare il polling per il completamento e visualizzare i risultati con le traduzioni.

Passaggi successivi

Endpoint compatibili con OpenAI

Risorse aggiuntive:

  • Last updated on