Selector Group Chat

Selector Group Chat (모델 기반 발언자 선택)

SelectorGroupChat은 참가자들이 모든 멤버에게 메시지를 번갈아 브로드캐스트하는 팀이에요. 생성 모델(예: LLM)이 공유 컨텍스트를 바탕으로 다음 발언자를 선택해서, 동적이고 컨텍스트에 민감한 협업이 가능해요.

주요 특징은 다음과 같아요.

모델 기반 발언자 선택
참가자 역할·설명(configurable participant roles and descriptions)
같은 발언자가 연속으로 말하는 것 방지(선택적)
커스터마이즈 가능한 선택 프롬프트
기본 모델 기반 선택을 덮어쓰는 커스텀 선택 함수
모델로 선택 후보 에이전트 집합을 좁히는 커스텀 후보 함수

SelectorGroupChat은 고수준 API예요. 더 많은 제어·커스터마이징이 필요하면 Core API 문서의 Group Chat 패턴을 보고 직접 그룹 채팅 로직을 구현하세요.

어떻게 동작할까요?

SelectorGroupChat은 RoundRobinGroupChat과 비슷한 그룹 채팅이지만 모델 기반 다음 발언자 선택 메커니즘이 있어요. team.run()이나 team.run_stream()으로 팀이 작업을 받으면 다음 단계가 실행돼요.

팀이 현재 대화 컨텍스트(대화 기록·참가자의 name·description 포함)를 분석해 모델로 다음 발언자를 정해요. 기본적으로 팀은 동일 발언자를 연속으로 고르지 않아요(유일한 에이전트일 때만 제외). allow_repeated_speaker=True로 바꿀 수 있고, 커스텀 선택 함수로 모델을 덮어쓸 수도 있어요.
팀이 선택된 발언자 에이전트에게 응답을 요청하고, 그 응답을 다른 모든 참가자에게 브로드캐스트해요.
종료 조건을 확인해 대화를 끝낼지 결정하고, 안 끝났으면 1단계부터 반복해요.
대화가 끝나면 팀이 이 작업의 대화 기록이 담긴 TaskResult를 반환해요.

팀이 작업을 끝내면 대화 컨텍스트는 팀과 모든 참가자 안에 유지돼서, 다음 작업이 이전 대화 컨텍스트에서 이어질 수 있어요. team.reset()으로 컨텍스트를 비울 수 있어요.

이번 장에서 웹 검색·데이터 분석 작업의 간단한 예시로 SelectorGroupChat을 사용해 볼게요.

예시: 웹 검색/분석 (Web Search/Analysis)

출처: 공식 문서 - Selector Group Chat

from typing import List, Sequence

from autogen_agentchat.agents import AssistantAgent, UserProxyAgent
from autogen_agentchat.conditions import MaxMessageTermination, TextMentionTermination
from autogen_agentchat.messages import BaseAgentEvent, BaseChatMessage
from autogen_agentchat.teams import SelectorGroupChat
from autogen_agentchat.ui import Console
from autogen_ext.models.openai import OpenAIChatCompletionClient

에이전트 (Agents)

이 시스템은 세 개의 전문 에이전트를 사용해요.

Planning Agent: 복잡한 작업을 작은 하위 작업으로 쪼개는 전략적 조정자.
Web Search Agent: search_web_tool과 연결되는 정보 검색 전문가.
Data Analyst Agent: percentage_change_tool을 장착한 계산 전문 에이전트.

search_web_tool과 percentage_change_tool은 에이전트가 작업을 수행하는 데 쓰는 외부 도구예요.

# Note: This example uses mock tools instead of real APIs for demonstration purposes
def search_web_tool(query: str) -> str:
    if "2006-2007" in query:
        return """Here are the total points scored by Miami Heat players in the 2006-2007 season:
        Udonis Haslem: 844 points
        Dwayne Wade: 1397 points
        James Posey: 550 points
        ...
        """
    elif "2007-2008" in query:
        return "The number of total rebounds for Dwayne Wade in the Miami Heat season 2007-2008 is 214."
    elif "2008-2009" in query:
        return "The number of total rebounds for Dwayne Wade in the Miami Heat season 2008-2009 is 398."
    return "No data found."


