Extensión durable

La extensión durable para Microsoft Agent Framework aporta una ejecución duradera a agentes, orquestaciones multiagente y flujos de trabajo del marco de agente de Microsoft. Puede usarlo para conservar las sesiones del agente, la orquestación de puntos de control y el progreso del flujo de trabajo, recuperarse de errores y escalar el trabajo entre hosts distribuidos sin cambiar la lógica del agente principal.

La extensión admite dos modelos de hospedaje en C# y Python:

  • Azure Functions para el hospedaje administrado sin servidor con el modelo de programación de Azure Functions.
  • Traiga su propio proceso o autohospedado para ejecutar agentes y flujos de trabajo duraderos en su propio proceso de trabajo, servicio, contenedor, entorno de Kubernetes o infraestructura de aplicaciones existente.

Overview

Los agentes duraderos combinan el modelo de programación de Agent Framework con la infraestructura de Durable Task, como durable Task Scheduler, para crear agentes que:

  • Conservar el estado automáticamente entre las solicitudes y las ejecuciones de trabajo
  • Reanudación después de errores sin perder el contexto de conversación o repetir el trabajo completado
  • Escalado entre trabajos distribuidos y sin estado en función de la demanda
  • Orquestación de flujos de trabajo de varios agentes con garantías de ejecución confiables
  • Flujos de trabajo de Checkpoint Agent Framework creados con el modelo de flujo de trabajo basado en grafos
  • Pausar para eventos externos o de entrada humana sin consumir tokens de proceso o modelo mientras espera
  • Transmitir respuestas de forma confiable cuando se configuran con un agente de flujo confiable, como Redis
  • Administración del ciclo de vida de la sesión con la limpieza del período de vida (TTL) y la supervisión basada en paneles

Cuándo usar agentes duraderos

Elija agentes duraderos cuando necesite:

  • Estado de conversación persistente: las sesiones del agente sobrevive a los bloqueos del proceso, los reinicios y los eventos de escalabilidad horizontal.
  • Orquestaciones complejas: coordina varios agentes con flujos de trabajo deterministas y confiables que se pueden ejecutar durante días o semanas
  • Orquestación controlada por eventos: integración con desencadenadores, colas, webhooks, temporizadores o eventos de aplicación existentes
  • Estado automático de la conversación: el historial de conversaciones del agente se administra automáticamente y se conserva sin necesidad de controlar el estado explícito en el código.
  • Flujos de trabajo del marco de agentedurable: hacer que los flujos de trabajo del marco del agente Microsoft basados en grafos sean duraderos para que cada paso se pueda controlar y reanudar.
  • Sesiones de larga duración: mantenga las conversaciones útiles disponibles al usar la limpieza del período de vida (TTL) de la sesión para quitar automáticamente las sesiones inactivas.
  • Respuestas confiables en tiempo real: transmisión de salida de token duradera para aplicaciones que necesitan experiencia de usuario en tiempo real con garantías de entrega

Este enfoque de hospedaje difiere del hospedaje de agentes administrados basados en servicios (como foundry Agent Service), que proporciona una infraestructura totalmente administrada sin necesidad de implementar o administrar hosts de trabajo. Los agentes duraderos son ideales cuando se necesita la flexibilidad de la implementación en primer lugar del código combinada con la administración de estado duradera.

Elección de un modelo de hospedaje

Modelo de alojamiento Selecciónela cuando necesite
Funciones de Azure Un modelo de hospedaje administrado y sin servidor; escalabilidad horizontal integrada y escalado horizontal a cero; Azure Functions desencadenadores y enlaces; Puntos de conexión HTTP generados por el modelo de programación de Functions; el desencadenador del servidor MCP; y la administración mínima de la infraestructura de host.
Bring-your-own-compute/autohospedado Más control sobre el proceso de host, el entorno de implementación, el ciclo de vida del entorno de ejecución, la infraestructura, las redes, la autenticación o la integración con una aplicación o servicio existente. Use este modelo para contenedores, Kubernetes, trabajos de larga duración, aplicaciones de consola, servicios personalizados o entornos de hospedaje que no sean de Functions.

Cuando se hospeda en el plan de hospedaje Azure Functions Flex Consumption, los agentes pueden escalar hasta miles de instancias o reducirse a cero instancias cuando no están en uso, lo que le permite pagar solo por la capacidad de cálculo que necesita. En escenarios autohospedados, sus propios controles host procesan la duración, el escalado, las redes y la implementación.

Cómo empezar

En un proyecto de .NET, elija el conjunto de paquetes para el modelo de hospedaje.

