인터벤션 핸들러로 종료하기

인터벤션 핸들러로 종료하기 (Termination using Intervention Handler)

이 방법은 autogen_core.SingleThreadedAgentRuntime을 사용할 때 유효해요.

autogen_core에서 종료(termination)를 처리하는 방법은 여러 가지가 있어요. 궁극적으로 목표는 런타임을 더 이상 실행할 필요가 없음을 감지하고, finalization 작업으로 넘어갈 수 있다는 것을 알아내는 거예요. 한 가지 방법은 autogen_core.base.intervention.InterventionHandler를 사용해 종료 메시지를 감지하고 그에 따라 동작하는 거예요.

출처: 공식 문서 - Termination using Intervention Handler

from dataclasses import dataclass
from typing import Any

from autogen_core import (
    DefaultInterventionHandler,
    DefaultTopicId,
    MessageContext,
    RoutedAgent,
    SingleThreadedAgentRuntime,
    default_subscription,
    message_handler,
)

먼저 일반 메시지와 종료를 알리는 데 쓸 메시지용 dataclass를 정의해요.

@dataclass
class Message:
    content: Any


@dataclass
class Termination:
    reason: str

에이전트가 종료하기로 결정했을 때 종료 메시지를 게시하도록 코드를 작성해요.

@default_subscription
class AnAgent(RoutedAgent):
    def __init__(self) -> None:
        super().__init__("MyAgent")
        self.received = 0

    @message_handler
    async def on_new_message(self, message: Message, ctx: MessageContext) -> None:
        self.received += 1
        if self.received > 3:
            await self.publish_message(Termination(reason="Reached maximum number of messages"), DefaultTopicId())

다음으로, 종료 메시지를 감지하고 그에 따라 동작하는 InterventionHandler를 만들어요. 이 핸들러는 publish에 끼어들어(hook) Termination을 만나면 내부 상태를 바꿔 종료가 요청됐음을 나타내요.

class TerminationHandler(DefaultInterventionHandler):
    def __init__(self) -> None:
        self._termination_value: Termination | None = None

    async def on_publish(self, message: Any, *, message_context: MessageContext) -> Any:
        if isinstance(message, Termination):
            self._termination_value = message
        return message

    @property
    def termination_value(self) -> Termination | None:
        return self._termination_value

    @property
    def has_terminated(self) -> bool:
        return self._termination_value is not None

마지막으로 이 핸들러를 런타임에 추가하고, 종료 메시지를 받았을 때 종료를 감지해 런타임을 멈추는 데 사용해요.

termination_handler = TerminationHandler()
runtime = SingleThreadedAgentRuntime(intervention_handlers=[termination_handler])

await AnAgent.register(runtime, "my_agent", AnAgent)

runtime.start()

# Publish more than 3 messages to trigger termination.
await runtime.publish_message(Message("hello"), DefaultTopicId())
await runtime.publish_message(Message("hello"), DefaultTopicId())
await runtime.publish_message(Message("hello"), DefaultTopicId())
await runtime.publish_message(Message("hello"), DefaultTopicId())

# Wait for termination.
await runtime.stop_when(lambda: termination_handler.has_terminated)

print(termination_handler.termination_value)

핵심 패턴을 정리하면 이래요. 인터벤션 핸들러는 메시지가 게시·수신되는 흐름에 끼어들어 관찰할 수 있는 훅이에요. 여기서는 Termination 메시지가 게시될 때 그 값을 내부에 저장하고, 런타임은 stop_when(lambda: termination_handler.has_terminated)로 그 조건이 충족될 때까지 실행하다가 종료돼요. 에이전트가 언제 멈춰야 할지 스스로 아는 대신, 훅이 중립적으로 "언제 끝났는지"를 판단하는 구조예요.

더 알아보기 (Learn more)