分布式代理运行时

注意

分布式代理运行时是一项实验性功能。预计 API 会有重大更改。

分布式代理运行时有助于跨进程边界的通信和代理生命周期管理。它由一个宿主服务和一个或多个工作运行时组成。

宿主服务维护与所有活动工作运行时的连接，促进消息传递，并为所有直接消息（即 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)

worker1: Hello x 1
worker2: Hello x 1
worker2: Hello x 2
worker1: Hello x 2
worker1: Hello x 3
worker2: Hello x 3
worker2: Hello x 4
worker1: Hello x 4
worker1: Hello x 5
worker2: Hello x 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()

跨语言运行时

上述过程基本相同，但是所有消息类型必须对所有跨代理消息类型使用共享的 protobuf 模式。

下一步

要查看使用分布式运行时的完整示例，请查看以下示例

• 分布式工作者

• 分布式语义路由器

• 分布式群聊