Para Azure Functions hospedaje, agregue el paquete de integración Azure Functions y los paquetes de trabajo de 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

Note

Además de estos paquetes, asegúrese de que el proyecto usa la versión 2.2.0 o posterior del paquete Microsoft.Azure.Functions.Worker .

Para el hospedaje bring-your-own-compute, agregue el paquete de integración de Durable Task base y los paquetes de trabajo o cliente de Durable Task Scheduler usados por el 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

En un proyecto de Python, elija el paquete para el modelo de hospedaje.

Para Azure Functions hospedaje, instale el paquete de integración de Azure Functions.

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

Para el hospedaje bring-your-own-compute, instale el paquete de integración de Durable Task.

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

hospedaje de Azure Functions

Con la extensión durable, puede implementar y hospedar agentes de Microsoft Agent Framework en Azure Functions con puntos de conexión HTTP integrados y invocación basada en orquestación. Azure Functions proporciona precios de pago por invocación controlados por eventos con escalado automático y administración de infraestructura mínima.

Al configurar un agente duradero en Azure Functions, la extensión crea automáticamente puntos de conexión HTTP para el agente y administra la infraestructura subyacente para almacenar el estado de conversación, controlar las solicitudes simultáneas y coordinar flujos de trabajo de varios agentes. La integración de hospedaje de Azure Functions también proporciona comodidades específicas de Functions, como las API REST generadas para enviar mensajes, comprobar el estado y administrar sesiones, además de desencadenadores como el desencadenador de servidor MCP para agentes de hospedaje como servidores MCP sin escribir pegamento de desencadenador.

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

Warning

DefaultAzureCredential es conveniente para el desarrollo, pero requiere una consideración cuidadosa en producción. En producción, considere usar una credencial específica (por ejemplo, ManagedIdentityCredential) para evitar problemas de latencia, sondeos de credenciales no deseados y posibles riesgos de seguridad de los mecanismos de respaldo.

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

Bring-your-own-compute/autohospedado

Use el hospedaje bring-your-own-compute cuando desee las funcionalidades de Durable Extension sin usar el modelo de programación de Azure Functions. En este modelo, el proceso inicia un trabajo de Durable Task, registra agentes o flujos de trabajo duraderos y se conecta a un back-end del Programador de tareas durables. El código de cliente se puede ejecutar en el mismo proceso o en un servicio independiente.

Los trabajos autohospedados usan las mismas funcionalidades principales de Durable Extension que Azure Functions hospedaje: punto de control y reanudación, orquestación de agentes deterministas, flujos de trabajo durables del marco del agente, esperas humanas en el bucle, streaming confiable, limpieza de sesión inactiva, visibilidad del panel y ejecución distribuida en instancias de trabajo sin estado. El host es responsable de exponer sus propias API, administración del ciclo de vida, redes, autenticación y modelo de implementación.

Configure el host con el paquete de integración de Durable Task base. Use ConfigureDurableAgents para agentes duraderos y ConfigureDurableWorkflows para flujos de trabajo de Microsoft Agent Framework basados en grafos.

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

Consulte los ejemplos de consola de .NET Durable Agents y .NET ejemplos de consola de Durable Workflows para ver ejemplos autohospedados ejecutables.

Use el paquete de integración de Durable Task para ejecutar un proceso de trabajo que registra agentes y escucha las solicitudes. El código de cliente puede conectarse al mismo centro de tareas de Durable Task Scheduler desde otro proceso.

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

Consulte los ejemplos de Python Durable Task para obtener ejemplos de cliente de trabajo, incluido el hospedaje de un solo agente, el enrutamiento multiagente, el streaming confiable, el encadenamiento de orquestaciones, la simultaneidad, los condicionales y los patrones de bucle humano en el bucle.

Flujos de trabajo de Durable Agent Framework

La durabilidad no se limita a las orquestaciones duraderas. Microsoft los flujos de trabajo de Agent Framework creados con el modelo de flujo de trabajo basado en grafos también se pueden hacer duraderos. La ejecución del flujo de trabajo de los puntos de control de extensión durable, por lo que los pasos completados del ejecutor y del agente no se repiten después de un reinicio del proceso o un error.

Use orquestaciones duraderas cuando desee la coordinación imperativa con la bifurcación, temporizadores, actividades y eventos externos basados en código. Use flujos de trabajo durables de Agent Framework cuando desee un grafo declarativo de ejecutores y agentes con enrutamiento tipado, distribución de distribución/distribución ramificada, bordes condicionales, eventos de flujo de trabajo, estado compartido, subprocesos o puertos de solicitud humanos en el bucle.

Note

