다음을 통해 공유

Microsoft 에이전트 프레임워크 워크플로 오케스트레이션 - 그룹 채팅

그룹 채팅 오케스트레이션은 화자 선택 및 대화 흐름을 결정하는 오케스트레이터가 조정하는 여러 에이전트 간의 공동 대화를 모델로 합니다. 이 패턴은 반복적인 구체화, 공동 작업 문제 해결 또는 다중 관점 분석이 필요한 시나리오에 적합합니다.

내부적으로, 그룹 채팅 오케스트레이션은 중앙에 오케스트레이터가 있는 별 형 토폴로지에서 에이전트를 조립합니다. 오케스트레이터는 라운드 로빈, 프롬프트 기반 선택 또는 대화 컨텍스트를 기반으로 하는 사용자 지정 논리와 같이 다음에 말할 에이전트를 선택하기 위한 다양한 전략을 구현하여 다중 에이전트 협업을 위한 유연하고 강력한 패턴을 만들 수 있습니다.

그룹 채팅 관리

그룹 채팅과 기타 패턴의 차이점

그룹 채팅 오케스트레이션에는 다른 다중 에이전트 패턴에 비해 고유한 특성이 있습니다.

  • 중앙 집중식 조정: 에이전트가 직접 제어를 전송하는 핸드오프 패턴과 달리 그룹 채팅은 오케스트레이터를 사용하여 다음에 말하는 사람을 조정합니다.
  • 반복 구체화: 에이전트는 여러 번의 라운드에 걸쳐 서로의 응답을 검토하고 발전시킬 수 있습니다.
  • 유연한 발표자 선택: 오케스트레이터는 다양한 전략(라운드 로빈, 프롬프트 기반 사용자 지정 논리)을 사용하여 스피커를 선택할 수 있습니다.
  • 공유 컨텍스트: 모든 에이전트가 전체 대화 기록을 확인하여 공동 작업을 구체화할 수 있습니다.

학습 내용

  • 그룹 공동 작업을 위한 특수 에이전트를 만드는 방법
  • 화자 선택 전략을 구성하는 방법
  • 반복 에이전트 구체화를 사용하여 워크플로를 빌드하는 방법
  • 사용자 지정 오케스트레이터를 사용하여 대화 흐름을 사용자 지정하는 방법

Azure OpenAI 클라이언트 설정

using System;
using System.Collections.Generic;
using System.Threading.Tasks;
using Azure.AI.OpenAI;
using Azure.Identity;
using Microsoft.Agents.AI.Workflows;
using Microsoft.Extensions.AI;
using Microsoft.Agents.AI;

// Set up the Azure OpenAI client
var endpoint = Environment.GetEnvironmentVariable("AZURE_OPENAI_ENDPOINT") ??
    throw new InvalidOperationException("AZURE_OPENAI_ENDPOINT is not set.");
var deploymentName = Environment.GetEnvironmentVariable("AZURE_OPENAI_DEPLOYMENT_NAME") ?? "gpt-4o-mini";
var client = new AzureOpenAIClient(new Uri(endpoint), new DefaultAzureCredential())
    .GetChatClient(deploymentName)
    .AsIChatClient();

경고

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

에이전트 설정

그룹 대화에서 다양한 역할에 대한 특수 에이전트를 만듭니다.

// Create a copywriter agent
ChatClientAgent writer = new(client,
    "You are a creative copywriter. Generate catchy slogans and marketing copy. Be concise and impactful.",
    "CopyWriter",
    "A creative copywriter agent");

// Create a reviewer agent
ChatClientAgent reviewer = new(client,
    "You are a marketing reviewer. Evaluate slogans for clarity, impact, and brand alignment. " +
    "Provide constructive feedback or approval.",
    "Reviewer",
    "A marketing review agent");

Round-Robin Orchestrator를 사용하여 그룹 채팅 구성

다음을 사용하여 그룹 채팅 워크플로를 구축합니다: AgentWorkflowBuilder.

// Build group chat with round-robin speaker selection
// The manager factory receives the list of agents and returns a configured manager
var workflow = AgentWorkflowBuilder
    .CreateGroupChatBuilderWith(agents =>
        new RoundRobinGroupChatManager(agents)
        {
            MaximumIterationCount = 5  // Maximum number of turns
        })
    .AddParticipants(writer, reviewer)
    .Build();

