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.

Información general

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
• Reanude la entrega transmitida cuando se configura con un agente de transmisiones, como Redis; confirmar la finalización duradera por separado
• 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 en tiempo real: entrega la salida del token incremental con reconexión basada en agente, al tiempo que se comprueba el resultado duradero por separado.

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

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

Use los comandos de instalación para el modelo de hospedaje siguiente. Mantenga las opciones de versión preliminar que se muestran para los paquetes en versión preliminar. Para obtener aplicaciones completas, consulte el origen y los ejemplos.

La documentación de la rama de origen puede describir las funcionalidades que aún no están disponibles en los paquetes publicados. Aplique los pasos de configuración y migración solo para las características incluidas en la versión instalada.

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

dotnet add package Microsoft.Agents.AI.Foundry --prerelease
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

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

pip install agent-framework-durabletask agent-framework-openai azure-identity --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();

import os
from agent_framework_azurefunctions 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_DEPLOYMENT", "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.

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

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

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.

Samples

Estos ejemplos muestran compatibilidad, no contratos de API idénticos entre idiomas. Functions genera puntos de conexión HTTP; Las aplicaciones autohospedada implementan y protegen sus propias API.

Subprocesos de agente con estado con historial de conversaciones

Los agentes mantienen sesiones persistentes que sobreviven en varias interacciones. Cada sesión tiene un identificador de sesión único y almacena el estado de conversación en un 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ó.

Configure deliberadamente la limpieza de sesión, la retención del historial y la retención de resultados. Estas políticas tienen efectos diferentes; consulte el ciclo de vida y la capacidad del estado.

En el ejemplo de Azure Functions siguiente se muestran varias solicitudes HTTP a la misma sesión. Ambas integraciones de lenguaje devuelven x-ms-session-id y aceptan el parámetro de consulta canónico sessionId . Tratar el identificador devuelto como opaco: en el ejemplo se muestra una dirección de entidad .NET schema-v2, mientras que Python formato de sesión depende de la estrategia de implementación. En las aplicaciones autohospedadas, use las API de cliente de Durable Task desde su propio proceso o servicio.

# First interaction - start a new session
curl -i -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 session ID in x-ms-session-id header and joke as plain text
# HTTP/1.1 200 OK
# Content-Type: text/plain
# x-ms-session-id: @dafx-v2-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 - reuse the exact session ID returned by the first request
curl -X POST "https://your-function-app.azurewebsites.net/api/agents/Joker/run?sessionId=@dafx-v2-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.

Use sessionId y waitForResponse en nuevos clientes HTTP del agente. thread_id es un alias de solicitud en desuso, no el encabezado de respuesta que se va a capturar. Para Python respuestas JSON, lea sessionId y correlationId. Consulte la guía de migración de LA API HTTP para conocer los alias y el control de conflictos.

Confirmación de la finalización duradera

Para los puntos de conexión del agente de Functions generados .NET, interprete la finalización como se indica a continuación:

Lea x-ms-durable-outcome cuando una finalización no tenga ningún resultado disponible. No trate 410 como pendiente, infiere un resultado del historial de conversaciones o vuelva a enviar automáticamente con una nueva identidad. Los autores de llamadas JSON usan cadena status y numérico statusCode; el campo anterior status_code es un alias de compatibilidad temporal.

Python Functions también distingue el éxito disponible, los resultados con errores retenidos y los resultados completados no disponibles. Sus sobres de error y las opciones del SDK no son idénticas a .NET. Consulte el contrato de finalización .NET y Python contrato HTTP.

Estos estados describen los puntos de conexión del agente. La API de administración de orquestaciones de Durable Functions tiene su propia respuesta de estado y contrato de sondeo.

Streaming confiable

Durable Extension admite el streaming confiable para las aplicaciones que necesitan entrega de tokens en tiempo real junto con la ejecución del agente duradero. 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, para que los clientes puedan reanudar la lectura de flujos de token retenidos después de volver a conectarse o cambios de trabajo, sujeto a la configuración de retención y persistencia del agente.

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. La finalización de la secuencia no confirma por sí mismo que el resultado y el estado del agente se conservaron de forma duradera; compruebe el resultado duradero autoritativo por separado. 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.