Los flujos de trabajo de Durable Agent Framework son diferentes del almacenamiento de puntos de control en flujos de trabajo estándar. El almacenamiento de puntos de control ayuda a reanudar una ejecución de flujo de trabajo en el entorno de ejecución de Agent Framework. La extensión durable ejecuta el flujo de trabajo en la infraestructura de Durable Task para que el progreso del flujo de trabajo se controle y recupere entre los trabajos duraderos distribuidos. Para obtener puntos de control de flujo de trabajo estándar, consulte Puntos de comprobación y reanudación.

Registre flujos de trabajo basados en grafos con ConfigureDurableWorkflows para aplicaciones autohospedadas o ConfigureDurableWorkflows en el generador de aplicaciones de Functions para Azure Functions hospedaje.

Consulte los ejemplos de Azure Functions de .NET Durable Workflows y .NET ejemplos de consola de Durable Workflows.

Los ejemplos de flujo de trabajo duraderos están disponibles para Azure Functions hospedaje, incluido el estado compartido, sin estado compartido, ejecución de flujo de trabajo paralelo y flujos de trabajo humanos en bucle.

Consulte los ejemplos de Python Azure Functions para obtener ejemplos de agente duradero, orquestación, servidor MCP y flujo de trabajo.

Samples

Lenguaje Modelo de alojamiento Samples
C# Azure Functions .NET Durable Agents - Azure Functions, .NET Durable Workflows - Azure Functions
C# Bring-your-own-compute/autohospedado .NET Durable Agents: aplicaciones de consola, .NET flujos de trabajo duraderos: aplicaciones de consola
Python Azure Functions ejemplos de Python Azure Functions
Python Bring-your-own-compute/autohospedado Python ejemplos de Durable Task

Subprocesos de agente con estado con historial de conversaciones

Los agentes mantienen hilos persistentes que persisten a lo largo de múltiples interacciones. Cada subproceso se identifica mediante un identificador de subproceso único y almacena el historial de conversación completo en el almacenamiento duradero administrado por la infraestructura de Durable Task, como durable Task Scheduler.

Este patrón permite la continuidad conversacional en la que se conserva el estado del agente a través de fallos y reinicios del proceso, lo que permite mantener el historial de conversaciones completo en los hilos de usuario. El almacenamiento duradero garantiza que incluso si un proceso de host se reinicia o se reanuda el trabajo en una instancia de trabajo diferente, la conversación continúa sin problemas desde donde se dejó.

Use la limpieza del período de vida (TTL) de la sesión para las cargas de trabajo que necesitan continuidad duradera durante el uso activo, pero que deben limpiar automáticamente las conversaciones inactivas. La limpieza basada en TTL impide que las sesiones sin usar y el historial de conversaciones se acumulan indefinidamente mientras se conserva el estado de sesión activo.

En el ejemplo de Azure Functions siguiente se muestran varias solicitudes HTTP al mismo subproceso, que muestran cómo persiste el contexto de conversación. En las aplicaciones autohospedadas, use las API de cliente de Durable Task desde su propio proceso o servicio.

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

El estado del agente se mantiene en almacenamiento duradero, lo que permite la ejecución distribuida en varias instancias. Cualquier instancia puede reanudar la ejecución de un agente después de interrupciones o errores, lo que garantiza la operación continua.

Streaming confiable

Durable Extension admite el streaming confiable para aplicaciones que necesitan entrega de tokens en tiempo real con garantías de entrega duraderas. El streaming se puede usar con la extensión principal en ambos modelos de hospedaje, pero los hosts distribuidos necesitan un agente de flujo confiable, como Redis, por lo que las secuencias de token se pueden entregar de forma coherente en los reinicios del proceso, las reconectaciones o los cambios de trabajo.

Use el streaming confiable cuando la experiencia del usuario dependa de las respuestas incrementales, pero la carga de trabajo todavía necesita semántica de ejecución duradera. Para ver ejemplos ejecutables, consulte los ejemplos de Python Durable Task, que incluyen patrones de streaming confiables.

Orquestaciones de varios agentes deterministas

Durable Extension admite la creación de flujos de trabajo deterministas que coordinan varios agentes mediante orquestaciones de Durable Task. En Azure Functions, estos usan Durable Functions orquestaciones; en los hosts bring-your-own-compute, se ejecutan a través del trabajo de Durable Task y el cliente que configure.

Las orquestaciones son flujos de trabajo basados en código que coordinan varias operaciones (como llamadas de agente, llamadas API externas o temporizadores) de forma confiable. Determinista significa que el código de orquestación se ejecuta de la misma manera cuando se reproduce después de un fallo. Esto hace que los flujos de trabajo sean confiables y fáciles de depurar. Al reproducir el historial de una orquestación, se puede ver exactamente qué ocurrió en cada paso.