def percentage_change_tool(start: float, end: float) -> float:
    return ((end - start) / start) * 100

AssistantAgent 클래스로 전문 에이전트를 만들어 볼게요. 에이전트의 name과 description 속성은 모델이 다음 발언자를 정할 때 쓰이니, 의미 있는 이름과 설명을 주는 게 좋아요.

model_client = OpenAIChatCompletionClient(model="gpt-4o")

planning_agent = AssistantAgent(
    "PlanningAgent",
    description="An agent for planning tasks, this agent should be the first to engage when given a new task.",
    model_client=model_client,
    system_message="""
    You are a planning agent.
    Your job is to break down complex tasks into smaller, manageable subtasks.
    Your team members are:
        WebSearchAgent: Searches for information
        DataAnalystAgent: Performs calculations

    You only plan and delegate tasks - you do not execute them yourself.

    When assigning tasks, use this format:
    1. <agent> : <task>

    After all tasks are complete, summarize the findings and end with "TERMINATE".
    """,
)

web_search_agent = AssistantAgent(
    "WebSearchAgent",
    description="An agent for searching information on the web.",
    tools=[search_web_tool],
    model_client=model_client,
    system_message="""
    You are a web search agent.
    Your only tool is search_tool - use it to find information.
    You make only one search call at a time.
    Once you have the results, you never do calculations based on them.
    """,
)

data_analyst_agent = AssistantAgent(
    "DataAnalystAgent",
    description="An agent for performing calculations.",
    model_client=model_client,
    tools=[percentage_change_tool],
    system_message="""
    You are a data analyst.
    Given the tasks you have been assigned, you should analyze the data and provide results using the tools provided.
    If you have not seen the data, ask for it.
    """,
)

기본적으로 AssistantAgent는 도구 출력을 그대로 응답으로 돌려줘요. 도구가 자연어로 잘 형성된 문자열을 반환하지 않으면, 에이전트를 만들 때 reflect_on_tool_use=True로 설정해 반성(reflection) 단계를 추가할 수 있어요. 그러면 에이전트가 도구 출력을 반성하고 자연어 응답을 제공하게 돼요.

워크플로

SelectorGroupChat이 작업을 받아, 에이전트 설명을 바탕으로 초기 작업을 처리할 가장 적합한 에이전트를 골라요(보통 Planning Agent).
Planning Agent가 작업을 분석해 하위 작업으로 쪼개고, <agent> : <task> 형식으로 각각 가장 적합한 에이전트에게 배정해요.
대화 컨텍스트와 에이전트 설명을 바탕으로 매니저가 하위 작업을 처리할 다음 에이전트를 동적으로 선택해요.
Web Search Agent가 한 번에 하나씩 검색을 수행하고 결과를 공유 대화 기록에 저장해요.
Data Analyst가 선택되면 사용 가능한 계산 도구로 수집된 정보를 처리해요.
다음 중 하나가 될 때까지 에이전트가 동적으로 선택되며 워크플로가 이어져요.
- Planning Agent가 모든 하위 작업이 완료됐다고 판단해 "TERMINATE"를 보냄
- 다른 종료 조건이 충족(예: 최대 메시지 수)

에이전트를 정의할 때 설명(description)을 꼭 알차게 넣으세요. 이 설명이 다음 에이전트를 고르는 데 쓰이니까요.

종료 조건 (Termination Conditions)

두 가지 종료 조건을 써 볼게요. Planning Agent가 "TERMINATE"를 보내면 대화를 끝내는 TextMentionTermination과, 무한 루프를 막으려고 대화를 25개 메시지로 제한하는 MaxMessageTermination이에요.

text_mention_termination = TextMentionTermination("TERMINATE")
max_messages_termination = MaxMessageTermination(max_messages=25)
termination = text_mention_termination | max_messages_termination

