메시지와 통신

메시지와 통신

AutoGen Core의 에이전트는 메시지를 반응하고(react), 보내고(send), 게시할(publish) 수 있어요. 그리고 메시지가 곧 에이전트끼리 소통할 수 있는 유일한 수단이에요.

출처: Message and Communication — AutoGen 공식 문서

메시지

메시지는 직렬화 가능한 객체로, 다음 두 가지 방식으로 정의할 수 있어요.

• Pydantic의 BaseModel의 하위 클래스
• dataclass

예를 들어:

from dataclasses import dataclass


@dataclass
class TextMessage:
    content: str
    source: str


@dataclass
class ImageMessage:
    url: str
    source: str

참고: 메시지는 순수한 데이터일 뿐, 어떤 로직도 담고 있으면 안 돼요.

메시지 핸들러

에이전트가 메시지를 받으면 런타임은 에이전트의 메시지 핸들러(on_message())를 호출해요. 이 핸들러가 에이전트의 메시지 처리 로직을 구현해야 합니다. 만약 이 메시지를 해당 에이전트가 처리할 수 없다면, CantHandleException을 던져야 해요.

기본 클래스인 BaseAgent는 메시지 처리 로직을 아예 제공하지 않아요. on_message() 메서드를 직접 구현하는 건 고급 사용 사례가 아니라면 권장하지 않습니다.

개발자는 메시지 라우팅 기능이 내장된 RoutedAgent 기본 클래스를 구현하는 것부터 시작해야 해요.

타입별 메시지 라우팅

RoutedAgent 기본 클래스는 message_handler() 데코레이터로 메시지 타입과 메시지 핸들러를 연결하는 메커니즘을 제공해요. 그래서 개발자가 on_message() 메서드를 직접 구현할 필요가 없죠.

예를 들어, 아래 타입 라우팅 에이전트는 TextMessage와 ImageMessage를 각각 다른 메시지 핸들러로 응답해요.

from autogen_core import AgentId, MessageContext, RoutedAgent, SingleThreadedAgentRuntime, message_handler


class MyAgent(RoutedAgent):
    @message_handler
    async def on_text_message(self, message: TextMessage, ctx: MessageContext) -> None:
        print(f"Hello, {message.source}, you said {message.content}!")

    @message_handler
    async def on_image_message(self, message: ImageMessage, ctx: MessageContext) -> None:
        print(f"Hello, {message.source}, you sent me {message.url}!")

에이전트 런타임을 만들고 에이전트 타입을 등록해요 (Agent and Agent Runtime 참고):

runtime = SingleThreadedAgentRuntime()
await MyAgent.register(runtime, "my_agent", lambda: MyAgent("My Agent"))

TextMessage와 ImageMessage로 이 에이전트를 테스트해 봐요.

runtime.start()
agent_id = AgentId("my_agent", "default")
await runtime.send_message(TextMessage(content="Hello, World!", source="User"), agent_id)
await runtime.send_message(ImageMessage(url="https://example.com/image.jpg", source="User"), agent_id)
await runtime.stop_when_idle()

런타임은 첫 메시지를 전달할 때 AgentId("my_agent", "default")라는 에이전트 ID로 MyAgent 인스턴스를 자동으로 만들어요.

같은 타입의 메시지 라우팅

어떤 시나리오에서는 같은 타입의 메시지를 서로 다른 핸들러로 라우팅하는 게 유용해요. 예를 들어 서로 다른 발신자 에이전트가 보낸 메시지를 다르게 처리해야 하는 경우죠. 이때 message_handler() 데코레이터의 match 매개변수를 쓰면 돼요.

match 매개변수는 같은 메시지 타입의 핸들러를 특정 메시지와 연결해요. 이건 메시지 타입 라우팅의 부차적인(secondary) 기준이에요. 메시지와 MessageContext를 인자로 받아, 그 메시지를 데코레이터가 달린 핸들러가 처리해야 하는지 여부를 나타내는 불리언을 반환하는 callable을 받죠. 이 callable은 핸들러의 알파벳 순서로 검사돼요.

match 매개변수로 발신자 에이전트에 따라 메시지를 라우팅하는 에이전트 예시를 볼게요.

