워크플로의 에이전트

만들게 될 것들

다음과 같은 워크플로를 만듭니다.

• Azure Foundry 에이전트 서비스를 사용하여 지능형 에이전트 만들기
• 입력을 프랑스어로 번역하는 프랑스어 번역 에이전트를 구현합니다.
• 프랑스어를 스페인어로 번역하는 스페인어 번역 에이전트 구현
• 스페인어를 영어로 다시 번역하는 영어 번역 에이전트를 구현합니다.
• 순차 워크플로 파이프라인에서 에이전트 연결
• 에이전트가 요청을 처리할 때 실시간 업데이트를 스트리밍합니다.
• Azure Foundry 에이전트에 대한 적절한 리소스 정리를 보여 줍니다.

다루는 개념

• 워크플로의 에이전트
• 직접 연결
• 워크플로 작성기

필수 조건

• .NET 8.0 SDK 이상
• 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 AgentRunUpdateEvent 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 클라이언트 설정: 인증을 위해 Azure CLI 자격 증명과 함께 사용 PersistentAgentsClient
2. 에이전트 만들기: 변환에 대한 특정 지침을 사용하여 Azure Foundry에 영구 에이전트 만들기
3. 순차적 처리: 프랑스어 에이전트가 먼저 입력을 번역하고 그 다음 스페인어 에이전트가, 마지막으로 영어 에이전트가 번역합니다.
4. 토큰 패턴 전환: 에이전트는 메시지를 캐시하고 메시지를 받을 때만 처리합니다. TurnToken
5. 스트리밍 업데이트: AgentRunUpdateEvent 에이전트가 응답을 생성할 때 실시간 토큰 업데이트를 제공합니다.
6. 리소스 관리: 관리 API를 사용하여 Azure Foundry 에이전트의 적절한 정리

주요 개념

• Azure Foundry 에이전트 서비스: 고급 추론 기능이 있는 클라우드 기반 AI 에이전트
• PersistentAgentsClient: Azure Foundry에서 에이전트를 만들고 관리하기 위한 클라이언트
• AgentRunUpdateEvent: 에이전트 실행 중 실시간 스트리밍 업데이트
• TurnToken: 메시지 캐싱 후 에이전트 처리를 트리거하는 신호
• 순차 워크플로: 출력이 하나에서 다음으로 흐르는 파이프라인에 연결된 에이전트

전체 구현

이 Azure Foundry 에이전트 워크플로의 전체 작업 구현은 Agent Framework 리포지토리의 FoundryAgent Program.cs 샘플을 참조하세요.

만들게 될 것들

다음과 같은 워크플로를 만듭니다.

• Azure AI 에이전트 서비스를 사용하여 지능형 에이전트 만들기
• 프롬프트에 따라 콘텐츠를 생성하는 작성기 에이전트를 구현합니다.
• 콘텐츠에 대한 피드백을 제공하는 검토자 에이전트를 구현합니다.
• 순차 워크플로 파이프라인에서 에이전트 연결
• 에이전트가 요청을 처리할 때 실시간 업데이트를 스트리밍합니다.
• Azure AI 클라이언트에 대한 적절한 비동기 컨텍스트 관리를 보여 줍니다.

다루는 개념

• 워크플로의 에이전트
• 직접 연결
• 워크플로 작성기

필수 조건

• Python 3.10 이상
• 에이전트 프레임워크 설치: pip install agent-framework-azure-ai --pre
• 적절한 환경 변수로 구성된 Azure AI 에이전트 서비스
• Azure CLI 인증: az login

1단계: 필요한 종속성 가져오기

먼저 Azure AI 에이전트 및 워크플로에 필요한 구성 요소를 가져옵니다.

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

2단계: Azure AI 에이전트 팩터리 만들기

적절한 비동기 컨텍스트 처리를 사용하여 Azure AI 에이전트 만들기를 관리하는 도우미 함수를 만듭니다.

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

3단계: 특수한 Azure AI 에이전트 만들기

콘텐츠 만들기 및 검토를 위한 두 개의 특수 에이전트를 만듭니다.

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단계: 워크플로 빌드

흐름 작성기를 사용하여 순차 워크플로에서 에이전트를 연결합니다.

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

6단계: 주 함수 완료

적절한 비동기 실행을 사용하여 주 함수의 모든 항목을 래핑합니다.

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

작동 방식

1. Azure AI 클라이언트 설정: 인증을 위해 Azure CLI 자격 증명과 함께 사용 AzureAIAgentClient
2. 에이전트 팩터리 패턴: 여러 에이전트에 대한 비동기 컨텍스트 수명 주기를 관리하는 팩터리 함수를 만듭니다.
3. 순차적 처리: 기록기 에이전트가 먼저 콘텐츠를 생성한 다음 검토자 에이전트에 전달합니다.
4. 스트리밍 업데이트: AgentRunUpdateEvent 에이전트가 응답을 생성할 때 실시간 토큰 업데이트를 제공합니다.
5. 컨텍스트 관리: AsyncExitStack를 사용하여 Azure AI 리소스의 적절한 정리

주요 개념

• Azure AI 에이전트 서비스: 고급 추론 기능이 있는 클라우드 기반 AI 에이전트
• AgentRunUpdateEvent: 에이전트 실행 중 실시간 스트리밍 업데이트
• AsyncExitStack: 여러 리소스에 대한 적절한 비동기 컨텍스트 관리
• 에이전트 팩터리 패턴: 공유 클라이언트 구성을 사용하여 다시 사용할 수 있는 에이전트 만들기
• 순차 워크플로: 출력이 하나에서 다음으로 흐르는 파이프라인에 연결된 에이전트

전체 구현

이 Azure AI 에이전트 워크플로의 전체 작업 구현은 Agent Framework 리포지토리의 azure_ai_agents_streaming.py 샘플을 참조하세요.