선택 프롬프트 (Selector Prompt)

SelectorGroupChat은 모델로 대화 컨텍스트를 기반으로 다음 발언자를 선택해요. 워크플로에 맞게 커스텀 선택 프롬프트를 쓸 수 있어요.

selector_prompt = """Select an agent to perform task.

{roles}

Current conversation context:
{history}

Read the above conversation, then select an agent from {participants} to perform the next task.
Make sure the planner agent has assigned tasks before other agents start working.
Only select one agent.
"""

선택 프롬프트에서 사용 가능한 문자열 변수는 다음과 같아요.

{participants}: 선택 후보 이름. 형식은 ["<name1>", "<name2>", ...].
{roles}: 후보 에이전트의 이름·설명을 줄바꿈으로 나열. 각 줄 형식은 "<name> : <description>".
{history}: 이름과 메시지 내용을 이중 줄바꿈으로 나눈 대화 기록. 각 메시지 형식은 "<name> : <message content>".

선택 프롬프트에 모델을 과부하시키는 지시를 너무 많이 넣지 마세요. "너무 많다"의 기준은 사용 중인 모델 능력에 달려 있어요. GPT-4o급 모델이면 각 발언자가 선택돼야 할 조건을 넣은 선택 프롬프트를 쓸 수 있어요. Phi-4 같은 작은 모델이면 이 예시처럼 가능한 한 단순하게 유지하세요. 일반적으로 각 에이전트에 조건을 여러 개 쓰고 있다면, 커스텀 선택 함수를 쓰거나 작업을 더 작은 순차 작업으로 쪼개 별개 에이전트·팀에 맡기는 걸 고려해야 한다는 신호예요.

팀 실행 (Running the Team)

에이전트·종료 조건·커스텀 선택 프롬프트로 팀을 만들어 볼게요.

team = SelectorGroupChat(
    [planning_agent, web_search_agent, data_analyst_agent],
    model_client=model_client,
    termination_condition=termination,
    selector_prompt=selector_prompt,
    allow_repeated_speaker=True,  # Allow an agent to speak multiple turns in a row.
)

NBA 선수에 대한 정보를 찾는 작업으로 팀을 실행해요.

task = "Who was the Miami Heat player with the highest points in the 2006-2007 season, and what was the percentage change in his total rebounds between the 2007-2008 and 2008-2009 seasons?"

# Use asyncio.run(...) if you are running this in a script.
await Console(team.run_stream(task=task))

Web Search Agent가 필요한 검색을 수행하고 Data Analyst Agent가 필요한 계산을 끝내자, 2006-2007 시즌 최다 득점자는 Dwayne Wade이고 2007-2008과 2008-2009 시즌 사이 리바운드 변화율은 85.98%라는 걸 확인할 수 있어요.

커스텀 선택 함수 (Custom Selector Function)

선택 과정을 더 잘 제어하고 싶을 때가 많아요. 그럴 땐 selector_func 인자에 커스텀 선택 함수를 넣어 기본 모델 기반 선택을 덮어쓸 수 있어요. 더 복잡한 선택 로직과 상태 기반 전환을 구현할 수 있죠. 예를 들어, 전문 에이전트가 말한 직후엔 항상 Planning Agent가 진행 상황을 확인하려 해요.

커스텀 선택 함수에서 None을 반환하면 기본 모델 기반 선택을 사용해요.

커스텀 선택 함수는 SelectorGroupChat 팀에 .dump_component()를 호출해도 직렬화되지 않아요. 커스텀 선택 함수가 있는 팀 설정을 직렬화해야 한다면 커스텀 워크플로·직렬화 로직을 구현하는 걸 고려하세요.

def selector_func(messages: Sequence[BaseAgentEvent | BaseChatMessage]) -> str | None:
    if messages[-1].source != planning_agent.name:
        return planning_agent.name
    return None


