Agent Group Chat 오케스트레이션

Group Chat 오케스트레이션

[!IMPORTANT] Agent Framework의 오케스트레이션 기능은 실험 단계(experimental)예요. 활발히 개발 중이며, preview/RC 단계로 넘어가기 전에 크게 바뀔 수 있어요.

그룹 채팅(group chat) 오케스트레이션 은 에이전트들 사이의 협업 대화를 모델링해요. 필요하면 인간 참가자도 끼울 수 있죠. 그룹 채팅 매니저가 흐름을 조율해서 "다음에 누가 응답할지", "언제 인간 입력을 요청할지"를 결정합니다. 회의·토론·협업 문제 해결 세션을 시뮬레이션할 때 특히 강력한 패턴이에요. 이 패턴을 언제 쓰고 언제 피해야 하는지는 Azure 그룹 채팅 오케스트레이션 문서를 참고하세요.

흔한 사용 사례

서로 다른 부서를 대표하는 에이전트들이 사업 제안을 논의하고, 매니저 에이전트가 대화를 중재하며 필요할 때 인간을 끌어들이는 시나리오가 전형적이에요.

배울 내용

  • 그룹 채팅을 위해 서로 다른 역할을 가진 에이전트를 정의하는 법
  • 그룹 채팅 매니저로 대화 흐름을 제어하는 법
  • 대화에 인간 참가자를 끌어들이는 법
  • 대화를 관찰하고 최종 결과를 모으는 법

에이전트 정의하기

그룹 채팅의 각 에이전트는 명확한 역할을 가져요. 아래 예시에서는 라이터(copywriter)와 리뷰어(reviewer)를 정의해 봤어요. 여기서는 ChatCompletionAgent 를 쓰지만, 어떤 에이전트 타입이든 쓸 수 있어요.

from semantic_kernel.agents import Agent, ChatCompletionAgent
from semantic_kernel.connectors.ai.open_ai import AzureChatCompletion

def get_agents() -> list[Agent]:
    writer = ChatCompletionAgent(
        name="Writer",
        description="A content writer.",
        instructions=(
            "You are an excellent content writer. You create new content and edit contents based on the feedback."
        ),
        service=AzureChatCompletion(),
    )
    reviewer = ChatCompletionAgent(
        name="Reviewer",
        description="A content reviewer.",
        instructions=(
            "You are an excellent content reviewer. You review the content and provide feedback to the writer."
        ),
        service=AzureChatCompletion(),
    )
    return [writer, reviewer]

에이전트 응답 관찰하기(선택)

대화가 진행되면서 각 에이전트의 메시지를 출력하는 콜백을 정의할 수 있어요.

from semantic_kernel.contents import ChatMessageContent

def agent_response_callback(message: ChatMessageContent) -> None:
    print(f"**{message.name}**\n{message.content}")

그룹 채팅 오케스트레이션 구성

GroupChatOrchestration 객체를 만들 때 에이전트 목록, 그룹 채팅 매니저(여기선 RoundRobinGroupChatManager), 응답 콜백을 넘겨요. 매니저가 흐름을 제어하는데, 여기서는 정해진 라운드 수만큼 돌아가며(round-robin) 차례를 바꿔가요.

from semantic_kernel.agents import GroupChatOrchestration, RoundRobinGroupChatManager

agents = get_agents()
group_chat_orchestration = GroupChatOrchestration(
    members=agents,
    manager=RoundRobinGroupChatManager(max_rounds=5),  # 홀수라서 라이터가 마지막 말을 가져감
    agent_response_callback=agent_response_callback,
)

C#에서는 RoundRobinGroupChatManager { MaximumInvocationCount = 5 } 처럼 MaximumInvocationCount 로 라운드 수를 정해요.

런타임 시작

에이전트 실행을 관리할 런타임이 필요해요. InProcessRuntime 을 만들어 오케스트레이션 호출 전에 시작합니다.

from semantic_kernel.agents.runtime import InProcessRuntime

runtime = InProcessRuntime()
runtime.start()

오케스트레이션 호출

초기 작업(예: "새 전기 SUV의 슬로건을 만들어 줘…")으로 오케스트레이션을 호출하면, 에이전트들이 번갈아 응답하며 결과를 다듬어요.