그룹 채팅 워크플로 실행

워크플로를 실행하고 반복 대화를 관찰합니다.

// Start the group chat
var messages = new List<ChatMessage> {
    new(ChatRole.User, "Create a slogan for an eco-friendly electric vehicle.")
};

StreamingRun run = await InProcessExecution.StreamAsync(workflow, messages);
await run.TrySendMessageAsync(new TurnToken(emitEvents: true));

await foreach (WorkflowEvent evt in run.WatchStreamAsync().ConfigureAwait(false))
{
    if (evt is AgentResponseUpdateEvent update)
    {
        // Process streaming agent responses
        AgentResponse response = update.AsResponse();
        foreach (ChatMessage message in response.Messages)
        {
            Console.WriteLine($"[{update.ExecutorId}]: {message.Text}");
        }
    }
    else if (evt is WorkflowOutputEvent output)
    {
        // Workflow completed
        var conversationHistory = output.As<List<ChatMessage>>();
        Console.WriteLine("\n=== Final Conversation ===");
        foreach (var message in conversationHistory)
        {
            Console.WriteLine($"{message.AuthorName}: {message.Text}");
        }
        break;
    }
}

샘플 상호 작용

[CopyWriter]: "Green Dreams, Zero Emissions" - Drive the future with style and sustainability.

[Reviewer]: The slogan is good, but "Green Dreams" might be a bit abstract. Consider something
more direct like "Pure Power, Zero Impact" to emphasize both performance and environmental benefit.

[CopyWriter]: "Pure Power, Zero Impact" - Experience electric excellence without compromise.

[Reviewer]: Excellent! This slogan is clear, impactful, and directly communicates the key benefits.
The tagline reinforces the message perfectly. Approved for use.

[CopyWriter]: Thank you! The final slogan is: "Pure Power, Zero Impact" - Experience electric
excellence without compromise.

채팅 클라이언트 설정

from agent_framework.azure import AzureOpenAIChatClient
from azure.identity import AzureCliCredential

# Initialize the Azure OpenAI chat client
chat_client = AzureOpenAIChatClient(credential=AzureCliCredential())

에이전트 설정

고유한 역할을 사용하여 특수 에이전트를 만듭니다.

from agent_framework import Agent

# Create a researcher agent
researcher = Agent(
    name="Researcher",
    description="Collects relevant background information.",
    instructions="Gather concise facts that help answer the question. Be brief and factual.",
    chat_client=chat_client,
)

# Create a writer agent
writer = Agent(
    name="Writer",
    description="Synthesizes polished answers using gathered information.",
    instructions="Compose clear, structured answers using any notes provided. Be comprehensive.",
    chat_client=chat_client,
)

단순 선택기를 사용하여 그룹 채팅 구성

사용자 지정 화자 선택 논리를 사용하여 그룹 채팅을 빌드합니다.

from agent_framework.orchestrations import GroupChatBuilder, GroupChatState

def round_robin_selector(state: GroupChatState) -> str:
    """A round-robin selector function that picks the next speaker based on the current round index."""

    participant_names = list(state.participants.keys())
    return participant_names[state.current_round % len(participant_names)]


# Build the group chat workflow
workflow = GroupChatBuilder(
    participants=[researcher, writer],
    termination_condition=lambda conversation: len(conversation) >= 4,
    selection_func=round_robin_selector,
).build()

Agent-Based Orchestrator를 사용하여 그룹 채팅 구성

또는 지능형 스피커 선택에 에이전트 기반 오케스트레이터를 사용합니다. 오케스트레이터는 도구, 컨텍스트 및 관측 가능성에 대한 전체 액세스 권한을 가진 Agent 입니다.

# Create orchestrator agent for speaker selection
orchestrator_agent = Agent(
    name="Orchestrator",
    description="Coordinates multi-agent collaboration by selecting speakers",
    instructions="""
You coordinate a team conversation to solve the user's task.

Guidelines:
- Start with Researcher to gather information
- Then have Writer synthesize the final answer
- Only finish after both have contributed meaningfully
""",
    chat_client=chat_client,
)

