AutoGen 멀티 에이전트 토론

AutoGen 멀티 에이전트 토론 (Multi-Agent Debate)

여러 에이전트가 답을 내놓고 서로 비판하며 다듬어 나가면, 혼자 풀 때보다 더 정확한 답에 도달할 수 있어요. Multi-Agent Debate은 각 턴에서 에이전트들이 서로 응답을 교환하고, 다른 에이전트의 응답을 바탕으로 자신의 응답을 다듬는 다중 턴 상호작용을 시뮬레이션하는 멀티 에이전트 디자인 패턴이에요.

이 예제는 GSM8K 벤치마크의 수학 문제를 푸는 멀티 에이전트 토론 패턴의 구현을 보여줍니다.

이 패턴에는 두 종류의 에이전트가 있어요: 솔버(solver) 에이전트와 어그리게이터(aggregator) 에이전트. 솔버 에이전트들은 Improving Multi-Agent Debate with Sparse Communication Topology에 설명된 기법을 따라 희소(sparse) 방식으로 연결됩니다. 솔버 에이전트는 수학 문제를 풀고 서로 응답을 교환하는 역할을 해요. 어그리게이터 에이전트는 수학 문제를 솔버 에이전트들에게 배포하고, 최종 응답을 기다려 집계해 최종 답을 얻는 역할을 합니다.

패턴은 다음과 같이 동작합니다:

사용자가 어그리게이터 에이전트에게 수학 문제를 보냅니다.
어그리게이터 에이전트는 문제를 솔버 에이전트들에게 배포합니다.
각 솔버 에이전트는 문제를 처리하고 이웃에게 응답을 게시합니다.
각 솔버 에이전트는 이웃의 응답을 사용해 자신의 응답을 다듬고 새 응답을 게시합니다.
고정된 라운드 수만큼 4단계를 반복합니다. 마지막 라운드에서 각 솔버 에이전트는 최종 응답을 게시합니다.
어그리게이터 에이전트는 다수결(majority voting)로 최종 답을 결정해 사용자에게 반환합니다.

출처: AutoGen 공식 문서 - Multi-Agent Debate

Message Protocol

먼저 에이전트들이 사용하는 메시지를 정의해요. IntermediateSolverResponse는 각 라운드에서 솔버 에이전트들이 교환하는 메시지이고, FinalSolverResponse는 마지막 라운드에서 솔버 에이전트들이 게시하는 메시지예요.

@dataclass
class Question:
    content: str


@dataclass
class Answer:
    content: str


@dataclass
class SolverRequest:
    content: str
    question: str


@dataclass
class IntermediateSolverResponse:
    content: str
    round: int


@dataclass
class FinalSolverResponse:
    content: str

Solver Agent

솔버 에이전트는 수학 문제를 풀고 다른 솔버 에이전트들과 응답을 교환하는 역할을 해요. SolverRequest를 받으면 솔버 에이전트는 LLM을 사용해 답을 생성합니다. 그런 다음 라운드 번호에 따라 IntermediateSolverResponse나 FinalSolverResponse를 게시해요.

솔버 에이전트에는 토픽 타입(topic type)이 주어지는데, 이는 에이전트가 중간 응답을 게시해야 하는 토픽을 나타냅니다. 이 토픽은 이웃들이 구독해서 이 에이전트의 응답을 받습니다 — 이 방법은 나중에 보여줄 거예요.

우리는 default_subscription을 사용해 솔버 에이전트가 기본 토픽을 구독하게 하고, 이 기본 토픽을 어그리게이터 에이전트가 솔버 에이전트의 최종 응답을 모으는 데 사용해요.

@default_subscription
class MathSolver(RoutedAgent):
    def __init__(self, model_client: ChatCompletionClient, topic_type: str, num_neighbors: int, max_round: int) -> None:
        super().__init__("Math Solver")
        self._model_client = model_client
        self._topic_type = topic_type
        self._num_neighbors = num_neighbors
        self._max_round = max_round

    @message_handler
    async def handle_request(self, message: SolverRequest, ctx: MessageContext) -> None:
        # 문제를 풀기 위해 LLM을 사용한다.
        response = await self._model_client.create(
            [SystemMessage(content="You are a math solver. Solve the math problem and return the answer."),
             UserMessage(content=message.content, source="user")],
            cancellation_token=ctx.cancellation_token,
        )
        answer = response.content[0].content

        # 라운드 번호에 따라 중간 응답 또는 최종 응답을 게시한다.
        if message.round < self._max_round:
            await self.publish_message(
                IntermediateSolverResponse(content=answer, round=message.round),
                topic_id=TopicId(self._topic_type, self.id.key),
            )
        else:
            await self.publish_message(
                FinalSolverResponse(content=answer),
                topic_id=DefaultTopicId(),
            )