# Reset the previous team and run the chat again with the selector function.
await team.reset()
team = SelectorGroupChat(
    [planning_agent, web_search_agent, data_analyst_agent],
    model_client=model_client,
    termination_condition=termination,
    selector_prompt=selector_prompt,
    allow_repeated_speaker=True,
    selector_func=selector_func,
)

await Console(team.run_stream(task=task))

대화 로그에서 Planning Agent가 전문 에이전트 직후에 항상 말하는 걸 확인할 수 있어요.

각 참가 에이전트는 매 차례 한 단계(도구 실행, 응답 생성 등)만 수행해요. AssistantAgent가 필요한 도구를 모두 실행하고 더 이상 ToolCallSummaryMessage를 반환하지 않을 때까지 반복하게 하려면, 마지막 메시지를 확인해 그게 ToolCallSummaryMessage면 그 에이전트를 반환하면 돼요.

커스텀 후보 함수 (Custom Candidate Function)

또 하나 가능한 요구는 필터링된 에이전트 목록에서 다음 발언자를 자동으로 고르는 거예요. candidate_func 파라미터에 커스텀 후보 함수를 넣으면, 매 그룹 채팅 차례의 발언자 선택 후보 목록을 좁힐 수 있어요. 특정 에이전트 뒤에는 발언자 선택을 특정 에이전트 집합으로 제한할 수 있죠.

candidate_func는 selector_func가 설정되지 않았을 때만 유효해요. 커스텀 후보 함수에서 None이나 빈 리스트 []를 반환하면 ValueError가 나요.

def candidate_func(messages: Sequence[BaseAgentEvent | BaseChatMessage]) -> List[str]:
    # keep planning_agent first one to plan out the tasks
    if messages[-1].source == "user":
        return [planning_agent.name]

    # if previous agent is planning_agent and if it explicitely asks for web_search_agent
    # or data_analyst_agent or both (in-case of re-planning or re-assignment of tasks)
    # then return those specific agents
    last_message = messages[-1]
    if last_message.source == planning_agent.name:
        participants = []
        if web_search_agent.name in last_message.to_text():
            participants.append(web_search_agent.name)
        if data_analyst_agent.name in last_message.to_text():
            participants.append(data_analyst_agent.name)
        if participants:
            return participants  # SelectorGroupChat will select from the remaining two agents.

    # we can assume that the task is finished once the web_search_agent
    # and data_analyst_agent have took their turns, thus we send
    # in planning_agent to terminate the chat
    previous_set_of_agents = set(message.source for message in messages)
    if web_search_agent.name in previous_set_of_agents and data_analyst_agent.name in previous_set_of_agents:
        return [planning_agent.name]

    # if no-conditions are met then return all the agents
    return [planning_agent.name, web_search_agent.name, data_analyst_agent.name]


# Reset the previous team and run the chat again with the selector function.
await team.reset()
team = SelectorGroupChat(
    [planning_agent, web_search_agent, data_analyst_agent],
    model_client=model_client,
    termination_condition=termination,
    candidate_func=candidate_func,
)

await Console(team.run_stream(task=task))

대화 로그를 보면 Web Search Agent와 Data Analyst Agent가 차례를 마친 뒤 Planning Agent가 대화로 돌아와 작업이 예상대로 끝나지 않았음을 확인하고, 리바운드 값을 얻으려고 WebSearchAgent를 다시 부르고 변화율을 얻으려고 DataAnalysetAgent를 부르는 걸 볼 수 있어요.

사용자 피드백 (User Feedback)

UserProxyAgent를 팀에 추가하면 실행 중 사용자 피드백을 받을 수 있어요. UserProxyAgent에 대해 자세히 알아보려면 Human-in-the-Loop 문서를 참고하세요.

웹 검색 예시에 UserProxyAgent를 쓰려면 팀에 그냥 추가하고, 선택 함수를 바꿔 planning agent가 말한 뒤 항상 사용자 피드백을 확인하게 하면 돼요. 사용자가 "APPROVE"라고 응답하면 대화가 이어지고, 아니면 planning agent가 다시 시도해 사용자가 승인할 때까지 반복해요.

