Agentes en flujos de trabajo

En este tutorial se muestra cómo integrar agentes de inteligencia artificial en flujos de trabajo mediante Agent Framework. Aprenderá a crear flujos de trabajo que aprovechan el poder de los agentes especializados de inteligencia artificial para la creación de contenido, revisión y otras tareas colaborativas.

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

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

Configuración del cliente de Azure Foundry: se usa PersistentAgentsClient con credenciales de la CLI de Azure para la autenticación
Creación del agente: crea agentes persistentes en Azure Foundry con instrucciones específicas para la traducción.
Procesamiento secuencial: el agente francés traduce primero la entrada, luego el agente español y el agente inglés.
Patrón de token de turno: los agentes almacenan en caché los mensajes y solo procesan cuando reciben un TurnToken
Actualizaciones de streaming: AgentRunUpdateEvent proporciona actualizaciones de tokens en tiempo real a medida que los agentes generan respuestas.
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

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

Configuración del cliente de Azure AI: se usa AzureAIAgentClient con las credenciales de la CLI de Azure para la autenticación
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.
Procesamiento secuencial: el agente de escritura genera primero el contenido y, a continuación, lo pasa al agente revisor.
Actualizaciones de streaming: AgentRunUpdateEvent proporciona actualizaciones de tokens en tiempo real a medida que los agentes generan respuestas.
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.

Pasos siguientes

Más información sobre la bifurcación en flujos de trabajo

Comentarios

¿Le ha resultado útil esta página?

Last updated on 2025-11-21

Compartir a través de

Agentes en flujos de trabajo

Lo que vas a construir

Conceptos tratados

Prerrequisitos

Paso 1: Instalar paquetes NuGet

Paso 2: Configuración del cliente de Azure Foundry

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

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

Paso 5: Compilar el flujo de trabajo

Paso 6: Ejecutar con streaming

Paso 7: Limpieza de recursos

Funcionamiento

Conceptos clave

Implementación completa

Lo que vas a construir

Conceptos tratados

Prerrequisitos

Paso 1: Importar dependencias necesarias

Paso 2: Creación de Azure AI Agent Factory

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

Paso 4: Compilar el flujo de trabajo

Paso 5: Ejecutar con streaming

Paso 6: Completar la función principal

Funcionamiento

Conceptos clave

Implementación completa

Pasos siguientes

Comentarios

Recursos adicionales