El punto de control duradero no crea herramientas ni escrituras externas transaccionales o exactamente una vez. Haga que los efectos secundarios de la actividad y la herramienta sean idempotentes y definan límites de reintento. Error después de una escritura externa, pero antes de que una confirmación duradera pueda hacer que la operación se vuelva a ejecutar.

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.

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

from durabletask.task import OrchestrationContext
from agent_framework_azurefunctions 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: OrchestrationContext, email: dict):
    # 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,
        options={"response_format": SpamDetectionResult},
    )
    spam_result = SpamDetectionResult.model_validate(spam_result_raw.value)

    if spam_result.is_spam:
        result = yield context.call_activity("handle_spam_email", input=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,
        options={"response_format": EmailResponse},
    )
    email_response = EmailResponse.model_validate(email_response_raw.value)

    result = yield context.call_activity("send_email", input=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;
}

from durabletask.task import OrchestrationContext, when_all
from agent_framework_azurefunctions import AgentFunctionApp

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

@app.orchestration_trigger(context_name="context")
def research_orchestration(context: OrchestrationContext, topic: str):
    # 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 when_all([technical_task, market_task, competitor_task])

    # Aggregate results
    all_research = "\n\n".join(r.text 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.text

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

from durabletask.task import OrchestrationContext, when_any
from datetime import timedelta
from agent_framework_azurefunctions import AgentFunctionApp

app = AgentFunctionApp(agents=[content_agent])

@app.orchestration_trigger(context_name="context")
def content_approval_workflow(context: OrchestrationContext, topic: str):
    # 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", input=draft_content.text)

    # 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 when_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", input=draft_content.text)
            return result
        return "Content rejected"

    # Timeout occurred - escalate for review
    result = yield context.call_activity("escalate_for_review", input=draft_content.text)
    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.

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

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.

Implementación y migración de esquemas

El estado del agente schema-v2 cambia tanto el estado persistente como las identidades de envío. Planee la coexistencia de trabajo por separado de la conversión de estado.

Ciclo de vida y capacidad de estado

Elija directivas para cada capa de forma independiente:

En .NET estado de entrada schema-v2, el TTL heredado de 14 días sin configurar no elimina automáticamente la entidad. Configure un TTL global o por agente explícitamente si desea limpiar la sesión inactiva. Pasar timeToLive: null a AddAIAgent hereda el valor predeterminado global configurado; no invalida un valor predeterminado global de siete días con "nunca expira". Establézcalo DefaultTimeToLive = null para deshabilitar el TTL global para los agentes sin invalidaciones positivas por agente.

.NET valor predeterminado es DurableAgentHistoryRetentionMode.KeepAll; Auto habilita la retención de transcripciones basada en presión. ResultRetentionPeriod el valor predeterminado es de 60 segundos y es independiente del TTL de sesión. Python tiene distintas opciones de retención y presupuesto, incluida retention="keep_all" de forma predeterminada. No asuma las mismas opciones o valores predeterminados entre hosts.

Python admite la descarga opcional de carga de blobs administrados por SDK en ambos modelos de hospedaje. Configure los almacenes de productor y consumidor, las credenciales y el ciclo de vida de blobs. La descarga no quita los límites de capacidad de entidad completa ni reemplaza la retención; mantenga los blobs mientras se hace referencia al estado o al historial de reproducción los necesita. Consulte .NET estado y retención, retención y presupuestos de Python y descarga de blobs de Functions Python.

Para obtener instrucciones operativas, capacidad e infraestructura del programador, consulte Integración de Durable Task para Agent Framework.

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:

Use un nuevo centro de tareas para este tutorial, independiente de las aplicaciones duraderas existentes. Para una aplicación existente, siga la implementación y la migración de esquemas en su lugar.

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.

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.

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

import os
from agent_framework_azurefunctions 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_DEPLOYMENT", "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])

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:

   • Bash
   • PowerShell

   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 ...
    "AzureWebJobsStorage": "UseDevelopmentStorage=true",
    "AZURE_OPENAI_ENDPOINT": "https://your-openai-resource.openai.azure.com",
    "AZURE_OPENAI_DEPLOYMENT": "gpt-4o-mini",
    "DURABLE_TASK_SCHEDULER_CONNECTION_STRING": "Endpoint=http://localhost:8080;TaskHub=default;Authentication=None",
    "TASKHUB_NAME": "default"
  }
}

