워크플로의 에이전트

이 자습서에서는 Agent Framework를 사용하여 워크플로에 AI 에이전트를 통합하는 방법을 보여 줍니다. 콘텐츠 만들기, 검토 및 기타 공동 작업을 위해 특수한 AI 에이전트의 기능을 활용하는 워크플로를 만드는 방법을 알아봅니다.

만들게 될 것들

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

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

다루는 개념

필수 조건

1단계: NuGet 패키지 설치

먼저 .NET 프로젝트에 필요한 패키지를 설치합니다.

dotnet add package Microsoft.Agents.AI.Foundry --prerelease
dotnet add package Microsoft.Agents.AI.Workflows --prerelease

2단계: Azure Foundry 클라이언트 설정

환경 변수 및 인증을 사용하여 Azure Foundry 클라이언트를 구성합니다.

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

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

4단계: 특수화된 Azure Foundry 에이전트 만들기

도우미 메서드를 사용하여 세 개의 번역 에이전트를 만듭니다.

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

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

7단계: 리소스 정리

사용 후 Azure Foundry 에이전트를 올바르게 정리합니다.

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

작동 방식

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

주요 개념

  • Azure Foundry 에이전트 서비스: 고급 추론 기능이 있는 클라우드 기반 AI 에이전트
  • AIProjectClient: Azure Foundry에서 에이전트를 만들고 관리하기 위한 클라이언트
  • WorkflowEvent: 출력 이벤트(type="output")에는 에이전트 출력 데이터(AgentResponseUpdate 스트리밍의 경우, AgentResponse 비 스트리밍용)가 포함됩니다.
  • TurnToken: 메시지 캐싱 후 에이전트 처리를 트리거하는 신호
  • 순차 워크플로: 출력이 하나에서 다음으로 흐르는 파이프라인에 연결된 에이전트

전체 구현

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

만들게 될 것들

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

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

다루는 개념

필수 조건

  • Python 3.11 이상
  • 에이전트 프레임워크 설치: pip install agent-framework
  • 적절한 환경 변수로 구성된 Azure OpenAI 응답
  • Azure CLI 인증: az login

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

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

import asyncio
import os

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

2단계: Azure OpenAI 응답 클라이언트 만들기

여러 에이전트를 생성하는 데 사용할 수 있는 하나의 공유 클라이언트를 만듭니다.

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

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

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

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

4단계: 워크플로 빌드

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

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

5단계: 스트리밍으로 실행

스트리밍을 사용하여 워크플로를 실행하여 두 에이전트의 실시간 업데이트를 관찰합니다.

    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)

6단계: 주 함수 완료

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

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

작동 방식

  1. 클라이언트 설정: Azure CLI 자격 증명을 사용하여 인증하기 위해 하나를 사용합니다 FoundryChatClient.
  2. 에이전트 만들기: 동일한 클라이언트 구성에서 기록기 및 검토자 에이전트를 만듭니다.
  3. 순차적 처리: 기록기 에이전트는 먼저 콘텐츠를 생성한 다음 검토자 에이전트에 전달합니다.
  4. 스트리밍 업데이트: 데이터와 함께 type="output" 출력 이벤트(AgentResponseUpdate)는 에이전트가 응답을 생성할 때 실시간 토큰 업데이트를 제공합니다.

주요 개념

  • FoundryChatClient: 일관된 구성으로 워크플로 에이전트를 만드는 데 사용되는 공유 클라이언트입니다.
  • WorkflowEvent: 출력 이벤트(type="output")에는 에이전트 출력 데이터(AgentResponseUpdate 스트리밍, AgentResponse 비 스트리밍용)가 포함됩니다.
  • 순차 워크플로: 출력이 1에서 다음으로 흐르는 파이프라인에 연결된 에이전트입니다.

전체 구현

전체 작업 구현은 Agent Framework 리포지토리의 azure_ai_agents_streaming.py 참조하세요.

만들게 될 것들

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

  • Microsoft Foundry 모델 배포 에이전트를 워크플로 실행자로 사용
  • 프랑스어 번역 에이전트 구현
  • 스페인어 번역 에이전트 구현
  • 영어 번역 에이전트 구현
  • 순차 워크플로 파이프라인에서 에이전트 연결
  • 에이전트가 요청을 처리할 때 실시간 업데이트를 스트리밍합니다.

다루는 개념

필수 조건

  • Go 1.26 이상 버전
  • Microsoft Foundry 프로젝트 엔드포인트 및 모델 배포가 구성됨
  • Azure CLI 인증 또는 다른 Azure 자격 증명 원본

새 디렉터리에서 Go 모듈을 초기화하고 에이전트 프레임워크를 설치합니다.

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