Las orquestaciones se ejecutan de forma confiable, sobreviviendo a fallos entre las llamadas al agente, y proporcionan procesos predecibles y repetibles. Esto hace que sean ideales para escenarios complejos de varios agentes en los que se necesita un orden de ejecución garantizado y tolerancia a errores.

Orquestaciones secuenciales

En el patrón de multiagente secuencial, los agentes especializados se ejecutan en un orden específico, donde la salida de cada uno puede influir en la ejecución del siguiente. Este patrón admite la lógica condicional y la bifurcación en función de las respuestas del agente.

Al usar agentes en orquestaciones, debe usar la API context.GetAgent() para obtener una instancia de DurableAIAgent, que es una subclase especial del tipo estándar AIAgent que encapsula alguno de tus agentes registrados. El DurableAIAgent envoltorio garantiza que el marco de orquestación durable realiza el seguimiento correcto de las llamadas al agente y marca los puntos de control.

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

Al usar agentes en orquestaciones, debe usar el método app.get_agent() para obtener una instancia de agente persistente, que es un contenedor especial de uno de sus agentes registrados. El envoltorio del agente duradero asegura que el marco de orquestación duradero realiza un seguimiento correcto de las llamadas al agente y los puntos de control.

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

Las orquestaciones coordinan el trabajo a través de múltiples agentes, haciendo frente a los fallos entre las llamadas de un agente a otro. El contexto de orquestación proporciona métodos para recuperar e interactuar con agentes hospedados dentro de orquestaciones.

Orquestaciones paralelas

En el patrón multiagente paralelo, se ejecutan varios agentes simultáneamente y, a continuación, se agregan sus resultados. Este patrón es útil para recopilar diversas perspectivas o procesar subtareas independientes simultáneamente.

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

Se realiza un seguimiento de la ejecución en paralelo mediante una lista de tareas. La creación automática de puntos de control garantiza que las ejecuciones de agentes completadas no se repitan ni se pierdan si se produce una falla durante la agregación.

Orquestaciones humanas en bucle

Las orquestaciones de agentes deterministas pueden detenerse para intervención, aprobación o revisión humana sin consumir recursos de proceso. La ejecución duradera permite que las orquestaciones esperen días o incluso semanas mientras esperan respuestas humanas. Cuando se combina con el hospedaje sin servidor, todos los recursos de proceso se desactivan durante el período de espera, eliminando los costos de proceso hasta que la persona proporciona su intervención.

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

Las orquestaciones de agentes deterministas pueden esperar eventos externos, manteniendo persistentemente su estado mientras esperan comentarios humanos, y pueden sobrevivir a fallas, reinicios y períodos de espera prolongados. Cuando llega la respuesta humana, la orquestación se reanuda automáticamente con el contexto de conversación completo y el estado de ejecución intactos.

Proporcionar entradas humanas

Para enviar la aprobación o la entrada a una orquestación en espera, genere un evento externo a la instancia de orquestación mediante el SDK de cliente de Durable Task o los puntos de conexión de extensión durable de Azure Functions. Por ejemplo, un revisor podría aprobar el contenido a través de un formulario web que llama a:

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)

Rentabilidad

Los flujos de trabajo humanos en bucle con agentes duraderos son extremadamente rentables cuando se hospedan en el plan de consumo flexible de Azure Functions. Para un flujo de trabajo que espera 24 horas de aprobación, solo paga por unos segundos de tiempo de ejecución (el tiempo para generar contenido, enviar notificación y procesar la respuesta), no las 24 horas de espera. Durante el período de espera, no se consumen recursos de proceso.

Observabilidad con el programador de tareas durables

El Programador de Tareas Duradero (DTS) es el backend duradero recomendado para tus agentes duraderos, ya que ofrece el mejor rendimiento, infraestructura totalmente gestionada y observabilidad integrada a través de un panel de control de interfaz de usuario. Azure Functions aplicaciones pueden usar otros back-end de almacenamiento (como Azure Storage), pero DTS está optimizado específicamente para cargas de trabajo duraderas y proporciona funcionalidades de rendimiento y supervisión superiores. Los trabajos autohospedados también usan DTS para la programación duradera, el estado y la visibilidad del panel.

Información de sesión del agente

  • Historial de conversaciones: vea el historial de chat completo para cada sesión del agente, incluidos todos los mensajes, las llamadas a herramientas y el contexto de conversación en cualquier momento.
  • Tiempo de tarea: supervise cuánto tiempo tardan las tareas específicas y las interacciones del agente en completarse.

