Agenti nei flussi di lavoro

Cosa costruirai

Verrà creato un flusso di lavoro che:

• Usa il servizio agente di Azure Foundry per creare agenti intelligenti
• Implementa un agente di traduzione francese che converte l'input in francese
• Implementa un agente di traduzione spagnolo che converte il francese in spagnolo
• Implementa un agente di traduzione inglese che converte lo spagnolo in inglese
• Connette gli agenti in una pipeline del flusso di lavoro sequenziale
• Trasmette gli aggiornamenti in tempo reale come agenti elaborano le richieste
• Illustra la pulizia delle risorse appropriata per gli agenti di Azure Foundry

Concetti trattati

• Agenti nei flussi di lavoro
• Bordi diretti
• Generatore flussi di lavoro

Prerequisiti

• .NET 8.0 SDK o versione successiva
• Un endpoint del progetto e un modello di Azure Foundry configurati
• Interfaccia della riga di comando di Azure installata e autenticata (per l'autenticazione delle credenziali di Azure)
• Nuova applicazione console

Passaggio 1: Installare pacchetti NuGet

Prima di tutto, installare i pacchetti necessari per il progetto .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

Passaggio 2: Configurare il client Azure Foundry

Configurare il client Azure Foundry con variabili di ambiente e autenticazione:

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

Passaggio 3: Creare un metodo Factory per l'agente

Implementare un metodo helper per creare agenti Azure Foundry con istruzioni specifiche:

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

Passaggio 4: Creare agenti specializzati di Azure Foundry

Creare tre agenti di traduzione usando il metodo helper:

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

Passaggio 5: Compilare il flusso di lavoro

Connettere gli agenti in un flusso di lavoro sequenziale usando WorkflowBuilder:

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

Passaggio 6: Eseguire con lo streaming

Eseguire il flusso di lavoro con lo streaming per osservare gli aggiornamenti in tempo reale da tutti gli agenti:

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

Passaggio 7: Pulizia delle risorse

Pulire correttamente gli agenti di Azure Foundry dopo l'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);
    }

Funzionamento

1. Configurazione client di Azure Foundry: Utilizza le credenziali dell'interfaccia della riga di comando (CLI) di Azure per l'autenticazione
2. Creazione dell'agente: crea agenti con controllo delle versioni in Azure Foundry con istruzioni specifiche per la traduzione
3. Elaborazione sequenziale: l'agente francese converte prima l'input, quindi l'agente spagnolo, quindi l'agente inglese
4. Turn Token Pattern: Gli agenti memorizzano nella cache i messaggi e li elaborano solo quando ricevono un TurnToken
5. Aggiornamenti di streaming: AgentResponseUpdateEvent fornisce aggiornamenti dei token in tempo reale man mano che gli agenti generano risposte
6. Gestione risorse: pulizia corretta degli agenti di Azure Foundry tramite l'API di amministrazione

Concetti chiave

• Servizio agente di Azure Foundry: agenti di intelligenza artificiale basati sul cloud con funzionalità di ragionamento avanzate
• AIProjectClient: client per la creazione e la gestione degli agenti in Azure Foundry
• WorkflowEvent: gli eventi di output (type="output") contengono dati di output dell'agente (AgentResponseUpdate per lo streaming, AgentResponse per non streaming)
• TurnToken: Indica l'avvio dell'elaborazione dell'agente dopo la memorizzazione nella cache dei messaggi
• Flusso di lavoro sequenziale: agenti connessi in una pipeline in cui l'output passa da uno a quello successivo

Implementazione completa

Per l'implementazione completa di questo flusso di lavoro degli agenti di Azure Foundry, vedere l'esempio foundryAgent Program.cs nel repository di Agent Framework.

Cosa costruirai

Verrà creato un flusso di lavoro che:

• FoundryChatClient Usa per creare agenti intelligenti
• Implementa un agente writer che crea contenuto in base alle richieste
• Implementa un agente reviewer che fornisce commenti e suggerimenti sul contenuto
• Connette gli agenti in una pipeline del flusso di lavoro sequenziale
• Trasmette gli aggiornamenti in tempo reale come agenti elaborano le richieste

Concetti trattati

• Agenti nei flussi di lavoro
• Bordi diretti
• Generatore flussi di lavoro

Prerequisiti

• Python 3.10 o versione successiva
• Agent Framework installato: pip install agent-framework
• Risposte OpenAI di Azure configurate con variabili di ambiente appropriate
• Autenticazione dell'interfaccia della riga di comando di Azure: az login

Passaggio 1: Importare le dipendenze necessarie

Per iniziare, importare i componenti necessari per i flussi di lavoro e gli agenti risposte OpenAI di Azure:

import asyncio
import os

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