orchestration_result = await group_chat_orchestration.invoke(
    task="Create a slogan for a new electric SUV that is affordable and fun to drive.",
    runtime=runtime,
)

결과 수집

오케스트레이션이 끝날 때까지 기다렸다가 최종 출력을 받아요.

value = await orchestration_result.get()
print(f"***** Final Result *****\n{value}")

런타임 정지(선택)

처리가 끝나면 런타임을 정지해 리소스를 정리합니다.

await runtime.stop_when_idle()

샘플 출력

라이터와 리뷰어가 번갈아 슬로건을 내놓고 다듬다가, 결국 리뷰어의 승인을 받은 문구 하나가 최종 결과로 나와요. "Feel the Charge: Adventure Meets Affordability in Your New Electric SUV!" 같은 식이죠.

그룹 채팅 매니저 커스터마이즈

GroupChatManager 를 상속해 추상 메서드를 오버라이드하면 그룹 채팅 흐름을 직접 제어할 수 있어요. 결과를 어떻게 필터링할지, 다음 에이전트를 어떻게 고를지, 언제 사용자 입력을 요청하거나 채팅을 끝낼지를 결정하는 지점이에요.

from semantic_kernel.agents import GroupChatManager, BooleanResult, StringResult, MessageResult
from semantic_kernel.contents import ChatMessageContent, ChatHistory

class CustomGroupChatManager(GroupChatManager):
    async def filter_results(self, chat_history: ChatHistory) -> MessageResult:
        # 결과를 필터링하거나 요약하는 커스텀 로직
        summary = "Summary of the discussion."
        return MessageResult(result=ChatMessageContent(role="assistant", content=summary), reason="Custom summary logic.")

    async def select_next_agent(self, chat_history: ChatHistory, participant_descriptions: dict[str, str]) -> StringResult:
        # 참가자 중 랜덤하게 다음 에이전트 선택
        import random
        next_agent = random.choice(list(participant_descriptions.keys()))
        return StringResult(result=next_agent, reason="Custom selection logic.")

    async def should_request_user_input(self, chat_history: ChatHistory) -> BooleanResult:
        # 사용자 입력이 필요한지 결정하는 커스텀 로직
        return BooleanResult(result=False, reason="No user input required.")

    async def should_terminate(self, chat_history: ChatHistory) -> BooleanResult:
        # 기본 종료 로직을 먼저 확인
        base_result = await super().should_terminate(chat_history)
        if base_result.result:
            return base_result
        # 예: 10개 메시지가 넘으면 종료
        should_end = len(chat_history.messages) > 10
        return BooleanResult(result=should_end, reason="Custom termination logic.")

그리고 오케스트레이션에서 커스텀 매니저를 이렇게 사용해요.

from semantic_kernel.agents import GroupChatOrchestration
group_chat_orchestration = GroupChatOrchestration(manager=CustomGroupChatManager(max_rounds=5), ...)

그룹 채팅 매니저 메서드 호출 순서

그룹 채팅을 조율할 때 매니저의 메서드는 라운드마다 정해진 순서로 호출돼요.

  1. ShouldRequestUserInput: 다음 에이전트가 말하기 전에 사용자(인간) 입력이 필요한지 확인. 참이면 오케스트레이션이 사용자 입력을 기다리고, 그 입력을 매니저의 대화 기록에 추가해 모든 에이전트에게 보내요.
  2. ShouldTerminate: 그룹 채팅을 끝낼지 결정(예: 최대 라운드 도달 또는 커스텀 조건). 참이면 결과 필터링 단계로 넘어가요.
  3. FilterResults: 채팅이 종료되는 경우에만 호출돼, 최종 결과를 요약하거나 처리해요.
  4. SelectNextAgent: 채팅이 종료되지 않았다면, 다음으로 응답할 에이전트를 선택해요.

이 순서 덕분에 사용자 입력과 종료 조건을 대화를 진행시키기 전에 먼저 확인하고, 결과 필터링은 끝에서만 수행돼요. 각 단계의 로직은 커스텀 매니저에서 해당 메서드를 오버라이드해 바꾸면 됩니다.