Captura de pantalla del panel del programador de Durable Task que muestra el historial de chat del agente con los subprocesos de conversación y los mensajes.

Información de orquestación

  • Visualización de varios agentes: vea el flujo de ejecución al llamar a varios agentes especializados con representación visual de ejecuciones paralelas y bifurcación condicional.
  • Historial de ejecución: acceso a registros de ejecución detallados
  • Monitoreo en tiempo real: monitorizar las orquestaciones activas, los elementos de trabajo pendientes y los estados del agente en toda la implementación
  • Métricas de rendimiento: supervisar los tiempos de respuesta del agente, el uso de tokens y la duración de la orquestación

Captura de pantalla del panel del Planificador de tareas duraderas que muestra la visualización de la orquestación con varias interacciones de agentes y ejecución del flujo de trabajo.

Funcionalidades de depuración

  • Visualización de las salidas del agente estructurado y los resultados de la llamada a la herramienta
  • Invocaciones de la herramienta de seguimiento y sus resultados
  • Supervisión del manejo de eventos externos para escenarios con intervención humana

El panel le permite comprender exactamente lo que hacen los agentes, diagnosticar problemas rápidamente y optimizar el rendimiento en función de los datos de ejecución reales.

Tutorial: Creación y ejecución de un agente duradero con Azure Functions

En este tutorial se muestra cómo crear y ejecutar un agente de IA duradero mediante el modelo de hospedaje de Azure Functions para la extensión durable. Creará una aplicación de Azure Functions que hospeda un agente con estado con puntos de conexión HTTP integrados y aprenderá a supervisarla mediante el panel del Programador de tareas durable. Para los agentes autohospedados, consulte los ejemplos.

Prerequisites

Antes de comenzar, asegúrese de que tiene los siguientes requisitos previos:

Note

Microsoft Agent Framework es compatible con todas las versiones admitidas activamente de .NET. Para los fines de este ejemplo, se recomienda el SDK de .NET 9 o una versión posterior.

Descarga del proyecto de inicio rápido

Use la CLI para desarrolladores de Azure para inicializar un nuevo proyecto a partir de la plantilla de inicio rápido de durable agents.

  1. Cree un directorio para el proyecto y vaya a él:

    mkdir MyDurableAgent
cd MyDurableAgent

  1. Inicialice el proyecto a partir de la plantilla:

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

    Cuando se le solicite un nombre de entorno, escriba un nombre como my-durable-agent.

Esto descarga el proyecto de inicio rápido con todos los archivos necesarios, incluida la configuración de Azure Functions, el código del agente y la infraestructura como plantillas de código.

  1. Cree un directorio para el proyecto y vaya a él:

    mkdir MyDurableAgent
cd MyDurableAgent

  1. Inicialice el proyecto a partir de la plantilla:

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

    Cuando se le solicite un nombre de entorno, escriba un nombre como my-durable-agent.

  2. Cree y active un entorno virtual:

    uv venv .venv
source .venv/bin/activate

Note

python3 -m venv .venv también funciona, pero puede bloquearse indefinidamente en Windows con Microsoft Store Python debido a un problema conocido de ensurepip. Use uv venv .venv para evitar esto.

  1. Instale los paquetes necesarios:

    python -m pip install -r requirements.txt

Esto descarga el proyecto de inicio rápido con todos los archivos necesarios, incluida la configuración de Azure Functions, el código del agente y la infraestructura como plantillas de código. También prepara un entorno virtual con las dependencias necesarias.

Aprovisionamiento de los recursos de Azure

Use la CLI para desarrolladores de Azure para crear los recursos de Azure necesarios para el agente duradero.

  1. Prepare la infraestructura:

    azd provision

    Este comando crea:

    • Un servicio de Azure OpenAI con una implementación gpt-4o-mini
    • Una aplicación de Azure Functions con un plan de hospedaje de consumo flexible
    • Una cuenta de Azure Storage para el entorno de ejecución de Azure Functions y el almacenamiento duradero
    • Una instancia del Programador de tareas duradera (plan de consumo) para administrar el estado del agente
    • Configuraciones de identidad y redes necesarias

  2. Cuando se le solicite, seleccione la suscripción de Azure y elija una ubicación para los recursos.

El proceso de aprovisionamiento tarda unos minutos. Una vez completado, azd almacena la información de recursos creada en el entorno.

Revisión del código del agente

Ahora vamos a examinar el código que define el agente duradero.

Abra Program.cs para ver la configuración del 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();

