그룹챗(Group Chat) 멀티에이전트 패턴
그룹챗(Group Chat) 멀티에이전트 패턴
여러 에이전트가 함께 협업해야 할 때, **그룹챗(Group Chat)**은 아주 유용한 설계 패턴이에요. 그룹챗은 여러 에이전트가 공통의 메시지 스레드를 공유하는 방식이에요. 모두 같은 토픽을 구독하고 게시하죠. 각 참여 에이전트는 특정 작업에 특화되는데, 협업 글쓰기에서는 작가(writer)·일러스트레이터(illustrator)·편집자(editor) 같은 역할이 대표적이에요. 필요할 때 에이전트를 안내해줄 인간 사용자 역할의 에이전트도 포함할 수 있어요.
그룹챗에서 참여자들은 돌아가며 메시지를 게시해요. 과정은 순차적이라 한 번에 한 에이전트만 작업하죠. 내부적으로 발언 순서는 그룹챗 매니저(Group Chat Manager) 에이전트가 관리해요. 이 에이전트는 메시지를 받으면 다음에 말할 에이전트를 선택합니다. 다음 에이전트를 선택하는 정확한 알고리즘은 애플리케이션 요구사항에 따라 달라질 수 있어요. 보통 라운드로빈(round-robin) 알고리즘이나 LLM을 쓰는 셀렉터를 사용하죠.
그룹챗은 복잡한 작업을, 명확한 역할을 가진 특화 에이전트들이 처리할 수 있는 더 작은 작업들로 동적으로 분해할 때 유용해요. 또한 각 참여자가 재귀적 그룹챗이 되는 계층 구조로 그룹챗을 중첩할 수도 있어요.
출처: 공식문서
이 예제에서는 AutoGen의 Core API로 이벤트 기반 에이전트를 사용해 그룹챗 패턴을 구현해요. 먼저 Topics and Subscriptions를 읽어 개념을 이해하고, Messages and Communication에서 pub-sub API 사용법을 배우길 권해요. 동화책 콘텐츠를 만들기 위해 그룹챗 매니저로 LLM 기반 셀렉터를 쓰는 간단한 그룹챗 예제를 보여줄게요.
이 예제는 그룹챗 메커니즘을 보여주지만, 복잡하고 커스텀 에이전트와 발언자 선택 알고리즘으로 나만의 그룹챗 시스템을 만들기 위한 출발점이에요. [AgentChat API](../../agentchat-user-guide/index.md)에는 셀렉터 그룹챗의 내장 구현이 있어요. Core API를 쓰기 싫다면 그걸 사용하면 됩니다.
메시지를 예쁘게 표시하기 위해 rich 라이브러리를 사용할게요.
# ! pip install rich
import json
import string
import uuid
from typing import List
import openai
from autogen_core import (
DefaultTopicId,
FunctionCall,
Image,
MessageContext,
RoutedAgent,
SingleThreadedAgentRuntime,
TopicId,
TypeSubscription,
message_handler,
)
from autogen_core.models import (
AssistantMessage,
ChatCompletionClient,
LLMMessage,
SystemMessage,
UserMessage,
)
from autogen_core.tools import FunctionTool
from autogen_ext.models.openai import OpenAIChatCompletionClient
from IPython.display import display # type: ignore
from pydantic import BaseModel
from rich.console import Console
from rich.markdown import Markdown
메시지 프로토콜
그룹챗 패턴의 메시지 프로토콜은 간단해요.
- 시작하려면 사용자나 외부 에이전트가 모든 참여자의 공통 토픽에
GroupChatMessage메시지를 게시해요. - 그룹챗 매니저가 다음 발언자를 선택하고 그 에이전트에게
RequestToSpeak메시지를 보냅니다. - 에이전트는
RequestToSpeak메시지를 받으면 공통 토픽에GroupChatMessage메시지를 게시해요. - 그룹챗 매니저에서 종료 조건에 도달할 때까지 이 과정이 계속되고, 그러면 매니저가
RequestToSpeak발행을 멈추면서 그룹챗이 끝나요.
아래 다이어그램은 위 2~4단계를 보여줍니다.
class GroupChatMessage(BaseModel):
body: UserMessage
class RequestToSpeak(BaseModel):
pass
기본 그룹챗 에이전트
먼저 LLM 모델만 사용해 텍스트를 생성하는 에이전트 클래스를 정의할게요. 이 클래스는 그룹챗의 모든 AI 에이전트의 베이스 클래스로 쓰여요.
class BaseGroupChatAgent(RoutedAgent):
"""A group chat participant using an LLM."""
def __init__(
self,
description: str,
group_chat_topic_type: str,
model_client: ChatCompletionClient,
system_message: str,
) -> None:
super().__init__(description=description)
self._group_chat_topic_type = group_chat_topic_type
self._model_client = model_client
self._system_message = SystemMessage(content=system_message)
self._chat_history: List[LLMMessage] = []
@message_handler
async def handle_message(self, message: GroupChatMessage, ctx: MessageContext) -> None:
self._chat_history.extend(
[
UserMessage(content=f"Transferred to {message.body.source}", source="system"),
message.body,
]
)
@message_handler
async def handle_request_to_speak(self, message: RequestToSpeak, ctx: MessageContext) -> None:
# print(f"\n{'-'*80}\n{self.id.type}:", flush=True)
Console().print(Markdown(f"### {self.id.type}: "))
self._chat_history.append(
UserMessage(content=f"Transferred to {self.id.type}, adopt the persona immediately.", source="system")
)
completion = await self._model_client.create([self._system_message] + self._chat_history)
assert isinstance(completion.content, str)
self._chat_history.append(AssistantMessage(content=completion.content, source=self.id.type))
Console().print(Markdown(completion.content))
# print(completion.content, flush=True)
await self.publish_message(
GroupChatMessage(body=UserMessage(content=completion.content, source=self.id.type)),
topic_id=DefaultTopicId(type=self._group_chat_topic_type),
)
작가와 편집자 에이전트
베이스 클래스를 사용해 서로 다른 시스템 메시지를 가진 작가와 편집자 에이전트를 정의할 수 있어요.
class WriterAgent(BaseGroupChatAgent):
def __init__(self, description: str, group_chat_topic_type: str, model_client: ChatCompletionClient) -> None:
super().__init__(
description=description,
group_chat_topic_type=group_chat_topic_type,
model_client=model_client,
system_message="You are a Writer. You produce good work.",
)
class EditorAgent(BaseGroupChatAgent):
def __init__(self, description: str, group_chat_topic_type: str, model_client: ChatCompletionClient) -> None:
super().__init__(
description=description,
group_chat_topic_type=group_chat_topic_type,
model_client=model_client,
system_message="You are an Editor. Plan and guide the task given by the user. Provide critical feedbacks to the draft and illustration produced by Writer and Illustrator. "
"Approve if the task is completed and the draft and illustration meets user's requirements.",
)
이미지 생성이 있는 일러스트레이터 에이전트
이제 제공된 설명을 바탕으로 이미지를 생성하기 위해 DALL-E 모델을 사용하는 IllustratorAgent를 정의할게요. 이미지 생성기를 FunctionTool 래퍼로 도구로 설정하고, 모델 클라이언트로 그 도구 호출을 해요.
class IllustratorAgent(BaseGroupChatAgent):
def __init__(
self,
description: str,
group_chat_topic_type: str,
model_client: ChatCompletionClient,
image_client: openai.AsyncClient,
) -> None:
super().__init__(
description=description,
group_chat_topic_type=group_chat_topic_type,
model_client=model_client,
system_message="You are an Illustrator. You use the generate_image tool to create images given user's requirement. "
"Make sure the images have consistent characters and style.",
)
self._image_client = image_client
self._image_gen_tool = FunctionTool(
self._image_gen, name="generate_image", description="Call this to generate an image. "
)
async def _image_gen(
self, character_appearence: str, style_attributes: str, worn_and_carried: str, scenario: str
) -> str:
prompt = f"Digital painting of a {character_appearence} character with {style_attributes}. Wearing {worn_and_carried}, {scenario}."
response = await self._image_client.images.generate(
prompt=prompt, model="dall-e-3", response_format="b64_json", size="1024x1024"
)
return response.data[0].b64_json # type: ignore
@message_handler
async def handle_request_to_speak(self, message: RequestToSpeak, ctx: MessageContext) -> None: # type: ignore
Console().print(Markdown(f"### {self.id.type}: "))
self._chat_history.append(
UserMessage(content=f"Transferred to {self.id.type}, adopt the persona immediately.", source="system")
)
# Ensure that the image generation tool is used.
completion = await self._model_client.create(
[self._system_message] + self._chat_history,
tools=[self._image_gen_tool],
extra_create_args={"tool_choice": "required"},
cancellation_token=ctx.cancellation_token,
)
assert isinstance(completion.content, list) and all(
isinstance(item, FunctionCall) for item in completion.content
)
images: List[str | Image] = []
for tool_call in completion.content:
arguments = json.loads(tool_call.arguments)
Console().print(arguments)
result = await self._image_gen_tool.run_json(arguments, ctx.cancellation_token)
image = Image.from_base64(self._image_gen_tool.return_value_as_string(result))
image = Image.from_pil(image.image.resize((256, 256)))
display(image.image) # type: ignore
images.append(image)
await self.publish_message(
GroupChatMessage(body=UserMessage(content=images, source=self.id.type)),
DefaultTopicId(type=self._group_chat_topic_type),
)
사용자 에이전트
모든 AI 에이전트를 정의했으니, 그룹챗에서 인간 사용자 역할을 맡을 사용자 에이전트를 정의할 수 있어요. UserAgent 구현은 콘솔 입력으로 사용자의 입력을 받아요. 실제 시나리오에서는 이걸 프론트엔드와 통신하고 프론트엔드의 응답을 구독하도록 교체할 수 있어요.
class UserAgent(RoutedAgent):
def __init__(self, description: str, group_chat_topic_type: str) -> None:
super().__init__(description=description)
self._group_chat_topic_type = group_chat_topic_type
@message_handler
async def handle_message(self, message: GroupChatMessage, ctx: MessageContext) -> None:
# When integrating with a frontend, this is where group chat message would be sent to the frontend.
pass
@message_handler
async def handle_request_to_speak(self, message: RequestToSpeak, ctx: MessageContext) -> None:
user_input = input("Enter your message, type 'APPROVE' to conclude the task: ")
Console().print(Markdown(f"### User: \n{user_input}"))
await self.publish_message(
GroupChatMessage(body=UserMessage(content=user_input, source=self.id.type)),
DefaultTopicId(type=self._group_chat_topic_type),
)
그룹챗 매니저
마지막으로, 그룹챗을 관리하고 LLM으로 다음에 말할 에이전트를 선택하는 GroupChatManager 에이전트를 정의해요. 그룹챗 매니저는 메시지에서 "APPROVED" 키워드를 찾아 편집자가 초안을 승인했는지 확인해요. 편집자가 승인하면 그룹챗 매니저는 다음 발언자 선택을 멈추고 그룹챗이 끝나죠.
그룹챗 매니저의 생성자는 참여자 토픽 타입 목록을 인자로 받아요. 다음 발언자가 작업하게 하려면 GroupChatManager 에이전트가 다음 참여자의 토픽에 RequestToSpeak 메시지를 게시합니다. 또한 이 예제에서는 이전 발언자를 추적해 그룹챗 매니저가 항상 다른 참여자를 선택하도록 해요. 이러면 그룹챗이 단일 참여자에게 지배당하지 않도록 보장됩니다.
class GroupChatManager(RoutedAgent):
def __init__(
self,
participant_topic_types: List[str],
model_client: ChatCompletionClient,
participant_descriptions: List[str],
) -> None:
super().__init__("Group chat manager")
self._participant_topic_types = participant_topic_types
self._model_client = model_client
self._chat_history: List[UserMessage] = []
self._participant_descriptions = participant_descriptions
self._previous_participant_topic_type: str | None = None
@message_handler
async def handle_message(self, message: GroupChatMessage, ctx: MessageContext) -> None:
assert isinstance(message.body, UserMessage)
self._chat_history.append(message.body)
# If the message is an approval message from the user, stop the chat.
if message.body.source == "User":
assert isinstance(message.body.content, str)
if message.body.content.lower().strip(string.punctuation).endswith("approve"):
return
# Format message history.
messages: List[str] = []
for msg in self._chat_history:
if isinstance(msg.content, str):
messages.append(f"{msg.source}: {msg.content}")
elif isinstance(msg.content, list):
line: List[str] = []
for item in msg.content:
if isinstance(item, str):
line.append(item)
else:
line.append("[Image]")
messages.append(f"{msg.source}: {', '.join(line)}")
history = "\n".join(messages)
# Format roles.
roles = "\n".join(
[
f"{topic_type}: {description}".strip()
for topic_type, description in zip(
self._participant_topic_types, self._participant_descriptions, strict=True
)
if topic_type != self._previous_participant_topic_type
]
)
selector_prompt = """You are in a role play game. The following roles are available:
{roles}.
Read the following conversation. Then select the next role from {participants} to play. Only return the role.
{history}
Read the above conversation. Then select the next role from {participants} to play. Only return the role.
"""
system_message = SystemMessage(
content=selector_prompt.format(
roles=roles,
history=history,
participants=str(
[
topic_type
for topic_type in self._participant_topic_types
if topic_type != self._previous_participant_topic_type
]
),
)
)
completion = await self._model_client.create([system_message], cancellation_token=ctx.cancellation_token)
assert isinstance(completion.content, str)
selected_topic_type: str
for topic_type in self._participant_topic_types:
if topic_type.lower() in completion.content.lower():
selected_topic_type = topic_type
self._previous_participant_topic_type = selected_topic_type
await self.publish_message(RequestToSpeak(), DefaultTopicId(type=selected_topic_type))
return
raise ValueError(f"Invalid role selected: {completion.content}")
그룹챗 만들기
그룹챗을 설정하려면 SingleThreadedAgentRuntime을 만들고 에이전트들의 팩토리와 구독을 등록해요. 각 참여자 에이전트는 RequestToSpeak 메시지를 받기 위해 그룹챗 토픽과 자기 자신의 토픽 둘 다 구독하고, 그룹챗 매니저 에이전트는 그룹챗 토픽만 구독합니다.
runtime = SingleThreadedAgentRuntime()
editor_topic_type = "Editor"
writer_topic_type = "Writer"
illustrator_topic_type = "Illustrator"
user_topic_type = "User"
group_chat_topic_type = "group_chat"
editor_description = "Editor for planning and reviewing the content."
writer_description = "Writer for creating any text content."
user_description = "User for providing final approval."
illustrator_description = "An illustrator for creating images."
model_client = OpenAIChatCompletionClient(
model="gpt-4o-2024-08-06",
# api_key="YOUR_API_KEY",
)
editor_agent_type = await EditorAgent.register(
runtime,
editor_topic_type, # Using topic type as the agent type.
lambda: EditorAgent(
description=editor_description,
group_chat_topic_type=group_chat_topic_type,
model_client=model_client,
),
)
await runtime.add_subscription(TypeSubscription(topic_type=editor_topic_type, agent_type=editor_agent_type.type))
await runtime.add_subscription(TypeSubscription(topic_type=group_chat_topic_type, agent_type=editor_agent_type.type))
writer_agent_type = await WriterAgent.register(
runtime,
writer_topic_type, # Using topic type as the agent type.
lambda: WriterAgent(
description=writer_description,
group_chat_topic_type=group_chat_topic_type,
model_client=model_client,
),
)
await runtime.add_subscription(TypeSubscription(topic_type=writer_topic_type, agent_type=writer_agent_type.type))
await runtime.add_subscription(TypeSubscription(topic_type=group_chat_topic_type, agent_type=writer_agent_type.type))
illustrator_agent_type = await IllustratorAgent.register(
runtime,
illustrator_topic_type,
lambda: IllustratorAgent(
description=illustrator_description,
group_chat_topic_type=group_chat_topic_type,
model_client=model_client,
image_client=openai.AsyncClient(
# api_key="YOUR_API_KEY",
),
),
)
await runtime.add_subscription(
TypeSubscription(topic_type=illustrator_topic_type, agent_type=illustrator_agent_type.type)
)
await runtime.add_subscription(
TypeSubscription(topic_type=group_chat_topic_type, agent_type=illustrator_agent_type.type)
)
user_agent_type = await UserAgent.register(
runtime,
user_topic_type,
lambda: UserAgent(description=user_description, group_chat_topic_type=group_chat_topic_type),
)
await runtime.add_subscription(TypeSubscription(topic_type=user_topic_type, agent_type=user_agent_type.type))
await runtime.add_subscription(TypeSubscription(topic_type=group_chat_topic_type, agent_type=user_agent_type.type))
group_chat_manager_type = await GroupChatManager.register(
runtime,
"group_chat_manager",
lambda: GroupChatManager(
participant_topic_types=[writer_topic_type, illustrator_topic_type, editor_topic_type, user_topic_type],
model_client=model_client,
participant_descriptions=[writer_description, illustrator_description, editor_description, user_description],
),
)
await runtime.add_subscription(
TypeSubscription(topic_type=group_chat_topic_type, agent_type=group_chat_manager_type.type)
)
그룹챗 실행하기
런타임을 시작하고 작업을 위한 GroupChatMessage를 게시해 그룹챗을 시작해요.
runtime.start()
session_id = str(uuid.uuid4())
await runtime.publish_message(
GroupChatMessage(
body=UserMessage(
content="Please write a short story about the gingerbread man with up to 3 photo-realistic illustrations.",
source="User",
)
),
TopicId(type=group_chat_topic_type, source=session_id),
)
await runtime.stop_when_idle()
await model_client.close()
출력에서 작가, 일러스트레이터, 편집자 에이전트가 돌아가며 말하며 협업해 그림책을 만들고, 마지막으로 사용자의 최종 승인을 요청하는 걸 볼 수 있어요.
다음 단계
이 예제는 그룹챗 패턴의 단순한 구현을 보여주는 것으로, 실제 애플리케이션에서 쓰라고 만든 건 아니에요. 발언자 선택 알고리즘을 개선할 수 있어요. 예를 들어 단순한 규칙으로 충분하고 더 신뢰할 수 있는 경우엔 LLM을 피할 수 있어요. 작가 다음에는 항상 편집자가 말하는 규칙을 쓸 수 있죠.
AgentChat API는 셀렉터 그룹챗을 위한 고수준 API를 제공해요. 더 많은 기능이 있지만 대부분 이 구현과 설계를 공유합니다.
더 알아보기 (Learn more)
- 더 간편하게 쓰는 내장 셀렉터 그룹챗은 Selector Group Chat을 참고하세요.
- 그룹챗을 순차 워크플로우로 바꾸고 싶다면 순차 워크플로우를 확인하세요.
- 토픽과 구독의 기본은 토픽과 구독 문서를 보세요.