class RoutedBySenderAgent(RoutedAgent):
    @message_handler(match=lambda msg, ctx: msg.source.startswith("user1"))  # type: ignore
    async def on_user1_message(self, message: TextMessage, ctx: MessageContext) -> None:
        print(f"Hello from user 1 handler, {message.source}, you said {message.content}!")

    @message_handler(match=lambda msg, ctx: msg.source.startswith("user2"))  # type: ignore
    async def on_user2_message(self, message: TextMessage, ctx: MessageContext) -> None:
        print(f"Hello from user 2 handler, {message.source}, you said {message.content}!")

    @message_handler(match=lambda msg, ctx: msg.source.startswith("user2"))  # type: ignore
    async def on_image_message(self, message: ImageMessage, ctx: MessageContext) -> None:
        print(f"Hello, {message.source}, you sent me {message.url}!")

위 에이전트는 메시지의 source 필드로 발신자 에이전트를 판별해요. MessageContext의 sender 필드를 이용해, 가능하면 에이전트 ID로 발신자 에이전트를 판별할 수도 있어요.

source 값이 다른 메시지들로 이 에이전트를 테스트해 봐요.

runtime = SingleThreadedAgentRuntime()
await RoutedBySenderAgent.register(runtime, "my_agent", lambda: RoutedBySenderAgent("Routed by sender agent"))
runtime.start()
agent_id = AgentId("my_agent", "default")
await runtime.send_message(TextMessage(content="Hello, World!", source="user1-test"), agent_id)
await runtime.send_message(TextMessage(content="Hello, World!", source="user2-test"), agent_id)
await runtime.send_message(ImageMessage(url="https://example.com/image.jpg", source="user1-test"), agent_id)
await runtime.send_message(ImageMessage(url="https://example.com/image.jpg", source="user2-test"), agent_id)
await runtime.stop_when_idle()

위 예제에서 첫 번째 ImageMessage는 처리되지 않았어요. 메시지의 source 필드가 핸들러의 match 조건과 일치하지 않았기 때문이죠.

직접 메시지(Direct Messaging)

AutoGen Core에는 두 가지 통신 방식이 있어요.

• 직접 메시지(Direct Messaging): 다른 에이전트에게 직접 메시지를 보냅니다.
• 브로드캐스트(Broadcast): 토픽에 메시지를 게시합니다.

먼저 직접 메시지를 살펴볼게요. 다른 에이전트에게 직접 메시지를 보내려면, 메시지 핸들러 안에서는 autogen_core.BaseAgent.send_message() 메서드를 쓰고, 런타임에서는 autogen_core.AgentRuntime.send_message() 메서드를 쓰세요. 이 메서드를 await 하면 수신 에이전트의 메시지 핸들러가 반환하는 값을 돌려받아요. 수신 에이전트의 핸들러가 None을 반환하면 None이 반환돼요.

참고: 발신자가 await 하는 동안 호출된 에이전트가 예외를 던지면, 그 예외는 발신자에게 다시 전파돼요.

요청/응답(Request/Response)

직접 메시지는 발신자가 수신자로부터 응답을 기대하는 요청/응답 시나리오에 쓸 수 있어요. 수신자는 메시지 핸들러에서 값을 반환하는 방식으로 메시지에 응답할 수 있죠. 이건 에이전트 사이의 함수 호출처럼 생각하면 돼요.

예를 들어, 아래 에이전트들을 생각해 봐요.

from dataclasses import dataclass

from autogen_core import MessageContext, RoutedAgent, SingleThreadedAgentRuntime, message_handler


@dataclass
class Message:
    content: str


class InnerAgent(RoutedAgent):
    @message_handler
    async def on_my_message(self, message: Message, ctx: MessageContext) -> Message:
        return Message(content=f"Hello from inner, {message.content}")


class OuterAgent(RoutedAgent):
    def __init__(self, description: str, inner_agent_type: str):
        super().__init__(description)
        self.inner_agent_id = AgentId(inner_agent_type, self.id.key)

    @message_handler
    async def on_my_message(self, message: Message, ctx: MessageContext) -> None:
        print(f"Received message: {message.content}")
        # Send a direct message to the inner agent and receives a response.
        response = await self.send_message(Message(f"Hello from outer, {message.content}"), self.inner_agent_id)
        print(f"Received inner response: {response.content}")

