에이전트와 에이전트 런타임

이번 섹션과 다음 섹션에서는 AutoGen의 핵심 개념을 집중적으로 다뤄요. 에이전트(agent), 에이전트 런타임(agent runtime), 메시지, 통신 — 이 네 가지가 멀티에이전트 애플리케이션을 만드는 기초 블록이에요.

출처: Agent and Agent Runtime — AutoGen 공식 문서

참고: Core API는 의견을 강요하지 않는(unopinionated) 유연한 API로 설계되었어요. 그래서 가끔은 쓰기 까다로울 수 있는데요, 인터랙티브하고 확장 가능하며 분산된 멀티에이전트 시스템을 만들면서 모든 워크플로우를 완전히 통제하고 싶은 사람에게 적합합니다. 빠르게 뭔가 동작하는 걸 만들고 싶다면 AgentChat API가 더 쉽겠어요.

AutoGen의 에이전트는 기본 인터페이스인 Agent로 정의되는 엔티티예요. 여기에는 고유 식별자(AgentId 타입)와 메타데이터 딕셔너리(AgentMetadata 타입)가 붙어요.

대부분의 경우, 더 높은 수준의 클래스인 RoutedAgent를 상속해서 에이전트를 만들 수 있어요. 그러면 message_handler() 데코레이터와 message 변수의 타입 힌트를 이용해, 들어온 메시지를 적절한 메시지 핸들러로 자동 라우팅해 줘요. 그리고 에이전트 런타임은 AutoGen에서 에이전트가 실행되는 실행 환경이에요.

에이전트 런타임은 프로그래밍 언어의 런타임 환경과 비슷해요. 에이전트 간 통신을 도와주고, 에이전트 생명주기를 관리하고, 보안 경계를 강제하며, 모니터링과 디버깅을 지원하는 데 필요한 인프라를 제공하죠.

로컬 개발에서는 SingleThreadedAgentRuntime을 쓸 수 있는데, 이건 파이썬 애플리케이션에 내장해 사용할 수 있어요.

참고: 에이전트는 애플리케이션 코드가 직접 인스턴스화하고 관리하지 않아요. 대신 필요할 때 런타임이 만들고 런타임이 관리합니다.

이미 AgentChat에 익숙하다면 주의할 점이 있어요. AgentChat의 에이전트(AssistantAgent 등)는 애플리케이션이 만들어서 런타임이 직접 관리하지 않아요. AgentChat 에이전트를 Core에서 쓰려면, 메시지를 그 AgentChat 에이전트에게 위임하는 래퍼(wrapper) Core 에이전트를 만들고, 그 래퍼를 런타임이 관리하게 하면 됩니다.

에이전트 구현하기

에이전트를 구현하려면 RoutedAgent 클래스를 상속하고, 에이전트가 처리해야 하는 각 메시지 타입에 대해 message_handler() 데코레이터를 단 메시지 핸들러 메서드를 구현해야 해요. 예를 들어, 아래 에이전트는 간단한 MyMessageType 타입을 처리하고 받은 메시지를 출력해요.

from dataclasses import dataclass

from autogen_core import AgentId, MessageContext, RoutedAgent, message_handler


@dataclass
class MyMessageType:
    content: str


class MyAgent(RoutedAgent):
    def __init__(self) -> None:
        super().__init__("MyAgent")

    @message_handler
    async def handle_my_message_type(self, message: MyMessageType, ctx: MessageContext) -> None:
        print(f"{self.id.type} received message: {message.content}")

이 에이전트는 MyMessageType만 처리하고, 메시지는 handle_my_message_type 메서드로 전달돼요. 서로 다른 메시지 타입에 대해 message_handler() 데코레이터를 쓰고 핸들러 함수의 message 변수에 타입 힌트를 달면, 여러 개의 메시지 핸들러를 가질 수 있어요. 에이전트 로직에 더 잘 맞는다면, 한 메시지 핸들러 함수 안의 message 변수에 파이썬 typing union을 활용할 수도 있어요. 더 자세한 내용은 다음의 메시지와 통신 섹션을 보세요.

AgentChat 에이전트 사용하기

AgentChat 에이전트가 있는데 Core API에서 쓰고 싶다면, 그 에이전트에게 메시지를 위임하는 래퍼 RoutedAgent를 만들면 돼요. 아래 예제는 AgentChat의 AssistantAgent용 래퍼 에이전트를 만드는 방법을 보여줘요.

from autogen_agentchat.agents import AssistantAgent
from autogen_agentchat.messages import TextMessage
from autogen_ext.models.openai import OpenAIChatCompletionClient


