퀵스타트 — AutoGen Core 첫걸음

AutoGen의 Core API를 쓸 준비가 되셨나요? 본격적으로 개념을 파고들기 전에, 두 에이전트가 10부터 1까지 숫자를 세어 내려가는 가장 단순한 예제부터 시작해 볼게요. 이 예제 하나에 AutoGen이 "어떻게 여러 에이전트를 돌리는지"의 뼈대가 거의 다 들어 있어요.

출처: Quick Start — AutoGen 공식 문서

먼저 에이전트 클래스와, 메시지를 처리하는 각자의 로직을 정의합니다. 두 에이전트 클래스를 만들 건데요, Modifier는 주어진 숫자를 변형하고 Checker는 그 값이 특정 조건을 만족하는지 확인해요. 그리고 에이전트 사이를 오가는 메시지를 담을 Message 데이터 클래스도 함께 정의해요.

from dataclasses import dataclass
from typing import Callable

from autogen_core import DefaultTopicId, MessageContext, RoutedAgent, default_subscription, message_handler


@dataclass
class Message:
    content: int


@default_subscription
class Modifier(RoutedAgent):
    def __init__(self, modify_val: Callable[[int], int]) -> None:
        super().__init__("A modifier agent.")
        self._modify_val = modify_val

    @message_handler
    async def handle_message(self, message: Message, ctx: MessageContext) -> None:
        val = self._modify_val(message.content)
        print(f"{'-'*80}\nModifier:\nModified {message.content} to {val}")
        await self.publish_message(Message(content=val), DefaultTopicId())  # type: ignore


@default_subscription
class Checker(RoutedAgent):
    def __init__(self, run_until: Callable[[int], bool]) -> None:
        super().__init__("A checker agent.")
        self._run_until = run_until

    @message_handler
    async def handle_message(self, message: Message, ctx: MessageContext) -> None:
        if not self._run_until(message.content):
            print(f"{'-'*80}\nChecker:\n{message.content} passed the check, continue.")
            await self.publish_message(Message(content=message.content), DefaultTopicId())
        else:
            print(f"{'-'*80}\nChecker:\n{message.content} failed the check, stopping.")

여기서 한 가지 눈여겨볼 점이 있어요. 에이전트의 로직(모델을 쓰든 코드 실행기를 쓰든)은 메시지가 어떻게 전달되는지와 완전히 분리되어 있어요. 이게 바로 이 프레임워크의 핵심 아이디어입니다. 즉 프레임워크는 통신 인프라를 제공하고, 에이전트는 자기 로직에만 책임을 지는 구조죠. 이 통신 인프라를 Agent Runtime(에이전트 런타임) 이라고 불러요.

에이전트 런타임은 이 프레임워크의 핵심 개념이라서, 메시지를 전달하는 것 외에도 에이전트의 생명주기(lifecycle)를 관리합니다. 그래서 에이전트 생성도 런타임이 담당해요.

아래 코드는 로컬에 내장된 구현체인 SingleThreadedAgentRuntime으로 에이전트를 등록하고 실행하는 방법을 보여줘요.

참고: VSCode나 다른 편집기를 쓴다면 asyncio를 import 하고, 코드를 async def main() -> None:으로 감싼 뒤 asyncio.run(main())으로 실행하는 걸 잊지 마세요.

from autogen_core import AgentId, SingleThreadedAgentRuntime

# Create a local embedded runtime.
runtime = SingleThreadedAgentRuntime()

# Register the modifier and checker agents by providing
# their agent types, the factory functions for creating instance and subscriptions.
await Modifier.register(
    runtime,
    "modifier",
    # Modify the value by subtracting 1
    lambda: Modifier(modify_val=lambda x: x - 1),
)

await Checker.register(
    runtime,
    "checker",
    # Run until the value is less than or equal to 1
    lambda: Checker(run_until=lambda x: x <= 1),
)

# Start the runtime and send a direct message to the checker.
runtime.start()
await runtime.send_message(Message(10), AgentId("checker", "default"))
await runtime.stop_when_idle()

실행 결과를 보면, modifier와 checker의 조건이 시키는 대로 값이 10에서 1까지 정확히 줄어드는 걸 확인할 수 있어요.

AutoGen은 또한 분산 에이전트 런타임을 지원해요. 이 경우 서로 다른 프로세스나 머신에서, 서로 다른 정체성·언어·의존성을 가진 에이전트들을 호스팅할 수 있죠.

에이전트 런타임, 통신, 메시지 처리, 구독을 어떻게 쓰는지 더 알고 싶다면 이 퀵스타트 뒤에 이어지는 섹션들을 계속 읽어 보세요.