Para Python, agregue "DURABLE_AGENTS_DEPLOYMENT_MODE": "isolated_v2" a Values solo después de confirmar que el centro de tutoriales está aislado. Configure la configuración de aplicación equivalente al implementar; La configuración local no se carga automáticamente.

Para el hospedaje respaldado por el programador, use el mismo centro de tareas en host.json (a través %TASKHUB_NAME%de ), TASKHUB_NAME, DURABLE_TASK_SCHEDULER_CONNECTION_STRING, el emulador y el panel. En este tutorial se usa default, que el emulador crea automáticamente. Si ese centro ya contiene historiales antiguos o los usan otros trabajadores, elija un nombre de concentrador único y sin usar, establezca el valor del DTS_TASK_HUB_NAMES emulador en ese nombre y actualice la configuración de la aplicación y la selección del panel de forma coherente. Simplemente definir un programador cadena de conexión no selecciona el proveedor de almacenamiento del programador.

En host.json, configure el proveedor del programador en extensions.durableTask, conservando la configuración del otro host de la plantilla:

{
  "extensions": {
    "durableTask": {
      "hubName": "%TASKHUB_NAME%",
      "storageProvider": {
        "type": "AzureManaged",
        "connectionStringName": "DURABLE_TASK_SCHEDULER_CONNECTION_STRING"
      }
    }
  }
}

Use las dependencias del proveedor del programador para el idioma de hospedaje. Esta configuración selecciona el back-end para una nueva implementación de tutorial aislada; no habilita el filtrado necesario para una actualización en paralelo.

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.

En estos pasos se da por hecho que el host de Functions usa el proveedor de almacenamiento durable Task Scheduler. Los ejemplos de Python Functions de la versión se Azure Storage de forma predeterminada. Si empieza por esos ejemplos, seleccione y configure el proveedor del programador antes de usar el panel de DTS; de lo contrario, Azurite contiene el estado duradero y el panel del programador permanece vacío. Consulte Configuración del programador de tareas durables.

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.

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 crea el default centro de tareas e inicia el emulador, que coincide con la configuración de este tutorial. El comando 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.

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-v2-MyDurableAgent: entityTrigger
   

Esto ilustra .NET registro schema-v2; Python nombres de desencadenador dependen de la estrategia de implementación. Estos puntos de conexión administran automáticamente el estado de conversación, por lo que no es necesario crear objetos de sesión usted mismo.

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 una nueva sesión y envíe el 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?"

$response = Invoke-WebRequest -Uri "http://localhost:7071/api/agents/MyDurableAgent/run" `
  -Method POST `
  -Headers @{"Content-Type"="text/plain"} `
  -Body "What are three popular programming languages?"
$response.Headers
$response.Content

Respuesta de ejemplo (el x-ms-session-id encabezado contiene el identificador de sesión; en este ejemplo se muestra un identificador de esquema-v2 de .NET):

HTTP/1.1 200 OK
Content-Type: text/plain
x-ms-session-id: @dafx-v2-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 sesión exacto del x-ms-session-id encabezado para la siguiente solicitud. No construya ni vuelva a escribirlo. Python también devuelve este encabezado, pero su formato de identificador puede diferir.

Continuar la conversación

Envíe un mensaje de seguimiento a la misma sesión mediante la inclusión del identificador devuelto en el parámetro de consulta canónico sessionId :

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

$sessionId = $response.Headers["x-ms-session-id"]
$encodedSessionId = [Uri]::EscapeDataString($sessionId)
Invoke-RestMethod -Uri "http://localhost:7071/api/agents/MyDurableAgent/run?sessionId=$encodedSessionId" `
  -Method POST `
  -Headers @{"Content-Type"="text/plain"} `
  -Body "Which one is best for beginners?"

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 valor predeterminado, o el mismo nombre de centro único que configuró en la aplicación y el emulador, 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.

   

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

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

