동시 에이전트(백그라운드 실행)

동시 에이전트(백그라운드 실행)

하나의 메시지를 여러 에이전트가 동시에 처리하도록 만들고 싶을 때가 있어요. 이번 섹션에서는 여러 에이전트가 동시에 작업하는 방법을 세 가지 패턴으로 살펴볼게요.

  1. 단일 메시지 & 다중 프로세서 — 같은 토픽을 구독한 여러 에이전트가 하나의 메시지를 동시에 처리
  2. 다중 메시지 & 다중 프로세서 — 토픽에 따라 특정 메시지 타입을 전담 에이전트로 라우팅
  3. 직접 메시징(Direct Messaging) — 에이전트끼리, 그리고 런타임에서 에이전트로 메시지를 보내기

출처: 공식문서

import asyncio
from dataclasses import dataclass

from autogen_core import (
    AgentId,
    ClosureAgent,
    ClosureContext,
    DefaultTopicId,
    MessageContext,
    RoutedAgent,
    SingleThreadedAgentRuntime,
    TopicId,
    TypeSubscription,
    default_subscription,
    message_handler,
    type_subscription,
)

@dataclass
class Task:
    task_id: str


@dataclass
class TaskResponse:
    task_id: str
    result: str

단일 메시지 & 다중 프로세서

첫 번째 패턴은 하나의 메시지를 여러 에이전트가 동시에 처리하는 방법을 보여줘요.

  • 각 Processor 에이전트는 default_subscription 데코레이터로 기본 토픽을 구독해요.
  • 기본 토픽에 메시지를 게시하면, 등록된 모든 에이전트가 그 메시지를 독립적으로 처리해요.