class MyAssistant(RoutedAgent):
    def __init__(self, name: str) -> None:
        super().__init__(name)
        model_client = OpenAIChatCompletionClient(model="gpt-4o")
        self._delegate = AssistantAgent(name, model_client=model_client)

    @message_handler
    async def handle_my_message_type(self, message: MyMessageType, ctx: MessageContext) -> None:
        print(f"{self.id.type} received message: {message.content}")
        response = await self._delegate.on_messages(
            [TextMessage(content=message.content, source="user")], ctx.cancellation_token
        )
        print(f"{self.id.type} responded: {response.chat_message}")

모델 클라이언트 사용법은 Model Client 섹션을 참고하세요.

Core API는 의견을 강요하지 않기 때문에, Core API를 쓰기 위해 반드시 AgentChat API를 써야 하는 건 아니에요. 직접 에이전트를 구현하거나 다른 에이전트 프레임워크를 써도 돼요.

에이전트 타입 등록하기

에이전트를 런타임에서 사용할 수 있게 하려면, BaseAgent 클래스의 register() 클래스 메서드를 쓰면 돼요. 등록 절차는 문자열로 고유하게 식별되는 에이전트 타입(agent type) 과, 해당 클래스의 에이전트 타입 인스턴스를 만들어 주는 팩토리 함수(factory function) 를 연결합니다. 이 팩토리 함수 덕분에 필요할 때 에이전트 인스턴스가 자동으로 생성될 수 있어요.

에이전트 타입(AgentType)은 에이전트 클래스와 다르다는 점에 주의하세요. 이 예제에서 에이전트 타입은 AgentType("my_agent") 또는 AgentType("my_assistant")이고, 에이전트 클래스는 파이썬 클래스인 MyAgent 또는 MyAssistantAgent예요. 팩토리 함수는 register() 클래스 메서드가 호출된 에이전트 클래스의 인스턴스를 반환해야 해요. 에이전트 타입과 정체성에 대해 더 알고 싶다면 Agent Identity and Lifecycles를 읽어 보세요.

참고: 서로 다른 에이전트 타입을, 같은 에이전트 클래스를 반환하는 팩토리 함수로 등록할 수 있어요. 예를 들어 팩토리 함수 안에서 생성자 매개변수를 다르게 해서 같은 에이전트 클래스의 다른 인스턴스를 만들 수 있죠.

아래 코드는 에이전트 타입을 SingleThreadedAgentRuntime에 등록하는 방법이에요.

from autogen_core import SingleThreadedAgentRuntime

runtime = SingleThreadedAgentRuntime()
await MyAgent.register(runtime, "my_agent", lambda: MyAgent())
await MyAssistant.register(runtime, "my_assistant", lambda: MyAssistant("my_assistant"))

에이전트 타입이 등록되면 AgentId를 이용해 에이전트 인스턴스에 직접 메시지를 보낼 수 있어요. 런타임은 이 인스턴스에 처음 메시지를 전달할 때 그 인스턴스를 생성해요.

runtime.start()  # Start processing messages in the background.
await runtime.send_message(MyMessageType("Hello, World!"), AgentId("my_agent", "default"))
await runtime.send_message(MyMessageType("Hello, World!"), AgentId("my_assistant", "default"))
await runtime.stop()  # Stop processing messages in the background.

참고: 런타임이 에이전트의 생명주기를 관리하기 때문에, AgentId는 에이전트와 통신하거나 그 메타데이터(예: 설명)를 가져올 때만 사용해요.

싱글 스레드 에이전트 런타임 실행하기

위 코드는 start()를 사용해 백그라운드 태스크를 시작하고, 수신자들의 메시지 핸들러에 메시지를 처리·전달해요. 이건 로컬 내장 런타임인 SingleThreadedAgentRuntime의 기능이에요.

백그라운드 태스크를 즉시 멈추려면 stop() 메서드를 쓰세요.

runtime.start()
# ... Send messages, publish messages, etc.
await runtime.stop()  # This will return immediately but will not cancel
# any in-progress message handling.

start()를 다시 호출하면 백그라운드 태스크를 재개할 수 있어요.

에이전트를 평가하는 벤치마크 같은 배치(batch) 시나리오에서는, 처리되지 않은 메시지도 없고 에이전트가 메시지를 처리 중이지도 않을 때 백그라운드 태스크가 자동으로 멈추기를 기다리고 싶을 거예요. 이럴 때 stop_when_idle() 메서드를 쓰면 됩니다.

runtime.start()
# ... Send messages, publish messages, etc.
await runtime.stop_when_idle()  # This will block until the runtime is idle.

런타임을 닫고 리소스를 해제하려면 close() 메서드를 쓰세요.

await runtime.close()

다른 런타임 구현체는 각자만의 방식으로 런타임을 실행합니다.

에이전트와 에이전트 런타임