Passaggio 2: Creare il client di risposte di Azure OpenAI

Creare un client condiviso che è possibile usare per costruire più agenti:

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

Passaggio 3: Creare agenti specializzati

Creare due agenti specializzati per la creazione e la revisione del contenuto:

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

Passaggio 4: Compilare il flusso di lavoro

Connettere gli agenti in un flusso di lavoro sequenziale usando il generatore:

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

Passaggio 5: Eseguire con lo streaming

Eseguire il flusso di lavoro con streaming per osservare gli aggiornamenti in tempo reale da entrambi gli agenti:

    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)

Passaggio 6: Completare la funzione principale

Racchiudere tutto nella funzione principale con un'esecuzione asincrona appropriata:

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

Funzionamento

1. Installazione client: utilizza uno FoundryChatClient con le credenziali di interfaccia della riga di comando di Azure per l'autenticazione.
2. Creazione dell'agente: crea agenti di scrittura e di revisione dalla stessa configurazione client.
3. Elaborazione sequenziale: l'agente writer genera prima il contenuto, quindi lo passa all'agente Reviewer.
4. Aggiornamenti di Streaming: gli eventi di output (type="output") con i dati di AgentResponseUpdate forniscono aggiornamenti dei token in tempo reale man mano che gli agenti generano risposte.

Concetti chiave

• FoundryChatClient: client condiviso usato per creare agenti del flusso di lavoro con una configurazione coerente.
• WorkflowEvent: gli eventi di output (type="output") contengono dati di output dell'agente (AgentResponseUpdate per lo streaming, AgentResponse per il non streaming).
• Flusso di lavoro sequenziale: agenti connessi in una pipeline in cui l'output passa da uno a quello successivo.

Implementazione completa

Per l'implementazione di lavoro completa, vedere azure_ai_agents_streaming.py nel repository di Agent Framework.

Cosa costruirai

Verrà creato un flusso di lavoro che:

• Usa agenti Azure OpenAI come esecutori del flusso di lavoro
• Implementa un agente di traduzione francese
• Implementa un agente di traduzione spagnolo
• Implementa un agente di traduzione in inglese
• Connette gli agenti in una pipeline del flusso di lavoro sequenziale
• Trasmette gli aggiornamenti in tempo reale come agenti elaborano le richieste

Concetti trattati

• Agenti nei flussi di lavoro
• Bordi diretti
• Generatore flussi di lavoro

Prerequisiti

• Passare alla versione 1.25 o successiva
• Endpoint del progetto Microsoft Foundry e distribuzione del modello configurati
• autenticazione di interfaccia della riga di comando di Azure o un'altra origine delle credenziali di Azure

Passaggio 1: Configurare la configurazione di Foundry

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
}

Passaggio 2: Creare la funzione factory dell'agente

Creare agenti con istruzioni di traduzione specifiche:

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

Passaggio 3: Creare agenti foundry specializzati

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

Passaggio 4: Compilare il flusso di lavoro

Gli agenti possono essere usati come executor del flusso di lavoro, abilitando i passaggi del flusso di lavoro basati su intelligenza artificiale.

Associa ogni agente come esecutore del flusso di lavoro, quindi collega gli esecutori con archi:

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
}

Passaggio 5: Eseguire con lo streaming

Esegui il flusso di lavoro e abilita gli eventi di aggiornamento con un 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())
        }
    }
}

Funzionamento

1. Installazione client: usa una credenziale Azure con il client OpenAI.
2. Creazione dell'agente: crea agenti specializzati con istruzioni specifiche del linguaggio.
3. Hosting dell'agente: usa agentworkflow.New per collegare ogni agente come esecutore del flusso di lavoro.
4. Elaborazione sequenziale: l'esecutore francese viene eseguito per primo, poi quello spagnolo, quindi quello inglese.
5. Turn Token Pattern: gli agenti ospitati memorizzano i messaggi nel buffer e vengono eseguiti quando ricevono un workflow.TurnToken.
6. Aggiornamenti in streaming: i valori workflow.OutputEvent possono contenere *agent.ResponseUpdate risultati per mostrare l'avanzamento in tempo reale.

Concetti chiave

• Agente Azure OpenAI: un agent.Agent basato su Azure OpenAI.
• agentworkflow.New: adatta un agente per l'uso come esecutore del flusso di lavoro.
• workflow.TurnToken: Segnale che attiva gli agenti ospitati per elaborare i messaggi nel buffer.
• OutputEvent del flusso di lavoro: contiene gli aggiornamenti delle risposte dell'agente e gli output finali del flusso di lavoro.
• Flusso di lavoro sequenziale: agenti connessi in una pipeline in cui l'output passa da uno a quello successivo.

Implementazione 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())
            }
        }
    }
}