Aggregator Agent

어그리게이터 에이전트는 사용자 질문을 처리하고 수학 문제를 솔버 에이전트들에게 배포하는 역할을 해요.

어그리게이터는 default_subscription으로 기본 토픽을 구독합니다. 기본 토픽은 사용자 질문을 받고, 솔버 에이전트의 최종 응답을 받으며, 사용자에게 최종 답을 게시하는 데 사용돼요.

멀티 에이전트 토론을 하위 컴포넌트로 격리하려는 더 복잡한 애플리케이션에서는 type_subscription을 사용해 어그리게이터-솔버 통신에 특정 토픽 타입을 설정하고, 솔버와 어그리게이터 모두 그 토픽 타입에 게시·구독하게 해야 합니다.

@default_subscription
class MathAggregator(RoutedAgent):
    def __init__(self, num_solvers: int) -> None:
        super().__init__("Math Aggregator")
        self._num_solvers = num_solvers
        self._solver_final_responses: Dict[AgentId, FinalSolverResponse] = {}

    @message_handler
    async def handle_question(self, message: Question, ctx: MessageContext) -> None:
        # 문제를 솔버에게 배포한다.
        for solver_type in ["MathSolverA", "MathSolverB", "MathSolverC", "MathSolverD"]:
            await self.send_message(
                SolverRequest(content=message.content, question=message.content, round=1),
                recipient=AgentId(solver_type, self.id.key),
            )

    @message_handler
    async def handle_final_response(self, message: FinalSolverResponse, ctx: MessageContext) -> None:
        self._solver_final_responses[ctx.sender] = message
        if len(self._solver_final_responses) == self._num_solvers:
            # 다수결 투표로 최종 답을 결정한다.
            answers = [r.content for r in self._solver_final_responses.values()]
            final = max(set(answers), key=answers.count)
            await self.publish_message(Answer(content=final), topic_id=DefaultTopicId())

Setting Up a Debate

4개의 솔버 에이전트와 1개의 어그리게이터 에이전트로 멀티 에이전트 토론을 설정해 봅시다. 솔버 에이전트들은 아래 그림처럼 희소하게 연결됩니다:

A --- B
|     |
|     |
D --- C

각 솔버 에이전트는 다른 두 솔버 에이전트에 연결됩니다. 예를 들어 에이전트 A는 에이전트 B와 C에 연결돼요.

먼저 런타임을 만들고 에이전트 타입을 등록합니다.

runtime = SingleThreadedAgentRuntime()

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

await MathSolver.register(
    runtime,
    "MathSolverA",
    lambda: MathSolver(
        model_client=model_client,
        topic_type="MathSolverA",
        num_neighbors=2,
        max_round=3,
    ),
)
# MathSolverB, MathSolverC, MathSolverD 도 같은 방식으로 등록한다.
await MathAggregator.register(runtime, "MathAggregator", lambda: MathAggregator(num_solvers=4))

이제 각 솔버 에이전트의 게시 토픽 타입을 이웃 에이전트 타입에 매핑하는 TypeSubscription으로 솔버 에이전트 토폴로지를 만듭니다.

# MathSolverA가 게시하는 토픽에 대한 구독.
await runtime.add_subscription(TypeSubscription("MathSolverA", "MathSolverD"))
await runtime.add_subscription(TypeSubscription("MathSolverA", "MathSolverB"))
# MathSolverB, MathSolverC, MathSolverD 에 대해서도 같은 방식으로 구성한다.

Solving Math Problems

이제 토론을 실행해 수학 문제를 풀어 봅시다. 기본 토픽에 SolverRequest를 게시하면 어그리게이터 에이전트가 토론을 시작합니다.

question = "Natalia sold clips to 48 of her friends in April, and then she sold half as many clips in May. How many clips did Natalia sell altogether in April and May?"
runtime.start()
await runtime.publish_message(
    Question(content=question),
    topic_id=DefaultTopicId(),
)
await runtime.stop_when_idle()

결과는 다수결 투표로 결정된 최종 답을 받을 수 있어요.

AutoGen 멀티 에이전트 토론