# Build group chat with agent-based orchestrator
workflow = GroupChatBuilder(
    participants=[researcher, writer],
    # Set a hard termination condition: stop after 4 assistant messages
    # The agent orchestrator will intelligently decide when to end before this limit but just in case
    termination_condition=lambda messages: sum(1 for msg in messages if msg.role == "assistant") >= 4,
    orchestrator_agent=orchestrator_agent,
).build()

그룹 채팅 워크플로 실행

워크플로를 실행하고 이벤트를 처리합니다.

from typing import cast
from agent_framework import AgentResponseUpdate, Role

task = "What are the key benefits of async/await in Python?"

print(f"Task: {task}\n")
print("=" * 80)

final_conversation: list[Message] = []
last_executor_id: str | None = None

# Run the workflow
async for event in workflow.run_stream(task):
    if event.type == "output" and isinstance(event.data, AgentResponseUpdate):
        # Print streaming agent updates
        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 event.type == "output":
        # Workflow completed - data is a list of Message
        final_conversation = cast(list[Message], event.data)

if final_conversation:
    print("\n\n" + "=" * 80)
    print("Final Conversation:")
    for msg in final_conversation:
        author = getattr(msg, "author_name", "Unknown")
        text = getattr(msg, "text", str(msg))
        print(f"\n[{author}]\n{text}")
        print("-" * 80)

print("\nWorkflow completed.")

샘플 상호 작용

Task: What are the key benefits of async/await in Python?

================================================================================

[Researcher]: Async/await in Python provides non-blocking I/O operations, enabling
concurrent execution without threading overhead. Key benefits include improved
performance for I/O-bound tasks, better resource utilization, and simplified
concurrent code structure using native coroutines.

[Writer]: The key benefits of async/await in Python are:

1. **Non-blocking Operations**: Allows I/O operations to run concurrently without
   blocking the main thread, significantly improving performance for network
   requests, file I/O, and database queries.

2. **Resource Efficiency**: Avoids the overhead of thread creation and context
   switching, making it more memory-efficient than traditional threading.

3. **Simplified Concurrency**: Provides a clean, synchronous-looking syntax for
   asynchronous code, making concurrent programs easier to write and maintain.

4. **Scalability**: Enables handling thousands of concurrent connections with
   minimal resource consumption, ideal for high-performance web servers and APIs.

--------------------------------------------------------------------------------

Workflow completed.

주요 개념

  • 중앙 집중식 관리자: 그룹 채팅은 관리자를 사용하여 화자 선택 및 흐름을 조정합니다.
  • AgentWorkflowBuilder.CreateGroupChatBuilderWith(): 관리자 팩터리 함수를 사용하여 워크플로를 만듭니다.
  • RoundRobinGroupChatManager: 라운드 로빈 방식으로 스피커를 대체하는 기본 제공 관리자
  • MaximumIterationCount: 종료 전 에이전트 턴의 최대 수를 제어합니다.
  • 사용자 지정 관리자: 사용자 지정 논리 확장 RoundRobinGroupChatManager 또는 구현
  • 반복적 구체화: 에이전트는 서로의 기여도를 검토하고 개선합니다.
  • 공유 컨텍스트: 모든 참가자가 전체 대화 기록을 볼 수 있습니다.
  • 유연한 오케스트레이터 전략: 생성자 매개 변수(selection_funcorchestrator_agent또는)를 통해 간단한 선택기, 에이전트 기반 오케스트레이터 또는 orchestrator사용자 지정 논리 중에서 선택합니다.
  • GroupChatBuilder: 구성 가능한 화자 선택을 사용하여 워크플로 만들기
  • GroupChatState: 선택 결정에 대한 대화 상태를 제공합니다.
  • 반복적인 공동 작업: 에이전트는 서로의 기여를 기반으로 합니다.
  • 이벤트 스트리밍: 실시간으로 데이터로 WorkflowOutputEvent 처리 AgentResponseUpdate
  • list[Message] 출력: 모든 오케스트레이션은 채팅 메시지 목록을 반환합니다.

고급: 사용자 지정 스피커 선택

사용자 지정 그룹 채팅 관리자를 만들어 사용자 지정 관리자 논리를 구현할 수 있습니다.

public class ApprovalBasedManager : RoundRobinGroupChatManager
{
    private readonly string _approverName;

    public ApprovalBasedManager(IReadOnlyList<AIAgent> agents, string approverName)
        : base(agents)
    {
        _approverName = approverName;
    }