메시지를 받으면 OuterAgent는 InnerAgent에게 직접 메시지를 보내고 응답을 받아요.

OuterAgent에 Message를 보내서 이 에이전트들을 테스트할 수 있어요.

runtime = SingleThreadedAgentRuntime()
await InnerAgent.register(runtime, "inner_agent", lambda: InnerAgent("InnerAgent"))
await OuterAgent.register(runtime, "outer_agent", lambda: OuterAgent("OuterAgent", "inner_agent"))
runtime.start()
outer_agent_id = AgentId("outer_agent", "default")
await runtime.send_message(Message(content="Hello, World!"), outer_agent_id)
await runtime.stop_when_idle()

두 출력 모두 OuterAgent의 메시지 핸들러가 만들어 냈지만, 두 번째 출력은 InnerAgent의 응답에 기반해 만들어진 거예요.

일반적으로 직접 메시지는 발신자와 수신자가 강하게 결합된(tightly coupled) 시나리오에 적합해요. 즉 둘이 함께 생성되고, 발신자가 수신자의 특정 인스턴스에 연결되는 경우죠. 예를 들어 에이전트가 도구 호출을 실행할 때 ToolAgent 인스턴스에 직접 메시지를 보내고, 그 응답으로 행동-관찰(action-observation) 루프를 구성하는 식이에요.

브로드캐스트(Broadcast)

브로드캐스트는 사실상 토픽과 구독이 있는 발행/구독(publish/subscribe) 모델이에요. 핵심 개념은 Topic and Subscription에서 배울 수 있어요.

직접 메시지와 브로드캐스트의 핵심 차이는, 브로드캐스트는 요청/응답 시나리오에 쓸 수 없다는 점이에요. 에이전트가 메시지를 게시하면 일방향(one way)일 뿐이라, 수신 에이전트의 핸들러가 값을 반환해도 어떤 다른 에이전트로부터도 응답을 받을 수 없어요.

참고: 게시된 메시지에 응답이 달리면 그 응답은 버려져요.

참고: 에이전트가 자신이 구독 중인 메시지 타입을 게시하면, 그 게시한 메시지를 자기 자신은 받지 않아요. 이건 무한 루프를 막기 위해서예요.

토픽 구독과 게시

타입 기반 구독(Type-based subscription)은 주어진 토픽 타입의 토픽에 게시된 메시지를, 주어진 에이전트 타입의 에이전트에게 매핑해요. RoutedAgent를 상속하는 에이전트가 주어진 토픽 타입의 토픽을 구독하게 하려면, type_subscription() 클래스 데코레이터를 쓰면 돼요.

아래 예제는 "default" 토픽 타입의 토픽을 type_subscription() 데코레이터로 구독하고, 받은 메시지를 출력하는 ReceiverAgent 클래스를 보여줘요.

from autogen_core import RoutedAgent, message_handler, type_subscription


@type_subscription(topic_type="default")
class ReceivingAgent(RoutedAgent):
    @message_handler
    async def on_my_message(self, message: Message, ctx: MessageContext) -> None:
        print(f"Received a message: {message.content}")

에이전트의 핸들러에서 메시지를 게시하려면, BaseAgent.publish_message 메서드를 쓰고 TopicId를 지정해요. 이 호출은 런타임이 모든 구독자에게 메시지 전달을 스케줄링하도록 여전히 await 되어야 하지만, 항상 None을 반환해요. 에이전트가 게시된 메시지를 처리하다 예외를 던지면, 이는 로깅되지만 게시한 에이전트에게는 전파되지 않아요.

아래 예제는 메시지를 받으면 토픽에 메시지를 게시하는 BroadcastingAgent를 보여줘요.

from autogen_core import TopicId


class BroadcastingAgent(RoutedAgent):
    @message_handler
    async def on_my_message(self, message: Message, ctx: MessageContext) -> None:
        await self.publish_message(
            Message("Publishing a message from broadcasting agent!"),
            topic_id=TopicId(type="default", source=self.id.key),
        )

