Agentes en flujos de trabajo

Lo que vas a construir

Creará un flujo de trabajo que:

• Usa azure Foundry Agent Service para crear agentes inteligentes
• Implementa un agente de traducción en francés que traduce la entrada al francés.
• Implementa un agente de traducción de español que traduce francés a español
• Implementa un agente de traducción en inglés que traduce el español a inglés
• Conecta agentes en una canalización de flujo de trabajo secuencial
• Transmite actualizaciones en tiempo real a medida que los agentes procesan solicitudes
• Muestra la limpieza de recursos adecuada para los agentes de Azure Foundry.

Conceptos tratados

• Agentes en flujos de trabajo
• Bordes directos
• Generador de flujos de trabajo

Prerrequisitos

• SDK de .NET 8.0 o posterior
• Puntos de conexión de servicio y despliegues de Azure Foundry configurados
• CLI de Azure instalada y autenticada (para la autenticación de credenciales de Azure)
• Una nueva aplicación de consola

Paso 1: Instalar paquetes NuGet

En primer lugar, instale los paquetes necesarios para el proyecto de .NET:

dotnet add package Azure.AI.Agents.Persistent --prerelease
dotnet add package Azure.Identity
dotnet add package Microsoft.Agents.AI.AzureAI --prerelease
dotnet add package Microsoft.Agents.AI.Workflows --prerelease

Paso 2: Configuración del cliente de Azure Foundry

Configure el cliente de Azure Foundry con variables de entorno y autenticación:

using System;
using System.Threading.Tasks;
using Azure.AI.Agents.Persistent;
using Azure.Identity;
using Microsoft.Agents.AI;
using Microsoft.Agents.AI.Workflows;
using Microsoft.Extensions.AI;

public static class Program
{
    private static async Task Main()
    {
        // Set up the Azure Foundry client
        var endpoint = Environment.GetEnvironmentVariable("AZURE_FOUNDRY_PROJECT_ENDPOINT") ?? throw new Exception("AZURE_FOUNDRY_PROJECT_ENDPOINT is not set.");
        var model = Environment.GetEnvironmentVariable("AZURE_FOUNDRY_PROJECT_MODEL_ID") ?? "gpt-4o-mini";
        var persistentAgentsClient = new PersistentAgentsClient(endpoint, new AzureCliCredential());

Paso 3: Crear método de fábrica del agente

Implemente un método auxiliar para crear agentes de Azure Foundry con instrucciones específicas:

    /// <summary>
    /// Creates a translation agent for the specified target language.
    /// </summary>
    /// <param name="targetLanguage">The target language for translation</param>
    /// <param name="persistentAgentsClient">The PersistentAgentsClient to create the agent</param>
    /// <param name="model">The model to use for the agent</param>
    /// <returns>A ChatClientAgent configured for the specified language</returns>
    private static async Task<ChatClientAgent> GetTranslationAgentAsync(
        string targetLanguage,
        PersistentAgentsClient persistentAgentsClient,
        string model)
    {
        var agentMetadata = await persistentAgentsClient.Administration.CreateAgentAsync(
            model: model,
            name: $"{targetLanguage} Translator",
            instructions: $"You are a translation assistant that translates the provided text to {targetLanguage}.");

        return await persistentAgentsClient.GetAIAgentAsync(agentMetadata.Value.Id);
    }
}

Paso 4: Creación de agentes especializados de Azure Foundry

Cree tres agentes de traducción mediante el método auxiliar:

        // Create agents
        AIAgent frenchAgent = await GetTranslationAgentAsync("French", persistentAgentsClient, model);
        AIAgent spanishAgent = await GetTranslationAgentAsync("Spanish", persistentAgentsClient, model);
        AIAgent englishAgent = await GetTranslationAgentAsync("English", persistentAgentsClient, model);

Paso 5: Compilar el flujo de trabajo

Conecte los agentes en un flujo de trabajo secuencial mediante WorkflowBuilder:

        // Build the workflow by adding executors and connecting them
        var workflow = new WorkflowBuilder(frenchAgent)
            .AddEdge(frenchAgent, spanishAgent)
            .AddEdge(spanishAgent, englishAgent)
            .Build();

Paso 6: Ejecutar con streaming

Ejecute el flujo de trabajo con streaming para observar las actualizaciones en tiempo real de todos los agentes:

        // Execute the workflow
        await using StreamingRun run = await InProcessExecution.StreamAsync(workflow, new ChatMessage(ChatRole.User, "Hello World!"));

        // Must send the turn token to trigger the agents.
        // The agents are wrapped as executors. When they receive messages,
        // they will cache the messages and only start processing when they receive a TurnToken.
        await run.TrySendMessageAsync(new TurnToken(emitEvents: true));
        await foreach (WorkflowEvent evt in run.WatchStreamAsync().ConfigureAwait(false))
        {
            if (evt is AgentRunUpdateEvent executorComplete)
            {
                Console.WriteLine($"{executorComplete.ExecutorId}: {executorComplete.Data}");
            }
        }

Paso 7: Limpieza de recursos

Limpie correctamente los agentes de Azure Foundry después de su uso:

        // Cleanup the agents created for the sample.
        await persistentAgentsClient.Administration.DeleteAgentAsync(frenchAgent.Id);
        await persistentAgentsClient.Administration.DeleteAgentAsync(spanishAgent.Id);
        await persistentAgentsClient.Administration.DeleteAgentAsync(englishAgent.Id);
    }

Funcionamiento

1. Configuración del cliente de Azure Foundry: se usa PersistentAgentsClient con credenciales de la CLI de Azure para la autenticación
2. Creación del agente: crea agentes persistentes en Azure Foundry con instrucciones específicas para la traducción.
3. Procesamiento secuencial: el agente francés traduce primero la entrada, luego el agente español y el agente inglés.
4. Patrón de token de turno: los agentes almacenan en caché los mensajes y solo procesan cuando reciben un TurnToken
5. Actualizaciones de streaming: AgentRunUpdateEvent proporciona actualizaciones de tokens en tiempo real a medida que los agentes generan respuestas.
6. Administración de recursos: limpieza adecuada de agentes de Azure Foundry mediante la API de administración

Conceptos clave

• Servicio agente de Azure Foundry: agentes de inteligencia artificial basados en la nube con funcionalidades de razonamiento avanzadas
• PersistentAgentsClient: cliente para crear y administrar agentes en Azure Foundry
• AgentRunUpdateEvent: actualizaciones de streaming en tiempo real durante la ejecución del agente
• TurnToken: señal que desencadena el procesamiento del agente después del almacenamiento en caché de mensajes
• Flujo de trabajo secuencial: agentes conectados en una canalización donde la salida fluye de una a la siguiente

Implementación completa

Para obtener la implementación completa de este flujo de trabajo de agentes de Azure Foundry, consulte el ejemplo foundryAgent Program.cs en el repositorio de Agent Framework.

Lo que vas a construir

Creará un flujo de trabajo que:

• Usa el servicio agente de Azure AI para crear agentes inteligentes
• Implementa un agente de escritor que crea contenido basado en mensajes.
• Implementa un agente revisor que proporciona comentarios sobre el contenido.
• Conecta agentes en una canalización de flujo de trabajo secuencial
• Transmite actualizaciones en tiempo real a medida que los agentes procesan solicitudes
• Muestra una administración de contexto asincrónica adecuada para los clientes de Azure AI.

Conceptos tratados

• Agentes en flujos de trabajo
• Bordes directos
• Generador de flujos de trabajo

Prerrequisitos

• Python 3.10 o posterior
• Agent Framework instalado: pip install agent-framework-azure-ai --pre
• Servicio agente de Azure AI configurado con variables de entorno adecuadas
• Autenticación de la CLI de Azure: az login

Paso 1: Importar dependencias necesarias

Empiece por importar los componentes necesarios para los agentes y flujos de trabajo de Azure AI:

import asyncio
from collections.abc import Awaitable, Callable
from contextlib import AsyncExitStack
from typing import Any

from agent_framework import AgentRunUpdateEvent, WorkflowBuilder, WorkflowOutputEvent
from agent_framework.azure import AzureAIAgentClient
from azure.identity.aio import AzureCliCredential