    // Override to add custom termination logic
    protected override ValueTask<bool> ShouldTerminateAsync(
        IReadOnlyList<ChatMessage> history,
        CancellationToken cancellationToken = default)
    {
        var last = history.LastOrDefault();
        bool shouldTerminate = last?.AuthorName == _approverName &&
            last.Text?.Contains("approve", StringComparison.OrdinalIgnoreCase) == true;

        return ValueTask.FromResult(shouldTerminate);
    }
}

// Use custom manager in workflow
var workflow = AgentWorkflowBuilder
    .CreateGroupChatBuilderWith(agents =>
        new ApprovalBasedManager(agents, "Reviewer")
        {
            MaximumIterationCount = 10
        })
    .AddParticipants(writer, reviewer)
    .Build();

대화 상태에 따라 정교한 선택 논리를 구현할 수 있습니다.

def smart_selector(state: GroupChatState) -> str:
    """Select speakers based on conversation content and context."""
    conversation = state.conversation

    last_message = conversation[-1] if conversation else None

    # If no messages yet, start with Researcher
    if not last_message:
        return "Researcher"

    # Check last message content
    last_text = last_message.text.lower()

    # If researcher finished gathering info, switch to writer
    if "I have finished" in last_text and last_message.author_name == "Researcher":
        return "Writer"

    # Else continue with researcher until it indicates completion
    return "Researcher"

workflow = GroupChatBuilder(
    participants=[researcher, writer],
    selection_func=smart_selector,
).build()

중요합니다

고급 시나리오에 대한 사용자 지정 구현 BaseGroupChatOrchestrator 을 사용하는 경우 , participant_registry및 max_rounds를 비롯한 termination_condition모든 속성을 설정해야 합니다. max_rounds 및 termination_condition 빌더에서 설정한 내용은 무시됩니다.

컨텍스트 동기화

이 가이드의 시작 부분에서 설명한 것처럼 그룹 채팅의 모든 에이전트는 전체 대화 기록을 봅니다.

에이전트 프레임워크의 에이전트는 에이전트 세션(AgentSession)을 사용하여 컨텍스트를 관리합니다. 그룹 채팅 오케스트레이션에서 에이전트는 동일한 세션 인스턴스를 공유 하지 않지만 오케스트레이터는 각 턴 전에 각 에이전트의 세션이 전체 대화 기록과 동기화되도록 합니다. 이를 위해 각 에이전트가 전환된 후 오케스트레이터는 다른 모든 에이전트에 응답을 브로드캐스트하여 모든 참가자가 다음 턴에 대한 최신 컨텍스트를 갖도록 합니다.

그룹 채팅 컨텍스트 동기화

팁 (조언)

에이전트 유형에 따라 추상화 구현이 다를 수 있으므로 에이전트 는 동일한 세션 인스턴스를 AgentSession 공유하지 않습니다. 동일한 세션 인스턴스를 공유하면 각 에이전트가 컨텍스트를 처리하고 유지 관리하는 방식이 불일치할 수 있습니다.

응답을 브로드캐스트한 후 오케스트레이터는 다음 스피커를 결정하고 선택한 에이전트에 요청을 보내며, 이제 응답을 생성하기 위한 전체 대화 기록이 있습니다.

그룹 채팅을 사용하는 경우

그룹 채팅 오케스트레이션은 다음 작업에 적합합니다.

  • 반복적 구체화: 여러 차례의 검토 및 개선
  • 공동 작업 문제 해결: 상호 보완적인 전문 지식을 갖춘 에이전트가 함께 작동합니다.
  • 콘텐츠 만들기: 문서 생성을 위한 작성자-검토자 워크플로
  • 다중 관점 분석: 동일한 입력에 대한 다양한 뷰포인트 가져오기
  • 품질 보증: 자동화된 검토 및 승인 프로세스

다음과 같은 경우 대안을 고려합니다.

  • 엄격한 순차적 처리가 필요합니다(순차 오케스트레이션 사용)
  • 에이전트는 완전히 독립적으로 작동해야 합니다(동시 오케스트레이션 사용)
  • 직접 에이전트 간 핸드오프 필요(핸드오프 오케스트레이션 사용)
  • 복잡한 동적 계획이 필요합니다(Magentic 오케스트레이션 사용)

다음 단계:

마젠틱 오케스트레이션

  • Last updated on