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 construirás

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

Prerequisites

SDK de .NET 8.0 o posterior
Un punto de conexión y un modelo de proyecto 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 Microsoft.Agents.AI.Foundry --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 Azure.AI.Projects;
using Azure.AI.Projects.Agents;
using Azure.Identity;
using Microsoft.Agents.AI;
using Microsoft.Agents.AI.Foundry;
using Microsoft.Agents.AI.Workflows;
using Microsoft.Extensions.AI;

public static class Program
{
    private static async Task Main()
    {
        // Set up the Azure AI Project client
        var endpoint = Environment.GetEnvironmentVariable("FOUNDRY_PROJECT_ENDPOINT")
            ?? throw new InvalidOperationException("FOUNDRY_PROJECT_ENDPOINT is not set.");
        var deploymentName = Environment.GetEnvironmentVariable("FOUNDRY_MODEL") ?? "gpt-4o-mini";
        var aiProjectClient = new AIProjectClient(new Uri(endpoint), new AzureCliCredential());

Paso 3: Crear el método de fábrica de agentes

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="aiProjectClient">The AIProjectClient to create the agent</param>
    /// <param name="model">The model to use for the agent</param>
    /// <returns>A FoundryAgent configured for the specified language</returns>
    private static async Task<FoundryAgent> GetTranslationAgentAsync(
        string targetLanguage,
        AIProjectClient aiProjectClient,
        string model)
    {
        string agentName = $"{targetLanguage}Translator";
        var version = await aiProjectClient.AgentAdministrationClient.CreateAgentVersionAsync(
            agentName,
            new ProjectsAgentVersionCreationOptions(
                new DeclarativeAgentDefinition(model)
                {
                    Instructions = $"You are a translation assistant that translates the provided text to {targetLanguage}."
                }));

        return aiProjectClient.AsAIAgent(version);
    }
}

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", aiProjectClient, deploymentName);
        AIAgent spanishAgent = await GetTranslationAgentAsync("Spanish", aiProjectClient, deploymentName);
        AIAgent englishAgent = await GetTranslationAgentAsync("English", aiProjectClient, deploymentName);

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.RunStreamingAsync(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())
        {
            if (evt is AgentResponseUpdateEvent 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 aiProjectClient.AgentAdministrationClient.DeleteAgentAsync(frenchAgent.Name);
        await aiProjectClient.AgentAdministrationClient.DeleteAgentAsync(spanishAgent.Name);
        await aiProjectClient.AgentAdministrationClient.DeleteAgentAsync(englishAgent.Name);
    }

Funcionamiento

Configuración del cliente de Azure Foundry: se usa AIProjectClient con credenciales de la CLI de Azure para la autenticación
Creación del agente: crea agentes con versiones 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: AgentResponseUpdateEvent 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
AIProjectClient: cliente para crear y administrar agentes en Azure Foundry
WorkflowEvent: los eventos de salida (type="output") contienen datos de salida del agente (AgentResponseUpdate para streaming, AgentResponse para no streaming)
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 construirás

Creará un flujo de trabajo que:

Usa FoundryChatClient 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

Conceptos tratados

Prerequisites

Python 3.11 o una versión posterior
Agent Framework instalado: pip install agent-framework
Respuestas de Azure OpenAI configuradas 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 flujos de trabajo y los agentes de respuestas de Azure OpenAI:

import asyncio
import os

from agent_framework import AgentResponseUpdate, WorkflowBuilder
from agent_framework.foundry import FoundryChatClient
from azure.identity import AzureCliCredential

Paso 2: Creación de un cliente de respuestas de Azure OpenAI

Cree un cliente compartido que pueda usar para construir varios agentes:

async def main() -> None:
    client = FoundryChatClient(
        project_endpoint=os.environ["FOUNDRY_PROJECT_ENDPOINT"],
        model=os.environ["FOUNDRY_MODEL"],
        credential=AzureCliCredential(),
    )

Paso 3: Crear agentes especializados

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

    # Create a Writer agent that generates content
    writer_agent = client.as_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_agent = client.as_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 mediante el generador:

        # Build the workflow with agents as executors
        workflow = WorkflowBuilder(start_executor=writer_agent).add_edge(writer_agent, reviewer_agent).build()

Paso 5: Ejecutar con streaming

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

    last_author: str | None = None

    events = workflow.run("Create a slogan for a new electric SUV that is affordable and fun to drive.", stream=True)
    async for event in events:
        if event.type == "output" and isinstance(event.data, AgentResponseUpdate):
            update = event.data
            author = update.author_name
            if author != last_author:
                if last_author is not None:
                    print()
                print(f"{author}: {update.text}", end="", flush=True)
                last_author = author
            else:
                print(update.text, end="", flush=True)

Paso 6: Completar la función principal

Envuelve todo en la función principal con la ejecución asíncrona adecuada:

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

Funcionamiento

Configuración de cliente: usa una FoundryChatClient con las credenciales de la CLI de Azure para la autenticación.
Creación del agente: crea agentes de escritor y revisor a partir de la misma configuración de cliente.
Procesamiento secuencial: el agente de escritor genera primero el contenido y, a continuación, lo pasa al agente revisor.
Actualizaciones en streaming: Los eventos de salida (type="output") con datos AgentResponseUpdate proporcionan actualizaciones de tokens en tiempo real a medida que los agentes generan respuestas.

Conceptos clave

FoundryChatClient: cliente compartido usado para crear agentes de flujo de trabajo con una configuración coherente.
WorkflowEvent: los eventos de salida (type="output") contienen datos de salida del agente (AgentResponseUpdate para streaming, AgentResponse para no streaming).
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, consulte azure_ai_agents_streaming.py en el repositorio de Agent Framework.

Lo que construirás

Creará un flujo de trabajo que:

Usa agentes de despliegue de modelos de Microsoft Foundry como ejecutores de flujo de trabajo
Implementa un agente de traducción al francés
Implementa un agente de traducción al español
Implementa un agente de traducción en 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

Conceptos tratados

Prerequisites

Go 1.26 o posterior
Configuración del punto de conexión del proyecto Microsoft Foundry y del despliegue del modelo
autenticación de CLI de Azure u otro origen de credenciales de Azure

En un nuevo directorio, inicialice un módulo go e instale Agent Framework:

go mod init example.com/agent-workflow
go get github.com/microsoft/agent-framework-go

Establezca el punto de conexión del proyecto y el nombre del modelo implementado:

$env:FOUNDRY_PROJECT_ENDPOINT = "https://<resource>.services.ai.azure.com/api/projects/<project>"
$env:FOUNDRY_MODEL = "<deployment-name>"

Paso 1: Configurar la configuración de Foundry

En las secciones siguientes se desarrolla la función runExample en la implementación completa. Comience con las importaciones y la configuración:

import (
    "context"
    "errors"
    "fmt"
    "log"
    "os"
    "strings"

    "github.com/Azure/azure-sdk-for-go/sdk/azidentity"
    "github.com/microsoft/agent-framework-go/agent"
    "github.com/microsoft/agent-framework-go/message"
    "github.com/microsoft/agent-framework-go/provider/foundryprovider"
    "github.com/microsoft/agent-framework-go/workflow"
    "github.com/microsoft/agent-framework-go/workflow/agentworkflow"
    "github.com/microsoft/agent-framework-go/workflow/inproc"
)

endpoint := strings.TrimSpace(os.Getenv("FOUNDRY_PROJECT_ENDPOINT"))
if endpoint == "" {
    return fmt.Errorf("set FOUNDRY_PROJECT_ENDPOINT")
}
model := strings.TrimSpace(os.Getenv("FOUNDRY_MODEL"))
if model == "" {
    return fmt.Errorf("set FOUNDRY_MODEL to your model deployment name")
}

token, err := azidentity.NewDefaultAzureCredential(nil)
if err != nil {
    return err
}

Warning

azidentity.NewDefaultAzureCredential es conveniente para el desarrollo, pero requiere una consideración cuidadosa en producción. En producción, considere usar una credencial específica, como azidentity.NewManagedIdentityCredential, para evitar problemas de latencia, intentos no deseados de probar credenciales y posibles riesgos de seguridad derivados de los mecanismos de respaldo.

Paso 2: Crear una función de factoría de agentes

Cree agentes con instrucciones de traducción específicas:

newTranslationAgent := func(language string) *agent.Agent {
    return foundryprovider.NewAgent(endpoint, token, foundryprovider.ModelDeployment(model), foundryprovider.AgentConfig{
        Instructions: fmt.Sprintf(
            "Translate the user's text to %s. Return only the translation.",
            language,
        ),
        Config: agent.Config{Name: language + "Agent"},
    })
}

Paso 3: Crear agentes especializados de Foundry

frenchAgent := newTranslationAgent("French")
spanishAgent := newTranslationAgent("Spanish")
englishAgent := newTranslationAgent("English")

Paso 4: Compilar el flujo de trabajo

Los agentes se pueden usar como ejecutores de flujo de trabajo, lo que permite los pasos de flujo de trabajo con tecnología de IA.

Enlace cada agente como ejecutor de flujo de trabajo y, a continuación, conecte los ejecutores con bordes:

cfg := agentworkflow.Config{ForwardIncomingMessages: new(false)}
french := agentworkflow.New(frenchAgent, cfg)
spanish := agentworkflow.New(spanishAgent, cfg)
english := agentworkflow.New(englishAgent, cfg)

wf, err := workflow.NewBuilder(french).
    AddEdge(french, spanish).
    AddEdge(spanish, english).
    WithIntermediateOutputFrom(french, spanish).
    WithOutputFrom(english).
    Build()
if err != nil {
    return err
}

ForwardIncomingMessages: new(false) pasa la respuesta de cada agente en sentido descendente sin reenviar la entrada de ese agente. Cada ejecutor hospedado administra su propia sesión del agente.

Registra francés y español como orígenes de salida intermedios para que quienes llaman puedan observar los tres agentes. El inglés sigue siendo el origen de salida del terminal. La designación de salida controla lo que ve el llamador, no qué mensajes fluyen por los perímetros.

Paso 5: Ejecutar con streaming

Ejecute el flujo de trabajo y habilite los eventos de actualización con un workflow.TurnToken. Gestione los eventos de fallo del flujo de trabajo, así como los errores del iterador de eventos:

run, err := inproc.Default.RunStreaming(ctx, wf, message.NewText("Hello World"))
if err != nil {
    return err
}
defer func() {
    err = errors.Join(err, run.Close(context.WithoutCancel(ctx)))
}()

emitEvents := true
if accepted, err := run.TrySendMessage(ctx, workflow.TurnToken{EmitEvents: &emitEvents}); err != nil {
    return err
} else if !accepted {
    return fmt.Errorf("workflow doesn't accept turn tokens")
}

lastExecutorID := ""
for evt, streamErr := range run.WatchStream(ctx) {
    if streamErr != nil {
        return streamErr
    }
    switch event := evt.(type) {
    case workflow.OutputEvent:
        update, ok := event.Output.(*agent.ResponseUpdate)
        if !ok || update.String() == "" {
            continue
        }
        if event.ExecutorID != lastExecutorID {
            if lastExecutorID != "" {
                fmt.Println()
            }
            fmt.Printf("%s: ", event.ExecutorID)
            lastExecutorID = event.ExecutorID
        }
        fmt.Print(update.String())
    case workflow.ErrorEvent:
        return event.Error
    case workflow.ExecutorFailedEvent:
        return fmt.Errorf("executor %q failed: %w",
            event.ExecutorID, event.Error)
    }
}
if lastExecutorID != "" {
    fmt.Println()
}

La salida contiene traducciones de francés, español e inglés, agrupadas por ejecutor. El ejemplo completo sale de runExample antes de registrar un error irrecuperable, por lo que la limpieza diferida del flujo de trabajo todavía se ejecuta.

A diferencia del ejemplo de C#, este ejemplo de Go no crea versiones de agente almacenadas que necesitan eliminación en Foundry.

Funcionamiento

Configuración de cliente: usa un punto de conexión de proyecto de Foundry, una implementación de modelos y una credencial de Azure.
Creación del agente: crea agentes especializados con instrucciones específicas del lenguaje.
Hospedaje del agente: usa agentworkflow.New para enlazar cada agente como ejecutor de flujo de trabajo.
Procesamiento secuencial: el ejecutor francés se ejecuta primero, luego español y luego inglés.
Patrón Turn Token: Los agentes hospedados almacenan los mensajes en búfer y se ejecutan cuando reciben un workflow.TurnToken.
Actualizaciones de streaming: los registros de salida intermedios y terminales muestran las actualizaciones de los tres agentes.
Control de errores: el host comprueba ErrorEvent y ExecutorFailedEvent, y cierra la ejecución del flujo de trabajo antes de notificar errores.

Conceptos clave

Agente Foundry: Un agent.Agent respaldado por una implementación de modelo de Foundry.
agentworkflow. Nuevo: adapta un agente para su uso como ejecutor de flujo de trabajo.
workflow.TurnToken: Señal que activa a los agentes alojados para que procesen los mensajes almacenados en búfer.
Workflow OutputEvent: contiene las actualizaciones de la respuesta del agente y las salidas finales del flujo de trabajo.
Flujo de trabajo secuencial: agentes conectados en una canalización donde la salida fluye de una a la siguiente.

Implementación completa

Guarde este ejemplo como main.go en el módulo Go:

package main

import (
    "context"
    "errors"
    "fmt"
    "log"
    "os"
    "strings"

    "github.com/Azure/azure-sdk-for-go/sdk/azidentity"
    "github.com/microsoft/agent-framework-go/agent"
    "github.com/microsoft/agent-framework-go/message"
    "github.com/microsoft/agent-framework-go/provider/foundryprovider"
    "github.com/microsoft/agent-framework-go/workflow"
    "github.com/microsoft/agent-framework-go/workflow/agentworkflow"
    "github.com/microsoft/agent-framework-go/workflow/inproc"
)

func main() {
    if err := runExample(context.Background()); err != nil {
        log.Fatal(err)
    }
}

func runExample(ctx context.Context) (err error) {
    endpoint := strings.TrimSpace(os.Getenv("FOUNDRY_PROJECT_ENDPOINT"))
    if endpoint == "" {
        return fmt.Errorf("set FOUNDRY_PROJECT_ENDPOINT")
    }
    model := strings.TrimSpace(os.Getenv("FOUNDRY_MODEL"))
    if model == "" {
        return fmt.Errorf("set FOUNDRY_MODEL to your model deployment name")
    }

    token, err := azidentity.NewDefaultAzureCredential(nil)
    if err != nil {
        return err
    }

    newTranslationAgent := func(language string) *agent.Agent {
        return foundryprovider.NewAgent(
            endpoint,
            token,
            foundryprovider.ModelDeployment(model),
            foundryprovider.AgentConfig{
                Instructions: fmt.Sprintf(
                    "Translate the user's text to %s. Return only the translation.",
                    language,
                ),
                Config: agent.Config{Name: language + "Agent"},
            },
        )
    }

    cfg := agentworkflow.Config{ForwardIncomingMessages: new(false)}
    french := agentworkflow.New(newTranslationAgent("French"), cfg)
    spanish := agentworkflow.New(newTranslationAgent("Spanish"), cfg)
    english := agentworkflow.New(newTranslationAgent("English"), cfg)

    wf, err := workflow.NewBuilder(french).
        AddEdge(french, spanish).
        AddEdge(spanish, english).
        WithIntermediateOutputFrom(french, spanish).
        WithOutputFrom(english).
        Build()
    if err != nil {
        return err
    }

    run, err := inproc.Default.RunStreaming(ctx, wf, message.NewText("Hello World"))
    if err != nil {
        return err
    }
    defer func() {
        err = errors.Join(err, run.Close(context.WithoutCancel(ctx)))
    }()

    emitEvents := true
    if accepted, err := run.TrySendMessage(ctx, workflow.TurnToken{EmitEvents: &emitEvents}); err != nil {
        return err
    } else if !accepted {
        return fmt.Errorf("workflow doesn't accept turn tokens")
    }

    lastExecutorID := ""
    for evt, streamErr := range run.WatchStream(ctx) {
        if streamErr != nil {
            return streamErr
        }
        switch event := evt.(type) {
        case workflow.OutputEvent:
            update, ok := event.Output.(*agent.ResponseUpdate)
            if !ok || update.String() == "" {
                continue
            }
            if event.ExecutorID != lastExecutorID {
                if lastExecutorID != "" {
                    fmt.Println()
                }
                fmt.Printf("%s: ", event.ExecutorID)
                lastExecutorID = event.ExecutorID
            }
            fmt.Print(update.String())
        case workflow.ErrorEvent:
            return event.Error
        case workflow.ExecutorFailedEvent:
            return fmt.Errorf("executor %q failed: %w",
                event.ExecutorID, event.Error)
        }
    }
    if lastExecutorID != "" {
        fmt.Println()
    }
    return nil
}

Ejecute el ejemplo:

go run .

Warning

Tip

Consulte el ejemplo de agentes en flujos de trabajo para obtener otra canalización de traducción ejecutable.

Pasos siguientes

Humano en el bucle

Comentarios

¿Le ha resultado útil esta página?

Se actualizó por última vez el 2026-10-29

Agentes en flujos de trabajo

Lo que construirás

Conceptos tratados

Prerequisites

Paso 1: Instalar paquetes NuGet

Paso 2: Configuración del cliente de Azure Foundry

Paso 3: Crear el método de fábrica de agentes

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 construirás

Conceptos tratados

Prerequisites

Paso 1: Importar dependencias necesarias

Paso 2: Creación de un cliente de respuestas de Azure OpenAI

Paso 3: Crear agentes especializados

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

Lo que construirás

Conceptos tratados

Prerequisites

Paso 1: Configurar la configuración de Foundry

Paso 2: Crear una función de factoría de agentes

Paso 3: Crear agentes especializados de Foundry

Paso 4: Compilar el flujo de trabajo

Paso 5: Ejecutar con streaming

Funcionamiento

Conceptos clave

Implementación completa

Pasos siguientes

Comentarios

Recursos adicionales