Агенты в рабочих процессах

В этом руководстве показано, как интегрировать агенты ИИ в рабочие процессы с помощью Agent Framework. Вы узнаете, как создавать рабочие процессы, использующие возможности специализированных агентов ИИ для создания контента, проверки и других задач совместной работы.

Что вы будете создавать

Вы создадите рабочий процесс, который:

• Использует службу агента Azure Foundry для создания интеллектуальных агентов
• Реализует агент перевода на французский язык, который переводит входные данные на французский
• Реализует агент перевода на испанский язык, который переводит французский на испанский
• Реализует агент перевода английского языка, который переводит испанский обратно на английский
• Подключает агенты в последовательном конвейере рабочего процесса
• Потоковая передача обновлений в режиме реального времени во время обработки запросов агентами
• Демонстрирует правильную очистку ресурсов для агентов Azure Foundry

Основные понятия, описанные в статье

• Агенты в рабочих процессах
• Прямые края
• Построитель рабочих процессов

Предпосылки

• Пакет SDK для .NET 8.0 или более поздней версии
• Конечная точка и развертывание службы Azure Foundry настроены
• Установленный и прошедший проверку подлинности Azure CLI (для проверки подлинности учетных данных Azure)
• Новое консольное приложение

Шаг 1. Установка пакетов NuGet

Сначала установите необходимые пакеты для проекта .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

Шаг 2. Настройка клиента Azure Foundry

Настройте клиент Azure Foundry с переменными среды и проверкой подлинности:

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

Шаг 3. Создание метода фабрики агентов

Реализуйте вспомогательный метод для создания агентов Azure Foundry с определенными инструкциями:

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

Шаг 4. Создание специализированных агентов Azure Foundry

Создайте три агента перевода с помощью вспомогательного метода:

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

Шаг 5. Создание рабочего процесса

Подключите агенты в последовательном рабочем процессе с помощью WorkflowBuilder:

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

Шаг 6. Выполнение с помощью потоковой передачи

Запустите рабочий процесс с потоковой передачей, чтобы наблюдать обновления в режиме реального времени со всех агентов:

        // 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 AgentResponseUpdateEvent executorComplete)
            {
                Console.WriteLine($"{executorComplete.ExecutorId}: {executorComplete.Data}");
            }
        }

Шаг 7. Очистка ресурсов

Правильно очистите агенты Azure Foundry после использования:

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

Принцип работы

1. Настройка клиента Azure Foundry: используется PersistentAgentsClient с учетными данными Azure CLI для проверки подлинности
2. Создание агента: Создает постоянные агенты в Azure Foundry с определенными инструкциями по переводу
3. Последовательная обработка: французский агент сначала преобразует входные данные, а затем испанский агент, а затем английский агент
4. Шаблон обработки токена: агенты кэшируют сообщения и обрабатывают их только при получении TurnToken
5. Потоковая передача обновлений: AgentResponseUpdateEvent предоставляет обновления маркеров в режиме реального времени по мере того как агенты создают ответы.
6. Управление ресурсами. Правильная очистка агентов Azure Foundry с помощью API администрирования

Основные понятия

• Служба агента Azure Foundry: агенты искусственного интеллекта на основе облака с расширенными возможностями анализа
• PersistentAgentsClient: клиент для создания агентов и управления ими в Azure Foundry
• AgentResponseUpdateEvent: обновления в реальном времени во время исполнения агента
• TurnToken: сигнал, который активирует обработку агента после кэширования сообщений
• Последовательный рабочий процесс: агенты, соединённые в цепочку, где выходные данные передаются от одного агента к следующему

Полная реализация

Полный поток выполнения агентов Azure Foundry можно найти в программе FoundryAgent в файле Program.cs в репозитории Agent Framework.

Что вы будете создавать

Вы создадите рабочий процесс, который:

• Использование службы агента ИИ Azure для создания интеллектуальных агентов
• Реализует агент-писатель, который генерирует контент на основе сигналов
• Реализует агент рецензента, предоставляющий отзывы о содержимом
• Подключает агенты в последовательном конвейере рабочего процесса
• Потоковая передача обновлений в режиме реального времени во время обработки запросов агентами
• Демонстрирует правильное асинхронное управление контекстом для клиентов ИИ Azure

Основные понятия, описанные в статье

• Агенты в рабочих процессах
• Прямые края
• Построитель рабочих процессов

Предпосылки

• Python 3.10 или более поздней версии
• Установленная платформа агента: pip install agent-framework-azure-ai --pre
• Служба агента ИИ Azure, настроенная с соответствующими переменными среды
• Проверка подлинности Azure CLI: az login

Шаг 1. Импорт необходимых зависимостей

Начните с импорта необходимых компонентов для агентов и рабочих процессов ИИ Azure:

import asyncio
from collections.abc import Awaitable, Callable
from contextlib import AsyncExitStack
from typing import Any

from agent_framework import AgentResponseUpdateEvent, WorkflowBuilder, WorkflowOutputEvent
from agent_framework.azure import AzureAIAgentClient
from azure.identity.aio import AzureCliCredential

Шаг 2. Создание фабрики агентов ИИ Azure

Создайте вспомогательную функцию для управления созданием агента ИИ Azure с правильной асинхронной обработкой контекста:

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.as_agent(**kwargs))

    async def close() -> None:
        await stack.aclose()

    return agent, close

Шаг 3. Создание специализированных агентов ИИ Azure

Создайте два специализированных агента для создания контента и проверки:

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

Шаг 4. Создание рабочего процесса

Подключите агенты в последовательном рабочем процессе с помощью построителя fluent:

        # Build the workflow with agents as executors
        workflow = WorkflowBuilder().set_start_executor(writer).add_edge(writer, reviewer).build()

Шаг 5. Выполнение с помощью потоковой передачи

Запустите рабочий процесс с потоковой передачей, чтобы наблюдать обновления в режиме реального времени от обоих агентов:

        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, AgentResponseUpdateEvent):
                # 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()

Шаг 6. Полная основная функция

Оберните все в главной функции с корректной асинхронной обработкой.

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

Принцип работы

1. Настройка клиента ИИ Azure: используется AzureAIAgentClient с учетными данными Azure CLI для проверки подлинности
2. Шаблон фабрики агентов: создает фабричную функцию, которая управляет жизненным циклом асинхронного контекста для нескольких агентов
3. Последовательная обработка: сначала агент письменного контента создает материал, затем передает его агенту проверки.
4. Потоковая передача обновлений: AgentResponseUpdateEvent предоставляет обновления маркеров в режиме реального времени по мере того как агенты создают ответы.
5. Управление контекстом: правильная очистка ресурсов ИИ Azure с помощью AsyncExitStack

Основные понятия

• Служба агента ИИ Azure: облачные агенты ИИ с расширенными возможностями анализа
• AgentResponseUpdateEvent: обновления в режиме реального времени при потоковом выполнении агента
• AsyncExitStack: правильное асинхронное управление контекстом для нескольких ресурсов
• Шаблон фабрики агентов: повторное создание агента с общей конфигурацией клиента
• Последовательный рабочий процесс: агенты, соединённые в цепочку, где выходные данные передаются от одного агента к следующему

Полная реализация

Полную рабочую реализацию этого рабочего процесса агентов ИИ Azure см. в примере azure_ai_agents_streaming.py в репозитории Agent Framework.

Дальнейшие шаги

Узнайте о ветвлениях в рабочих процессах