user_proxy_agent = UserProxyAgent("UserProxyAgent", description="A proxy for the user to approve or disapprove tasks.")


def selector_func_with_user_proxy(messages: Sequence[BaseAgentEvent | BaseChatMessage]) -> str | None:
    if messages[-1].source != planning_agent.name and messages[-1].source != user_proxy_agent.name:
        # Planning agent should be the first to engage when given a new task, or check progress.
        return planning_agent.name
    if messages[-1].source == planning_agent.name:
        if messages[-2].source == user_proxy_agent.name and "APPROVE" in messages[-1].content.upper():  # type: ignore
            # User has approved the plan, proceed to the next agent.
            return None
        # Use the user proxy agent to get the user's approval to proceed.
        return user_proxy_agent.name
    if messages[-1].source == user_proxy_agent.name:
        # If the user does not approve, return to the planning agent.
        if "APPROVE" not in messages[-1].content.upper():  # type: ignore
            return planning_agent.name
    return None


# Reset the previous agents and run the chat again with the user proxy agent and selector function.
await team.reset()
team = SelectorGroupChat(
    [planning_agent, web_search_agent, data_analyst_agent, user_proxy_agent],
    model_client=model_client,
    termination_condition=termination,
    selector_prompt=selector_prompt,
    selector_func=selector_func_with_user_proxy,
    allow_repeated_speaker=True,
)

await Console(team.run_stream(task=task))

이제 사용자 피드백이 대화 흐름에 반영되고, 사용자가 planning agent의 결정을 승인하거나 거부할 수 있어요.

추론 모델 사용 (Using Reasoning Models)

지금까지 예시에선 gpt-4o 모델을 썼어요. gpt-4o·gemini-1.5-flash 같은 모델은 지시를 잘 따르므로, 팀의 선택 프롬프트와 각 에이전트의 시스템 메시지에 비교적 상세한 지시를 넣을 수 있어요.

하지만 o3-mini 같은 추론(reasoning) 모델을 쓰면 선택 프롬프트와 시스템 메시지를 최대한 단순하고 핵심적으로 유지해야 해요. 추론 모델은 주어진 컨텍스트로 스스로 지시를 만들어 내는 데 능숙하기 때문이에요. 또 추론 모델을 쓰는 SelectorGroupChat은 스스로 작업을 쪼갤 수 있으니 planning agent가 필요 없어져요.

아래 예시는 에이전트와 팀에 o3-mini를 쓰고 planning agent 없이, 선택 프롬프트와 시스템 메시지를 최대한 단순하게 유지해요.

model_client = OpenAIChatCompletionClient(model="o3-mini")

web_search_agent = AssistantAgent(
    "WebSearchAgent",
    description="An agent for searching information on the web.",
    tools=[search_web_tool],
    model_client=model_client,
    system_message="""Use web search tool to find information.""",
)

data_analyst_agent = AssistantAgent(
    "DataAnalystAgent",
    description="An agent for performing calculations.",
    model_client=model_client,
    tools=[percentage_change_tool],
    system_message="""Use tool to perform calculation. If you have not seen the data, ask for it.""",
)

user_proxy_agent = UserProxyAgent(
    "UserProxyAgent",
    description="A user to approve or disapprove tasks.",
)

selector_prompt = """Select an agent to perform task.

{roles}

Current conversation context:
{history}

Read the above conversation, then select an agent from {participants} to perform the next task.
When the task is complete, let the user approve or disapprove the task.
"""

team = SelectorGroupChat(
    [web_search_agent, data_analyst_agent, user_proxy_agent],
    model_client=model_client,
    termination_condition=termination,  # Use the same termination condition as before.
    selector_prompt=selector_prompt,
    allow_repeated_speaker=True,
)

await Console(team.run_stream(task=task))

추론 모델 프롬프팅에 대한 더 자세한 안내는 Azure AI Services Blog의 Prompt Engineering for OpenAI's O1 and O3-mini Reasoning Models 문서를 참고하세요.

Selector Group Chat