아래에서 `Processor`를 `default_subscription` 데코레이터로 구독하고 있어요. 데코레이터 없이 에이전트를 구독하는 대안은 [Subscribe and Publish to Topics](../framework/message-and-communication.ipynb#subscribe-and-publish-to-topics)에서 확인할 수 있어요. 이 방법으로는 같은 에이전트 클래스를 서로 다른 토픽에 구독시킬 수도 있어요.

@default_subscription
class Processor(RoutedAgent):
    @message_handler
    async def on_task(self, message: Task, ctx: MessageContext) -> None:
        print(f"{self._description} starting task {message.task_id}")
        await asyncio.sleep(2)  # Simulate work
        print(f"{self._description} finished task {message.task_id}")

runtime = SingleThreadedAgentRuntime()

await Processor.register(runtime, "agent_1", lambda: Processor("Agent 1"))
await Processor.register(runtime, "agent_2", lambda: Processor("Agent 2"))

runtime.start()

await runtime.publish_message(Task(task_id="task-1"), topic_id=DefaultTopicId())

await runtime.stop_when_idle()

다중 메시지 & 다중 프로세서

두 번째 패턴은 서로 다른 타입의 메시지를 특정 프로세서로 라우팅하는 방법을 보여줘요.

  • UrgentProcessor는 "urgent" 토픽 구독
  • NormalProcessor는 "normal" 토픽 구독

에이전트가 특정 토픽 타입을 구독하게 하려면 type_subscription 데코레이터를 사용해요.

TASK_RESULTS_TOPIC_TYPE = "task-results"
task_results_topic_id = TopicId(type=TASK_RESULTS_TOPIC_TYPE, source="default")


@type_subscription(topic_type="urgent")
class UrgentProcessor(RoutedAgent):
    @message_handler
    async def on_task(self, message: Task, ctx: MessageContext) -> None:
        print(f"Urgent processor starting task {message.task_id}")
        await asyncio.sleep(1)  # Simulate work
        print(f"Urgent processor finished task {message.task_id}")

        task_response = TaskResponse(task_id=message.task_id, result="Results by Urgent Processor")
        await self.publish_message(task_response, topic_id=task_results_topic_id)


@type_subscription(topic_type="normal")
class NormalProcessor(RoutedAgent):
    @message_handler
    async def on_task(self, message: Task, ctx: MessageContext) -> None:
        print(f"Normal processor starting task {message.task_id}")
        await asyncio.sleep(3)  # Simulate work
        print(f"Normal processor finished task {message.task_id}")

        task_response = TaskResponse(task_id=message.task_id, result="Results by Normal Processor")
        await self.publish_message(task_response, topic_id=task_results_topic_id)

에이전트를 등록한 뒤 "urgent"와 "normal" 토픽에 메시지를 게시할 수 있어요.

runtime = SingleThreadedAgentRuntime()

await UrgentProcessor.register(runtime, "urgent_processor", lambda: UrgentProcessor("Urgent Processor"))
await NormalProcessor.register(runtime, "normal_processor", lambda: NormalProcessor("Normal Processor"))

runtime.start()

await runtime.publish_message(Task(task_id="normal-1"), topic_id=TopicId(type="normal", source="default"))
await runtime.publish_message(Task(task_id="urgent-1"), topic_id=TopicId(type="urgent", source="default"))

await runtime.stop_when_idle()

결과 수집하기

앞선 예제에서는 작업 완료를 확인하기 위해 콘솔 출력에 의존했어요. 하지만 실제 애플리케이션에서는 결과를 프로그래밍 방식으로 수집하고 처리하길 원하는 경우가 대부분이죠.

이 메시지들을 수집하기 위해 ClosureAgent를 사용할게요. UrgentProcessor와 NormalProcessor가 결과를 게시하는 전용 토픽 TASK_RESULTS_TOPIC_TYPE을 정의했어요. ClosureAgent는 이 토픽의 메시지를 처리합니다.

queue = asyncio.Queue[TaskResponse]()


async def collect_result(_agent: ClosureContext, message: TaskResponse, ctx: MessageContext) -> None:
    await queue.put(message)


runtime.start()

CLOSURE_AGENT_TYPE = "collect_result_agent"
await ClosureAgent.register_closure(
    runtime,
    CLOSURE_AGENT_TYPE,
    collect_result,
    subscriptions=lambda: [TypeSubscription(topic_type=TASK_RESULTS_TOPIC_TYPE, agent_type=CLOSURE_AGENT_TYPE)],
)

await runtime.publish_message(Task(task_id="normal-1"), topic_id=TopicId(type="normal", source="default"))
await runtime.publish_message(Task(task_id="urgent-1"), topic_id=TopicId(type="urgent", source="default"))

await runtime.stop_when_idle()

while not queue.empty():
    print(await queue.get())

직접 메시지 (Direct Messages)

앞선 패턴들과 달리, 이 패턴은 직접 메시지에 초점을 맞춰요. 보내는 방법에는 두 가지가 있어요.

  • 에이전트 간 직접 메시징
  • 런타임에서 특정 에이전트로 메시지 보내기

아래 예제에서 고려할 것들입니다.

  • 메시지는 AgentId로 주소가 지정돼요.
  • 보내는 쪽은 대상 에이전트로부터 응답을 받을 수 있어요.
  • WorkerAgent 클래스는 한 번만 등록하지만, 두 명의 서로 다른 워커에게 작업을 보내요.
    • 어떻게? Agent lifecycle에서 설명하듯, AgentId로 메시지를 전달할 때 런타임은 인스턴스를 가져오거나 없으면 생성해요. 이 경우 두 메시지가 전달될 때 런타임은 워커 인스턴스 두 개를 생성합니다.
class WorkerAgent(RoutedAgent):
    @message_handler
    async def on_task(self, message: Task, ctx: MessageContext) -> TaskResponse:
        print(f"{self.id} starting task {message.task_id}")
        await asyncio.sleep(2)  # Simulate work
        print(f"{self.id} finished task {message.task_id}")
        return TaskResponse(task_id=message.task_id, result=f"Results by {self.id}")


class DelegatorAgent(RoutedAgent):
    def __init__(self, description: str, worker_type: str):
        super().__init__(description)
        self.worker_instances = [AgentId(worker_type, f"{worker_type}-1"), AgentId(worker_type, f"{worker_type}-2")]

    @message_handler
    async def on_task(self, message: Task, ctx: MessageContext) -> TaskResponse:
        print(f"Delegator received task {message.task_id}.")

        subtask1 = Task(task_id="task-part-1")
        subtask2 = Task(task_id="task-part-2")

        worker1_result, worker2_result = await asyncio.gather(
            self.send_message(subtask1, self.worker_instances[0]), self.send_message(subtask2, self.worker_instances[1])
        )

        combined_result = f"Part 1: {worker1_result.result}, " f"Part 2: {worker2_result.result}"
        task_response = TaskResponse(task_id=message.task_id, result=combined_result)
        return task_response

runtime = SingleThreadedAgentRuntime()

await WorkerAgent.register(runtime, "worker", lambda: WorkerAgent("Worker Agent"))
await DelegatorAgent.register(runtime, "delegator", lambda: DelegatorAgent("Delegator Agent", "worker"))

runtime.start()

delegator = AgentId("delegator", "default")
response = await runtime.send_message(Task(task_id="main-task"), recipient=delegator)

print(f"Final result: {response.result}")
await runtime.stop_when_idle()

추가 자료

동시 처리에 더 관심이 있다면 동시 에이전트를 많이 사용하는 Mixture of Agents 패턴을 확인해보세요.

더 알아보기 (Learn more)