Este código:

  1. Recupera la configuración de Azure OpenAI de las variables de entorno.
  2. Crea un cliente de Azure OpenAI mediante credenciales de Azure.
  3. Crea un agente de IA con instrucciones y un nombre.
  4. Configura la aplicación de Azure Functions para hospedar el agente con una administración duradera de subprocesos.

Abra function_app.py para ver la configuración del 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])

Este código:

  • Recupera la configuración de Azure OpenAI de las variables de entorno.
  • Crea un cliente de Azure OpenAI mediante credenciales de Azure.
  • Crea un agente de IA con instrucciones y un nombre.
  • Configura la aplicación de Azure Functions para hospedar el agente con una administración duradera de subprocesos.

El agente ya está listo para hospedarse en Azure Functions. La extensión de tarea duradera crea automáticamente puntos de conexión HTTP para interactuar con el agente y administra el estado de conversación en varias solicitudes.

Configuración de valores locales

Cree un local.settings.json archivo para el desarrollo local basado en el archivo de ejemplo incluido en el proyecto.

  1. Copie el archivo de configuración de ejemplo:

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

  1. Obtenga el punto de conexión de Azure OpenAI de los recursos aprovisionados:

    azd env get-value AZURE_OPENAI_ENDPOINT

  2. Abra local.settings.json y reemplace <your-resource-name> en AZURE_OPENAI_ENDPOINT con el punto de conexión del comando anterior.

Su local.settings.json debería lucir de esta manera:

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

Note

El local.settings.json archivo solo se usa para el desarrollo local y no se implementa en Azure. En el caso de las implementaciones de producción, estas opciones se configuran automáticamente en la aplicación de Azure Functions mediante las plantillas de infraestructura.

Inicio de dependencias de desarrollo local

Para ejecutar agentes persistentes localmente, debe poner en marcha dos servicios:

  • Azurite: emula los servicios de Azure Storage (usados por Azure Functions para administrar desencadenadores y estado interno).
  • Emulador de Durable Task Scheduler (DTS): gestiona el estado duradero (historial de conversaciones, estado de orquestación) y la programación de tus agentes.

Iniciar Azurite

Azurite emula los servicios de Azure Storage localmente. Azure Functions lo usa para administrar el estado interno. Deberá ejecutarlo en una nueva ventana de terminal y mantenerla en ejecución mientras desarrolla y prueba el agente duradero.

  1. Abra una nueva ventana de terminal y extraiga la imagen de Docker de Azurite:

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

  2. Inicie Azurite en una ventana de terminal:

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

    Azurite se iniciará y escuchará en los puertos predeterminados para los servicios Blob (10000), Queue (10001) y Table (10002).

Mantenga abierta esta ventana de terminal mientras desarrolla y prueba el agente duradero.

Tip

Para más información sobre Azurite, incluidos los métodos de instalación alternativos, consulte Uso del emulador de Azurite para el desarrollo local de Azure Storage.

Iniciar el emulador del Programador de tareas duraderas

El emulador de DTS proporciona el back-end duradero para administrar el estado y las orquestaciones del agente. Almacena el historial de conversaciones y garantiza que el estado del agente persiste a través de los reinicios. También desencadena orquestaciones y agentes duraderos. Tendrá que ejecutar esto en una nueva ventana de terminal independiente y mantenerla en ejecución mientras desarrolla y prueba su agente duradero.

  1. Abra otra nueva ventana de terminal y extraiga la imagen de Docker del emulador de DTS:

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

  2. Ejecute el emulador de DTS:

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

    Este comando inicia el emulador y expone:

    • Puerto 8080: el punto de conexión de gRPC para el programador de tareas durables (usado por la aplicación de Functions)
    • Puerto 8082: Panel administrativo

  3. El panel estará disponible en http://localhost:8082.

Mantenga abierta esta ventana de terminal mientras desarrolla y prueba el agente duradero.

Tip

Para obtener más información sobre el emulador de DTS, incluida la configuración de varios centros de tareas y el acceso al panel, consulte Desarrollo con el programador de tareas durable.

Ejecución de la aplicación de funciones

Ahora está listo para ejecutar la aplicación de Azure Functions con el agente duradero.

  1. En una nueva ventana de terminal (mantener Azurite y el emulador de DTS en ejecución en ventanas independientes), vaya al directorio del proyecto.

  2. Inicie el entorno de ejecución de Azure Functions:

    func start

  3. Debería ver la salida que indica que la aplicación de funciones se está ejecutando, incluidos los puntos de conexión HTTP del agente:

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

Estos endpoints administran automáticamente el estado de conversación: usted no necesita crear ni administrar hilos.

Probar el agente localmente

