使用干预处理器实现终止

备注

此方法仅在使用 SingleThreadedAgentRuntime 时有效。

在 autogen_core 中有多种处理终止的方式。最终目标都是检测到运行时不再需要执行，从而可以继续执行最终化任务。其中一种方法是使用 autogen_core.base.intervention.InterventionHandler 来检测终止消息并据此采取行动。

from dataclasses import dataclass
from typing import Any

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

首先，我们为常规消息和用于发出终止信号的消息定义一个数据类。

@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())

接下来，我们创建一个干预处理器来检测终止消息并采取行动。这个处理器会监听发布事件，当遇到 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()

# 发布超过3条消息以触发终止。
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())

# 等待终止。
await runtime.stop_when(lambda: termination_handler.has_terminated)

print(termination_handler.termination_value)

Termination(reason='Reached maximum number of messages')