Agentes em fluxos de trabalho

O que você vai construir

Você criará um fluxo de trabalho que:

• Usa o Azure Foundry Agent Service para criar agentes inteligentes
• Implementa um agente de tradução francês que traduz a entrada para o francês
• Implementa um agente de tradução espanhol que traduz francês para espanhol
• Implementa um agente de tradução em inglês que traduz espanhol de volta para inglês
• Conecta agentes em um pipeline de fluxo de trabalho sequencial
• Transmite atualizações em tempo real à medida que os agentes processam solicitações
• Demonstra a limpeza adequada de recursos para agentes do Azure Foundry

Conceitos abordados

• Agentes em Fluxos de Trabalho
• Arestas Diretas
• Construtor de Fluxos de Trabalho

Pré-requisitos

• SDK do .NET 8.0 ou posterior
• Um endpoint e modelo de projeto Azure Foundry configurados
• CLI do Azure instalada e autenticada (para autenticação de credenciais do Azure)
• Um novo aplicativo de console

Etapa 1: Instalar pacotes NuGet

Primeiro, instale os pacotes necessários para seu projeto .NET:

dotnet add package Azure.AI.Projects --prerelease
dotnet add package Azure.Identity
dotnet add package Microsoft.Agents.AI.Foundry --prerelease
dotnet add package Microsoft.Agents.AI.Workflows --prerelease

Etapa 2: Configurar o Azure Foundry Client

Configure o cliente do Azure Foundry com variáveis de ambiente e autenticação:

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("AZURE_AI_PROJECT_ENDPOINT")
            ?? throw new InvalidOperationException("AZURE_AI_PROJECT_ENDPOINT is not set.");
        var deploymentName = Environment.GetEnvironmentVariable("AZURE_AI_MODEL_DEPLOYMENT_NAME") ?? "gpt-4o-mini";
        var aiProjectClient = new AIProjectClient(new Uri(endpoint), new AzureCliCredential());

Passo 3: Método de Criar Fábrica de Agentes

Implemente um método auxiliar para criar agentes do Azure Foundry com instruções 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 ChatClientAgent configured for the specified language</returns>
    private static async Task<ChatClientAgent> 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);
    }
}

Passo 4: Criar Agentes Especializados Azure Foundry

Crie três agentes de tradução usando o 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);

Etapa 5: Criar o fluxo de trabalho

Conecte os agentes em um fluxo de trabalho sequencial usando o WorkflowBuilder:

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

Etapa 6: Executar com streaming

Execute o fluxo de trabalho com streaming para observar atualizações em tempo real de todos os 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}");
            }
        }

Etapa 7: Limpeza de recursos

Limpe corretamente os agentes do Azure Foundry após o uso:

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

Como funciona

1. Configuração do Cliente do Azure Foundry: Usa AIProjectClient com credenciais da CLI do Azure para autenticação
2. Criação de Agentes: Cria agentes versionados no Azure Foundry com instruções específicas para tradução
3. Processamento sequencial: o agente francês traduz primeiro a entrada, depois o agente espanhol e, em seguida, o agente inglês
4. Padrão de token: os agentes armazenam mensagens em cache e só processam quando recebem um TurnToken
5. Streaming Updates: AgentResponseUpdateEvent fornece atualizações de token em tempo real à medida que os agentes geram respostas
6. Gerenciamento de recursos: limpeza adequada dos agentes do Azure Foundry usando a API de administração

Conceitos-chave

• Azure Foundry Agent Service: agentes de IA baseados em nuvem com recursos avançados de raciocínio
• AIProjectClient: Cliente para criar e gerir agentes no Azure Foundry
• WorkflowEvent: Os eventos de saída (type="output") contêm dados de saída do agente (AgentResponseUpdate para streaming, AgentResponse para não-streaming)
• TurnToken: Sinal que aciona o processamento do agente após o cache de mensagens
• Fluxo de trabalho sequencial: agentes conectados em um pipeline onde a saída flui de um para o próximo

Implementação Completa

Para obter a implementação de trabalho completa deste fluxo de trabalho de agentes do Azure Foundry, consulte o exemplo de Program.cs do FoundryAgent no repositório do Agent Framework.