Ahora puede interactuar con agentes duraderos a través de solicitudes HTTP. El agente mantiene el estado de la conversación a lo largo de varias solicitudes, lo que permite interacciones de varios turnos.

Iniciar una nueva conversación

Cree un nuevo hilo y envíe su primer mensaje:

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

Respuesta de ejemplo (tenga en cuenta que el x-ms-thread-id encabezado contiene el identificador de subproceso):

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.

Guarde el identificador de subproceso del encabezado x-ms-thread-id (por ejemplo, @dafx-mydurableagent@263fa373-fa01-4705-abf2-5a114c2bb87d) para la siguiente solicitud.

Continuar la conversación

Envíe un mensaje de seguimiento al mismo subproceso mediante la inclusión del identificador de subproceso como parámetro de consulta:

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?"

Reemplace @dafx-mydurableagent@263fa373-fa01-4705-abf2-5a114c2bb87d por el identificador de subproceso real del encabezado de la respuesta anterior x-ms-thread-id.

Respuesta de ejemplo:

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.

Observe que el agente recuerda el contexto del mensaje anterior (los tres lenguajes de programación) sin tener que volver a especificarlos. Dado que el Durable Task Scheduler almacena el estado de la conversación de forma duradera, este historial persiste incluso si reinicia la aplicación de funciones o la conversación se reanuda por una instancia diferente.

Supervisión con el panel de control del programador de tareas duraderas

Durable Task Scheduler proporciona un panel integrado para monitorizar y depurar tus agentes duraderos. El panel ofrece una visibilidad profunda de las operaciones del agente, el historial de conversaciones y el flujo de ejecución.

Acceso al panel

  1. Abra el panel de control del emulador de DTS local en su navegador web en http://localhost:8082.

  2. Seleccione el centro de tareas predeterminado de la lista para ver sus detalles.

  3. Seleccione el icono de engranaje de la esquina superior derecha para abrir la configuración y asegúrese de que la opción Habilitar páginas del Agente en Características de vista previa está seleccionada.

Explora conversaciones de agentes

  1. En el panel, vaya a la pestaña Agentes .

  2. Seleccione el subproceso del agente duradero (por ejemplo, mydurableagent - 263fa373-fa01-4705-abf2-5a114c2bb87d) de la lista.

    Verá una vista detallada del hilo del agente, incluido el historial de conversaciones completo con todos los mensajes y respuestas.

    Captura de pantalla del panel del programador de Durable Task que muestra el historial de conversaciones de un subproceso de agente.

El panel proporciona una vista de escala de tiempo para ayudarle a comprender el flujo de la conversación. La información clave incluye:

  • Marcas de tiempo y duración para cada interacción
  • Contenido de solicitud y respuesta
  • Número de tokens usados

Tip

El panel de DTS proporciona actualizaciones en tiempo real, por lo que puede ver el comportamiento del agente a medida que interactúa con él a través de los puntos de conexión HTTP.

Implementar en Azure

Ahora que ha probado el agente duradero localmente, impleméntelo en Azure.

  1. Implemente la aplicación:

    azd deploy

    Este comando empaqueta la aplicación e la implementa en la aplicación de Azure Functions creada durante el aprovisionamiento.

  2. Espere a que la implementación se complete. La salida confirmará que su agente se está ejecutando en Azure.

Prueba el agente implementado

Después de la implementación, pruebe el agente que se ejecuta en Azure.

Obtención de la clave de función

