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

Prerequisites

• .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 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("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());

Passaggio 3: Creare un metodo di Agent Factory

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

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.Name);
        await aiProjectClient.AgentAdministrationClient.DeleteAgentAsync(spanishAgent.Name);
        await aiProjectClient.AgentAdministrationClient.DeleteAgentAsync(englishAgent.Name);
    }

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

Prerequisites

• Python 3.11 o versioni successive
• 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 di distribuzione dei modelli di Microsoft Foundry come executor 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

Prerequisites

• Go 1.26 o versioni successive
• 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

In una nuova directory inizializzare un modulo Go e installare Agent Framework:

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

Impostare l'endpoint del progetto e il nome del modello distribuito:

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

Passaggio 1: Configurare la configurazione di Foundry

Le sezioni seguenti creano la funzione runExample nell'implementazione completa. Inizia con le sue importazioni e la sua configurazione:

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
}

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.

Associare ogni agente come executor del flusso di lavoro, quindi collegare gli executor con edge:

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) passa la risposta di ogni agente a valle senza inoltrare anche l'input dell'agente. Ogni executor ospitato gestisce la propria sessione dell'agente.

Registra francese e spagnolo come origini di output intermedie in modo che i chiamanti possano osservare tutti e tre gli agenti. L'inglese rimane l'origine dell'output del terminale. La designazione di output controlla ciò che il chiamante vede, non quali messaggi scorrono lungo gli edge.

Passaggio 5: Eseguire con lo streaming

Eseguire il flusso di lavoro e abilitare gli eventi di aggiornamento con un workflow.TurnToken. Gestire gli eventi di fallimento del flusso di lavoro e gli errori dell'iteratore dell'evento:

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

L'output contiene traduzioni in francese, spagnolo e inglese, raggruppate per executor. L'esempio completo ritorna da runExample prima di registrare un errore irreversibile, quindi la pulizia posticipata del flusso di lavoro viene comunque eseguita.

A differenza dell'esempio C#, questo esempio go non crea versioni dell'agente archiviate che richiedono l'eliminazione in Foundry.

Funzionamento

1. Installazione client: usa un endpoint del progetto Foundry, la distribuzione del modello e le credenziali Azure.
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'executor francese viene eseguito per primo, poi quello spagnolo, quindi quello inglese.
5. Modello di token di turno: gli agenti ospitati memorizzano i messaggi nel buffer e vengono eseguiti quando ricevono un workflow.TurnToken.
6. Aggiornamenti di streaming: le registrazioni di output intermedie e del terminale espongono gli aggiornamenti di tutti e tre gli agenti.
7. Gestione degli errori: l'host controlla ErrorEvent e ExecutorFailedEvent e chiude l'esecuzione del flusso di lavoro prima di segnalare un errore.

Concetti chiave

• Agente Foundry: un agent.Agent basato su una distribuzione di modelli Foundry.
• 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

Salvare questo esempio come main.go nel modulo 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
}

Esegui l'esempio:

go run .