프로젝트 엔드포인트 및 배포된 모델의 이름을 설정합니다.

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

1단계: Foundry 구성 설정

다음 섹션에서는 전체 구현에서 runExample 함수를 빌드합니다. 먼저 가져오기 및 구성 시작:

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
}

Warning

azidentity.NewDefaultAzureCredential 은 개발에 편리하지만 프로덕션 환경에서 신중하게 고려해야 합니다. 프로덕션 환경에서는 대기 시간 문제, 의도하지 않은 자격 증명 검색 및 대체 메커니즘의 잠재적인 보안 위험을 방지하기 위해 특정 자격 증명(예: azidentity.NewManagedIdentityCredential)을 사용하는 것이 좋습니다.

2단계: 에이전트 팩터리 함수 만들기

특정 번역 지침을 사용하여 에이전트를 만듭니다.

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

3단계: 특수한 파운드리 에이전트 만들기

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

4단계: 워크플로 빌드

에이전트를 워크플로 실행자로 사용하여 AI 기반 워크플로 단계를 사용하도록 설정할 수 있습니다.

각 에이전트를 워크플로 실행기로 바인딩한 다음, 실행기를 에지와 연결합니다.

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) 는 해당 에이전트의 입력을 전달하지 않고 각 에이전트의 응답을 다운스트림으로 전달합니다. 호스트된 각 실행기는 자체 에이전트 세션을 관리합니다.

호출자가 세 에이전트를 모두 관찰할 수 있도록 프랑스어 및 스페인어를 중간 출력 원본으로 등록합니다. 영어는 터미널 출력 소스로 남아 있습니다. 출력 지정은 어떤 메시지가 엣지를 따라 흐르는지가 아니라 호출자가 보는 것을 제어합니다.

5단계: 스트리밍으로 실행

워크플로를 실행하고 workflow.TurnToken을(를) 사용하여 업데이트 이벤트를 활성화하세요. 워크플로 실패 이벤트 및 이벤트 반복기의 오류를 처리합니다.

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

출력에는 실행기별로 그룹화된 프랑스어, 스페인어 및 영어 번역이 포함됩니다. 전체 예제는 치명적인 오류를 기록하기 전 runExample에서 돌아가므로 지연된 워크플로 정리 작업이 계속 실행됩니다.

C# 예제와 달리 이 Go 예제에서는 Foundry에서 삭제해야 하는 저장된 에이전트 버전을 만들지 않습니다.

작동 방식

  1. 클라이언트 설정: Foundry 프로젝트 엔드포인트, 모델 배포 및 Azure 자격 증명을 사용합니다.
  2. 에이전트 만들기: 언어별 지침이 포함된 특수 에이전트를 만듭니다.
  3. 에이전트 호스팅: 각 에이전트를 워크플로 실행기로 바인딩하는 데 사용합니다 agentworkflow.New .
  4. 순차적 처리: 프랑스어 실행기가 먼저 실행된 다음 스페인어, 영어가 실행됩니다.
  5. 턴 토큰 패턴: 호스팅된 에이전트는 메시지를 버퍼링한 후 workflow.TurnToken를 받으면 실행됩니다.
  6. 스트리밍 업데이트: 중간 및 최종 출력 등록은 세 에이전트 모두의 업데이트를 노출합니다.
  7. 오류 처리: 호스트는 ErrorEvent 및 ExecutorFailedEvent을(를) 확인하고 워크플로 실행을 종료한 후에야 오류를 보고합니다.

주요 개념

  • Foundry 에이전트: Foundry 모델 배포에서 백업되는 agent.Agent
  • agentworkflow. 새 기능: 워크플로 실행기로 사용할 에이전트를 조정합니다.
  • 워크플로. TurnToken: 버퍼링된 메시지를 처리하기 위해 호스트된 에이전트를 트리거하는 신호입니다.
  • 워크플로 OutputEvent: 에이전트 응답 업데이트 및 최종 워크플로 출력을 전달합니다.
  • 순차 워크플로: 출력이 1에서 다음으로 흐르는 파이프라인에 연결된 에이전트입니다.

전체 구현

이 예제를 Go 모듈에 main.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
}

예제를 실행하세요:

go run .

Warning

azidentity.NewDefaultAzureCredential 은 개발에 편리하지만 프로덕션 환경에서 신중하게 고려해야 합니다. 프로덕션 환경에서는 대기 시간 문제, 의도하지 않은 자격 증명 검색 및 대체 메커니즘의 잠재적인 보안 위험을 방지하기 위해 특정 자격 증명(예: azidentity.NewManagedIdentityCredential)을 사용하는 것이 좋습니다.

Tip

다른 실행 가능한 번역 파이프라인은 워크플로의 에이전트 샘플을 참조하세요.

다음 단계