분산 에이전트 런타임(Distributed Agent Runtime)

경고: 분산 에이전트 런타임은 실험적 기능이에요. API에 호환되지 않는 변경(breaking changes)이 있을 것으로 예상하세요.

분산 에이전트 런타임은 프로세스 경계를 넘어 통신과 에이전트 생명주기 관리를 가능하게 해줘요. 이 런타임은 호스트 서비스(host service) 와 최소 하나의 워커 런타임(worker runtime) 으로 구성됩니다.

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

호스트 서비스는 모든 활성 워커 런타임과의 연결을 유지하고, 메시지 전달을 도우며, 모든 직접 메시지(즉 RPC)에 대한 세션을 유지해요. 워커 런타임은 애플리케이션 코드(에이전트)를 처리하고 호스트 서비스에 연결돼요. 또한 자신이 지원하는 에이전트를 호스트 서비스에 광고해서, 호스트 서비스가 메시지를 올바른 워커에게 전달할 수 있게 해요.

참고: 분산 에이전트 런타임은 추가 의존성이 필요해요. 아래처럼 설치하세요.
pip install "autogen-ext[grpc]"

GrpcWorkerAgentRuntimeHost를 사용해 호스트 서비스를 시작할 수 있어요.

from autogen_ext.runtimes.grpc import GrpcWorkerAgentRuntimeHost

host = GrpcWorkerAgentRuntimeHost(address="localhost:50051")
host.start()  # Start a host service in the background.

위 코드는 호스트 서비스를 백그라운드로 시작하고, 포트 50051에서 워커 연결을 받아들여요.

워커 런타임을 실행하기 전에 에이전트를 정의해 봐요. 이 에이전트는 받는 메시지마다 새 메시지를 게시해요. 또한 자신이 게시한 메시지 수를 추적해서, 5개의 메시지를 게시하면 더 이상 새 메시지를 게시하지 않아요.

from dataclasses import dataclass

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


@dataclass
class MyMessage:
    content: str


@default_subscription
class MyAgent(RoutedAgent):
    def __init__(self, name: str) -> None:
        super().__init__("My agent")
        self._name = name
        self._counter = 0

    @message_handler
    async def my_message_handler(self, message: MyMessage, ctx: MessageContext) -> None:
        self._counter += 1
        if self._counter > 5:
            return
        content = f"{self._name}: Hello x {self._counter}"
        print(content)
        await self.publish_message(MyMessage(content=content), DefaultTopicId())

이제 워커 에이전트 런타임을 설정해 봐요. GrpcWorkerAgentRuntime을 사용합니다. 워커 런타임을 두 개 만들 거예요. 각 런타임은 에이전트 하나를 호스팅해요. 모든 에이전트는 기본 토픽에 게시·구독하므로, 게시되는 모든 메시지를 볼 수 있어요.

에이전트를 실행하려면 워커에서 메시지를 게시해요.

import asyncio

from autogen_ext.runtimes.grpc import GrpcWorkerAgentRuntime

worker1 = GrpcWorkerAgentRuntime(host_address="localhost:50051")
await worker1.start()
await MyAgent.register(worker1, "worker1", lambda: MyAgent("worker1"))

worker2 = GrpcWorkerAgentRuntime(host_address="localhost:50051")
await worker2.start()
await MyAgent.register(worker2, "worker2", lambda: MyAgent("worker2"))

await worker2.publish_message(MyMessage(content="Hello!"), DefaultTopicId())

# Let the agents run for a while.
await asyncio.sleep(5)

각 에이전트가 정확히 5개의 메시지를 게시한 걸 볼 수 있어요.

워커 런타임을 멈추려면 stop()을 호출하면 돼요.

await worker1.stop()
await worker2.stop()

# To keep the worker running until a termination signal is received (e.g., SIGTERM).
# await worker1.stop_when_signal()

호스트 서비스를 멈추려면 stop()을 호출해요.

await host.stop()

# To keep the host service running until a termination signal (e.g., SIGTERM)
# await host.stop_when_signal()

크로스 언어 런타임(Cross-Language Runtimes)

위에서 설명한 절차는 대체로 동일해요. 하지만 모든 에이전트 간 메시지 타입은 공유 protobuf 스키마를 반드시 사용해야 해요.

다음 단계

분산 런타임의 완전한 예제는 아래 샘플들을 참고하세요.

분산 에이전트 런타임(Distributed Agent Runtime)