Azure Functions requiere una clave de API para las funciones desencadenadas por HTTP en producción:

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`

Inicio de una nueva conversación en Azure

Cree un nuevo hilo y envíe su primer mensaje al agente desplegado.

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?"

Anote el identificador de subproceso devuelto en el encabezado de respuesta x-ms-thread-id.

Continuar la conversación en Azure

Envía un mensaje de seguimiento en el mismo hilo. Reemplace <thread-id> con el identificador de subproceso de la respuesta anterior.

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?"

El agente mantiene el contexto de conversación en Azure igual que lo hace localmente, lo que demuestra la durabilidad del estado del agente.

Supervisión del agente implementado

Puede monitorear su agente implementado mediante el tablero de control del Durable Task Scheduler en Azure.

  1. Obtenga el nombre de la instancia de Durable Task Scheduler:

    azd env get-value DTS_NAME

  2. Abra el Portal de Azure y busque el nombre de "Durable Task Scheduler" en el paso anterior.

  3. En la hoja de información general del recurso del Programador de Tareas Durable, seleccione el hub de tareas predeterminado de la lista.

  4. Seleccione Abrir panel en la parte superior de la página del centro de tareas para abrir el panel de supervisión.

  5. Vea las conversaciones del agente como hizo con el emulador local.

El panel hospedado en Azure proporciona las mismas funcionalidades de depuración y supervisión que el emulador local, lo que le permite inspeccionar el historial de conversaciones, las llamadas a la herramienta de seguimiento y analizar el rendimiento en el entorno de producción.

Tutorial: Orquestación de agentes duraderos con Azure Functions

En este tutorial se muestra cómo organizar varios agentes de inteligencia artificial duraderos mediante el modelo de hospedaje de Azure Functions y el patrón de distribución y distribución ramificada. Extenderá el agente duradero del tutorial anterior para crear un sistema multiagente que procese la pregunta de un usuario y, a continuación, traducirá la respuesta a varios idiomas simultáneamente. Para ver ejemplos de orquestación autohospedado, consulte los ejemplos.

Descripción del patrón de orquestación

La orquestación que va a crear sigue este flujo:

  1. Entrada del usuario : pregunta o mensaje del usuario
  2. Agente principal - El MyDurableAgent del primer tutorial procesa la pregunta
  3. Fan-out : la respuesta del agente principal se envía simultáneamente a ambos agentes de traducción.
  4. Agentes de traducción : dos agentes especializados traducen la respuesta (francés y español)
  5. Entrada de entrada: los resultados se agregan en una única respuesta JSON con la respuesta original y las traducciones.

Este patrón permite el procesamiento simultáneo, lo que reduce el tiempo total de respuesta en comparación con la traducción secuencial.

Registro de agentes en el inicio

Para usar correctamente agentes en orquestaciones duraderas, regístrelos en el inicio de la aplicación. Se pueden usar en ejecuciones de orquestación.

Actualice su Program.cs para registrar los agentes de traducción junto con los existentes 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();

Actualice su function_app.py para registrar los agentes de traducción junto con los existentes 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])

Creación de una función de orquestación

Una función de orquestación coordina el flujo de trabajo entre varios agentes. Recupera los agentes registrados del contexto duradero y orquesta su ejecución, primero llamando al agente principal y luego distribuyendo a los agentes de traducción simultáneamente.

Cree un nuevo archivo denominado AgentOrchestration.cs en el directorio del proyecto:

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

Agregue la función de orquestación al function_app.py archivo:

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

Prueba la orquestación

Asegúrese de que las dependencias de desarrollo local del primer tutorial siguen ejecutándose:

  • Azurite en una ventana de terminal
  • Emulador del Programador de tareas durables en otra ventana de terminal

Con las dependencias de desarrollo local en ejecución:

  1. Inicie la aplicación de Azure Functions en una nueva ventana de terminal:

    func start

  2. La extensión Durable Functions crea automáticamente puntos de conexión HTTP integrados para administrar orquestaciones. Inicie la orquestación mediante la API integrada:

    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 respuesta incluye direcciones URL para administrar la instancia de orquestación:

    {
  "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. Consulte el estado de la orquestación usando statusQueryGetUri (sustituya abc123def456 por el ID real de su instancia):

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

  1. Sondea el punto de conexión de estado hasta que runtimeStatus sea Completed. Cuando haya finalizado, verá el resultado de la orquestación con la respuesta del agente principal y sus traducciones.

    {
  "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..."
  }
}

Supervise la orquestación en el tablero de control

El panel de Durable Task Scheduler otorga visibilidad de su orquestación:

  1. Abra http://localhost:8082 en el explorador.

  2. Seleccione el centro de tareas "predeterminado".

  3. Seleccione la pestaña "Orquestaciones".

  4. Busque la instancia de orquestación en la lista.

  5. Seleccione la instancia para ver:

    • Escala de tiempo de orquestación
    • Ejecución del agente principal seguida de agentes de traducción simultáneos
    • Cada ejecución del agente (MyDurableAgent, luego los traductores en francés y español)
    • Patrones de distribución en abanico y de concentración visualizados
    • Tiempo y duración de cada paso

Implementación de la orquestación en Azure

Implemente la aplicación actualizada mediante la CLI para desarrolladores de Azure:

azd deploy

Esto implementa el código actualizado con la nueva función de orquestación y agentes adicionales en la aplicación de Azure Functions creada en el primer tutorial.

Prueba de la orquestación implementada

Después de la implementación, pruebe la orquestación que se ejecuta en Azure.

  1. Obtenga la clave del sistema para la extensión duradera:

    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. Inicie la orquestación mediante la API integrada:

    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. Utiliza statusQueryGetUri de la respuesta para verificar la finalización y visualizar los resultados junto con las traducciones.

Pasos siguientes

Puntos de conexión compatibles con OpenAI

Recursos adicionales:

  • Last updated on