O que você vai construir

Você criará um fluxo de trabalho que:

• Usa FoundryChatClient para criar agentes inteligentes
• Implementa um agente do Writer que cria conteúdo com base em prompts
• Implementa um agente revisor que fornece comentários sobre o conteúdo
• Conecta agentes em um pipeline de fluxo de trabalho sequencial
• Transmite atualizações em tempo real à medida que os agentes processam solicitações

Conceitos abordados

• Agentes em Fluxos de Trabalho
• Arestas Diretas
• Construtor de Fluxos de Trabalho

Pré-requisitos

• Python 3.10 ou posterior
• Agent Framework instalado: pip install agent-framework
• Azure OpenAI Responses configurado com variáveis de ambiente adequadas
• Autenticação da CLI do Azure: az login

Etapa 1: Importar dependências necessárias

Comece por importar os componentes necessários para fluxos de trabalho e agentes Azure OpenAI Responses:

import asyncio
import os

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

Passo 2: Criar o Cliente de Respostas Azure OpenAI

Crie um cliente partilhado que possa usar para construir múltiplos agentes:

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

Passo 3: Criar Agentes Especializados

Crie dois agentes especializados para criação e revisão de conteúdo:

    # 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."
        ),
    )

Etapa 4: Criar o fluxo de trabalho

Ligue os agentes num fluxo de trabalho sequencial usando o construtor:

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

Etapa 5: Executar com streaming

Execute o fluxo de trabalho com streaming para observar atualizações em tempo real de ambos os 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)

Passo 6: Completar a função principal

Encapsule tudo na função principal com execução assíncrona correta.

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

Como funciona

1. Configuração do Cliente: Usa um FoundryChatClient com credenciais de CLI Azure para autenticação.
2. Criação de Agentes: Cria agentes Escritor e Revisor a partir da mesma configuração de cliente.
3. Processamento Sequencial: O agente escritor gera primeiro o conteúdo e depois passa-o ao agente revisor.
4. Atualizações de Streaming: Os eventos de saída (type="output") com dados de AgentResponseUpdate fornecem atualizações de tokens em tempo real enquanto os agentes geram respostas.

Conceitos-chave

• FoundryChatClient: Cliente partilhado usado para criar agentes de workflow com configuração consistente.
• WorkflowEvent: Os eventos de saída (type="output") contêm dados de saída do agente (AgentResponseUpdate para streaming, AgentResponse para não streaming).
• Fluxo de Trabalho Sequencial: Agentes ligados num pipeline onde a saída flui de um para o outro.

Implementação Completa

Para a implementação funcional completa, veja azure_ai_agents_streaming.py no repositório Agent Framework.

O que você vai construir

Você criará um fluxo de trabalho que:

• Utiliza agentes Azure OpenAI como executores de workflow
• Implementa um agente de tradução em francês
• Implementa um agente de tradução para espanhol
• Implementa um agente de tradução para inglês
• Conecta agentes em um pipeline de fluxo de trabalho sequencial
• Transmite atualizações em tempo real à medida que os agentes processam solicitações

Conceitos abordados

• Agentes em Fluxos de Trabalho
• Arestas Diretas
• Construtor de Fluxos de Trabalho

Pré-requisitos

• Vai 1.25 ou mais tarde
• Endpoint do projeto Microsoft Foundry e implementação do modelo configurados
• Autenticação da CLI do Azure ou outra origem de credenciais do Azure

Passo 1: Configurar a Configuração da Fundição

endpoint := os.Getenv("FOUNDRY_PROJECT_ENDPOINT")
model := cmp.Or(os.Getenv("FOUNDRY_MODEL"), "gpt-4o-mini")

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

Passo 2: Criar Função de Fábrica do Agente

Crie agentes com instruções específicas de tradução:

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"},
    })
}

Passo 3: Criar Agentes Especializados em Fundição

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

Etapa 4: Criar o fluxo de trabalho

Os agentes podem ser usados como executores de workflows, permitindo passos de workflow potenciados por IA.

Associe cada agente como executor de fluxo de trabalho, depois ligue os executores entre si com arestas:

import (
    "github.com/microsoft/agent-framework-go/agent"
    "github.com/microsoft/agent-framework-go/workflow/agentworkflow"
    "github.com/microsoft/agent-framework-go/message"
    "github.com/microsoft/agent-framework-go/workflow"
    "github.com/microsoft/agent-framework-go/workflow/inproc"
)

cfg := agentworkflow.Config{DisableForwardIncomingMessages: true}
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).
    WithOutputFrom(english).
    Build()
if err != nil {
    return err
}

Etapa 5: Executar com streaming

Execute o fluxo de trabalho e ative os eventos de atualização com um workflow.TurnToken:

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

emitEvents := true
if err := run.SendMessage(ctx, workflow.TurnToken{EmitEvents: &emitEvents}); err != nil {
    return err
}

for evt, err := range run.WatchStream(ctx) {
    if err != nil {
        return err
    }
    if out, ok := evt.(workflow.OutputEvent); ok {
        if update, ok := out.Output.(*agent.ResponseUpdate); ok {
            fmt.Printf("%s: %s\n", out.ExecutorID, update.String())
        }
    }
}

Como funciona

1. Configuração do Cliente: Utiliza uma credencial Azure com o cliente OpenAI.
2. Criação de Agentes: Cria agentes especializados com instruções específicas da língua.
3. Alojamento de Agentes: Utiliza agentworkflow.New para associar cada agente como executor de fluxo de trabalho.
4. Processamento Sequencial: O executor francês executa primeiro, depois espanhol, depois inglês.
5. Turn Token Pattern: Os agentes alojados armazenam mensagens e executam quando recebem um workflow.TurnToken.
6. Atualizações em transmissão: os valores de workflow.OutputEvent podem conter resultados de *agent.ResponseUpdate para indicar o progresso em tempo real.

Conceitos-chave

• Azure OpenAI Agent: Apoiado agent.Agent pelo Azure OpenAI.
• agentworkflow. Novo: Adapta um agente para ser usado como executor de workflow.
• fluxo de trabalho. TurnToken: Sinal que desencadeia agentes alojados para processar mensagens em buffer.
• Workflow OutputEvent: Transporta atualizações de resposta do agente e resultados finais do fluxo de trabalho.
• Fluxo de Trabalho Sequencial: Agentes ligados num pipeline onde a saída flui de um para o outro.

Implementação Completa

package main

import (
    "cmp"
    "context"
    "fmt"
    "log"
    "os"

    "github.com/microsoft/agent-framework-go/agent"
    "github.com/microsoft/agent-framework-go/workflow/agentworkflow"
    "github.com/microsoft/agent-framework-go/message"
    "github.com/microsoft/agent-framework-go/workflow"
    "github.com/microsoft/agent-framework-go/workflow/inproc"

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

func main() {
    ctx := context.Background()
    endpoint := os.Getenv("FOUNDRY_PROJECT_ENDPOINT")
    model := cmp.Or(os.Getenv("FOUNDRY_MODEL"), "gpt-4o-mini")

    credential, err := azidentity.NewDefaultAzureCredential(nil)
    if err != nil {
        log.Fatal(err)
    }

    newTranslationAgent := func(language string) *agent.Agent {
        return foundryprovider.NewAgent(endpoint, credential, 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{DisableForwardIncomingMessages: true}
    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).
        WithOutputFrom(english).
        Build()
    if err != nil {
        log.Fatal(err)
    }

    run, err := inproc.Default.RunStreaming(ctx, wf, message.NewText("Hello World"))
    if err != nil {
        log.Fatal(err)
    }
    defer run.Close(ctx)

    emitEvents := true
    if err := run.SendMessage(ctx, workflow.TurnToken{EmitEvents: &emitEvents}); err != nil {
        log.Fatal(err)
    }

    for evt, err := range run.WatchStream(ctx) {
        if err != nil {
            log.Fatal(err)
        }
        if out, ok := evt.(workflow.OutputEvent); ok {
            if update, ok := out.Output.(*agent.ResponseUpdate); ok {
                fmt.Printf("%s: %s\n", out.ExecutorID, update.String())
            }
        }
    }
}