Para la autorización de nivel de función predeterminada de los puntos de conexión generados, incluya una clave de Functions en las solicitudes a Azure. Una clave de Functions no es autorización de usuario de aplicación; consulte seguridad y solución de problemas de 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`

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

Inicio de una nueva conversación en Azure

Cree una nueva sesión y envíe el primer mensaje al agente implementado:

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

$functionName = azd env get-value AZURE_FUNCTION_NAME
$response = Invoke-WebRequest -Uri "https://$functionName.azurewebsites.net/api/agents/MyDurableAgent/run?code=$API_KEY" `
  -Method POST `
  -Headers @{"Content-Type"="text/plain"} `
  -Body "What are three popular programming languages?"
$response.Headers
$response.Content

Guarde el identificador de sesión devuelto en el encabezado de x-ms-session-id respuesta.

Continuar la conversación en Azure

Envíe un mensaje de seguimiento en la misma sesión. Use el identificador exacto devuelto por la respuesta anterior:

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

$sessionId = $response.Headers["x-ms-session-id"]
$encodedSessionId = [Uri]::EscapeDataString($sessionId)
$functionName = azd env get-value AZURE_FUNCTION_NAME
Invoke-RestMethod -Uri "https://$functionName.azurewebsites.net/api/agents/MyDurableAgent/run?code=$API_KEY&sessionId=$encodedSessionId" `
  -Method POST `
  -Headers @{"Content-Type"="text/plain"} `
  -Body "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 centro de tareas configurado para la aplicación implementada.

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.

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

import os
from azure.identity import DefaultAzureCredential
from agent_framework_azurefunctions 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_DEPLOYMENT", "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.

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

from durabletask.task import OrchestrationContext, when_all

@app.orchestration_trigger(context_name="context")
def agent_orchestration_workflow(context: OrchestrationContext, input_text: str):
    """
    Orchestration function that coordinates multiple agents.
    Returns a dictionary with the original response and translations.
    """
    # 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 when_all(parallel_tasks)

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

   • Bash
   • PowerShell

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

   • Bash
   • PowerShell

   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 configurado para la aplicación.

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:

   • Bash
   • PowerShell

   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:

   • Bash
   • PowerShell

   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.

Seguridad y solución de problemas de producción

Trate estos límites de acceso por separado:

• Usuarios de la aplicación: Autentique a los autores de llamadas y autorice el acceso a cada sesión, flujo de trabajo y respuesta de aprobación humana. Los identificadores de sesión, correlación y solicitud son identificadores, no credenciales.
• Puntos de conexión de Functions: Las claves de nivel de función protegen los puntos de conexión HTTP generados, pero no establecen la identidad del usuario de la aplicación ni la propiedad de la sesión. Proteger y girar claves; no los exponga en el código de aplicación entregado por el explorador.
• Acceso al modelo: Conceda a la identidad de trabajo los permisos de modelo necesarios. Las credenciales del modelo no conceden acceso al programador ni a la sesión de otro usuario.
• Acceso del programador: Configure los permisos de trabajo, cliente y panel para sus roles previstos. Para obtener instrucciones de infraestructura, consulte Integración de Durable Task.

Pasos siguientes

Recursos adicionales:

• Extensión de tareaDurable para Microsoft Agent Framework
• Información general sobre el programador de tareas duraderas
• Tablero del planificador de tareas duraderas
• Plan de consumo flexible de Azure Functions
• Conceptos y patrones de Durable Functions