Paso 2: Creación de Azure AI Agent Factory

Cree una función auxiliar para administrar la creación de agentes de Azure AI con el control adecuado del contexto asincrónico:

async def create_azure_ai_agent() -> tuple[Callable[..., Awaitable[Any]], Callable[[], Awaitable[None]]]:
    """Helper method to create an Azure AI agent factory and a close function.

    This makes sure the async context managers are properly handled.
    """
    stack = AsyncExitStack()
    cred = await stack.enter_async_context(AzureCliCredential())

    client = await stack.enter_async_context(AzureAIAgentClient(async_credential=cred))

    async def agent(**kwargs: Any) -> Any:
        return await stack.enter_async_context(client.create_agent(**kwargs))

    async def close() -> None:
        await stack.aclose()

    return agent, close

Paso 3: Creación de agentes especializados de Azure AI

Cree dos agentes especializados para la creación y revisión de contenido:

async def main() -> None:
    agent, close = await create_azure_ai_agent()
    try:
        # Create a Writer agent that generates content
        writer = await agent(
            name="Writer",
            instructions=(
                "You are an excellent content writer. You create new content and edit contents based on the feedback."
            ),
        )

        # Create a Reviewer agent that provides feedback
        reviewer = await agent(
            name="Reviewer",
            instructions=(
                "You are an excellent content reviewer. "
                "Provide actionable feedback to the writer about the provided content. "
                "Provide the feedback in the most concise manner possible."
            ),
        )

Paso 4: Compilar el flujo de trabajo

Conecte los agentes en un flujo de trabajo secuencial usando el constructor fluido.

        # Build the workflow with agents as executors
        workflow = WorkflowBuilder().set_start_executor(writer).add_edge(writer, reviewer).build()

Paso 5: Ejecutar con streaming

Ejecute el flujo de trabajo con streaming para observar las actualizaciones en tiempo real de ambos agentes:

        last_executor_id: str | None = None

        events = workflow.run_stream("Create a slogan for a new electric SUV that is affordable and fun to drive.")
        async for event in events:
            if isinstance(event, AgentRunUpdateEvent):
                # Handle streaming updates from agents
                eid = event.executor_id
                if eid != last_executor_id:
                    if last_executor_id is not None:
                        print()
                    print(f"{eid}:", end=" ", flush=True)
                    last_executor_id = eid
                print(event.data, end="", flush=True)
            elif isinstance(event, WorkflowOutputEvent):
                print("\n===== Final output =====")
                print(event.data)
    finally:
        await close()

Paso 6: Completar la función principal

Encapsula todo en la función principal con una ejecución asincrónica adecuada:

if __name__ == "__main__":
    asyncio.run(main())

Funcionamiento

1. Configuración del cliente de Azure AI: se usa AzureAIAgentClient con las credenciales de la CLI de Azure para la autenticación
2. Patrón de factoría del agente: crea una función de fábrica que administra el ciclo de vida de contexto asincrónico para varios agentes.
3. Procesamiento secuencial: el agente de escritura genera primero el contenido y, a continuación, lo pasa al agente revisor.
4. Actualizaciones de streaming: AgentRunUpdateEvent proporciona actualizaciones de tokens en tiempo real a medida que los agentes generan respuestas.
5. Administración de contextos: limpieza adecuada de los recursos de Azure AI mediante AsyncExitStack

Conceptos clave

• Servicio agente de Azure AI: agentes de inteligencia artificial basados en la nube con funcionalidades avanzadas de razonamiento
• AgentRunUpdateEvent: actualizaciones de streaming en tiempo real durante la ejecución del agente
• AsyncExitStack: administración de contexto asincrónica adecuada para varios recursos
• Patrón de factoría del agente: creación reutilizable de agentes con configuración de cliente compartida
• Flujo de trabajo secuencial: agentes conectados en una canalización donde la salida fluye de una a la siguiente

Implementación completa

Para obtener la implementación de trabajo completa de este flujo de trabajo de agentes de Azure AI, consulte el ejemplo azure_ai_agents_streaming.py en el repositorio de Agent Framework.