BroadcastingAgent는 타입 "default"이고 소스가 에이전트 인스턴스의 에이전트 키인 토픽에 메시지를 게시해요.

구독은 에이전트 런타임에 등록돼요. 에이전트 타입 등록의 일부로 등록하거나 별도 API 메서드로 등록하면 됩니다. 아래는 수신 에이전트용 TypeSubscription을 type_subscription() 데코레이터로, 게시 에이전트용으론 데코레이터 없이 등록하는 방법이에요.

from autogen_core import TypeSubscription

runtime = SingleThreadedAgentRuntime()

# Option 1: with type_subscription decorator
# The type_subscription class decorator automatically adds a TypeSubscription to
# the runtime when the agent is registered.
await ReceivingAgent.register(runtime, "receiving_agent", lambda: ReceivingAgent("Receiving Agent"))

# Option 2: with TypeSubscription
await BroadcastingAgent.register(runtime, "broadcasting_agent", lambda: BroadcastingAgent("Broadcasting Agent"))
await runtime.add_subscription(TypeSubscription(topic_type="default", agent_type="broadcasting_agent"))

# Start the runtime and publish a message.
runtime.start()
await runtime.publish_message(
    Message("Hello, World! From the runtime!"), topic_id=TopicId(type="default", source="default")
)
await runtime.stop_when_idle()

위 예제에서 보듯, 에이전트 인스턴스를 만들 필요 없이 런타임의 publish_message() 메서드를 통해 토픽에 직접 게시할 수도 있어요.

출력에서 수신 에이전트가 두 메시지를 받은 걸 볼 수 있어요. 하나는 런타임을 통해 게시되었고, 다른 하나는 게시 에이전트가 게시했어요.

기본 토픽과 기본 구독

위 예제에서는 토픽과 구독을 지정하기 위해 TopicId와 TypeSubscription을 썼어요. 이건 많은 시나리오에 적합한 방법이에요. 하지만 게시 범위가 하나뿐인 경우, 즉 모든 에이전트가 브로드캐스트되는 모든 메시지를 게시하고 구독하는 경우에는, 편의 클래스인 DefaultTopicId와 default_subscription()을 써서 코드를 간단히 만들 수 있어요.

DefaultTopicId는 토픽 타입의 기본값으로 "default"를, 토픽 소스의 기본값으로 게시 에이전트의 키를 사용하는 토픽을 만들기 위한 거예요. default_subscription()은 기본 토픽을 구독하는 타입 구독을 만들기 위한 것이에요. BroadcastingAgent를 DefaultTopicId와 default_subscription()으로 간단히 만들 수 있어요.

from autogen_core import DefaultTopicId, default_subscription


@default_subscription
class BroadcastingAgentDefaultTopic(RoutedAgent):
    @message_handler
    async def on_my_message(self, message: Message, ctx: MessageContext) -> None:
        # Publish a message to all agents in the same namespace.
        await self.publish_message(
            Message("Publishing a message from broadcasting agent!"),
            topic_id=DefaultTopicId(),
        )

런타임이 register()를 호출해 에이전트 타입을 등록하면, 토픽 타입으로 "default"가 기본값으로 쓰이고 에이전트 타입이 같은 컨텍스트에서 등록되는 에이전트 타입과 같은 TypeSubscription을 만들어요.

runtime = SingleThreadedAgentRuntime()
await BroadcastingAgentDefaultTopic.register(
    runtime, "broadcasting_agent", lambda: BroadcastingAgentDefaultTopic("Broadcasting Agent")
)
await ReceivingAgent.register(runtime, "receiving_agent", lambda: ReceivingAgent("Receiving Agent"))
runtime.start()
await runtime.publish_message(Message("Hello, World! From the runtime!"), topic_id=DefaultTopicId())
await runtime.stop_when_idle()

참고: 시나리오가 모든 에이전트가 브로드캐스트되는 모든 메시지를 게시·구독하도록 허용한다면, DefaultTopicId와 default_subscription()으로 에이전트 클래스를 데코레이션하세요.

더 알아보기 (Learn more)

• Agent and Agent Runtime
• Topic and Subscription
• Quick Start