Condividi tramite

Agenti nei flussi di lavoro

Questa esercitazione illustra come integrare gli agenti di intelligenza artificiale nei flussi di lavoro usando Agent Framework. Si apprenderà come creare flussi di lavoro che sfruttano la potenza degli agenti di intelligenza artificiale specializzati per la creazione, la revisione e altre attività di collaborazione.

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

Prerequisiti

Passaggio 1: Installare pacchetti NuGet

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

Passaggio 2: Configurare il client Azure Foundry

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

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

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

Passaggio 4: Creare agenti specializzati di Azure Foundry

Creare tre agenti di traduzione usando il metodo helper:

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

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

Passaggio 7: Pulizia delle risorse

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

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 persistenti 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: AgentRunUpdateEvent 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
  • PersistentAgentsClient: client per la creazione e la gestione di agenti in Azure Foundry
  • AgentRunUpdateEvent: aggiornamenti di streaming in tempo reale durante l'esecuzione dell'agente
  • 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:

  • Usa il servizio Agente di intelligenza artificiale di Azure 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
  • Illustra la corretta gestione del contesto asincrono per i client di intelligenza artificiale di Azure

Concetti trattati

Prerequisiti

  • Python 3.10 o versione successiva
  • Agent Framework installato: pip install agent-framework-azure-ai --pre
  • Servizio Agente di intelligenza artificiale di Azure configurato 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 gli agenti e i flussi di lavoro di Intelligenza artificiale di Azure:

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

Passaggio 2: Creare la factory dell'agente di intelligenza artificiale di Azure

Creare una funzione helper per gestire la creazione dell'agente di Intelligenza artificiale di Azure con la gestione corretta del contesto asincrono:

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

Passaggio 3: Creare agenti di intelligenza artificiale di Azure specializzati

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

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

Passaggio 4: Compilare il flusso di lavoro

Connettere gli agenti in un flusso di lavoro sequenziale usando fluent builder:

        # Build the workflow with agents as executors
        workflow = WorkflowBuilder().set_start_executor(writer).add_edge(writer, reviewer).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_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()

Passaggio 6: Completare la funzione principale

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

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

Funzionamento

  1. Configurazione del client di Azure AI: utilizza AzureAIAgentClient con le credenziali di Azure CLI per l'autenticazione
  2. Modello Agent Factory: crea una funzione di fabbrica che gestisce il ciclo di vita del contesto asincrono per più agenti
  3. Elaborazione sequenziale: l'agente writer genera prima il contenuto, quindi lo passa all'agente Reviewer
  4. Aggiornamenti di streaming: AgentRunUpdateEvent fornisce aggiornamenti dei token in tempo reale man mano che gli agenti generano risposte
  5. Gestione del contesto: pulizia corretta delle risorse di Intelligenza artificiale di Azure con AsyncExitStack

Concetti chiave

  • Servizio agente di intelligenza artificiale di Azure: agenti di intelligenza artificiale basati sul cloud con funzionalità di ragionamento avanzate
  • AgentRunUpdateEvent: aggiornamenti di streaming in tempo reale durante l'esecuzione dell'agente
  • AsyncExitStack: gestione corretta del contesto asincrono per più risorse
  • Modello di factory dell'agente: creazione di agenti riutilizzabili con configurazione client condivisa
  • 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 Intelligenza artificiale di Azure, vedere l'esempio di azure_ai_agents_streaming.py nel repository di Agent Framework.

Passaggi successivi

Informazioni sulla diramazione nei flussi di lavoro

  • Last updated on