代理与代理运行时#

在本节及后续章节中，我们将聚焦AutoGen的核心概念：代理、代理运行时、消息与通信—— 这些是构建多代理应用的基石组件。

备注

核心API设计秉持无预设立场且高度灵活的原则，因此有时您可能会觉得颇具挑战性。如果您正在构建一个交互式、可扩展且分布式的多代理系统，并希望完全掌控所有工作流，请继续阅读。若您仅希望快速实现功能，可参阅AgentChat API。

AutoGen中的代理是由基础接口Agent定义的实体。它具有AgentId类型的唯一标识符，以及AgentMetadata类型的元数据字典。

大多数情况下，您可以从更高层级的RoutedAgent类派生子类，通过message_handler()装饰器和message变量的正确类型提示，将消息路由到指定的消息处理器。代理运行时是AutoGen中代理的执行环境。

类似于编程语言的运行时环境，代理运行时提供了必要的基础设施来促进代理间通信、管理代理生命周期、实施安全边界，并支持监控与调试。

对于本地开发，开发者可使用可嵌入Python应用的 SingleThreadedAgentRuntime。

备注

代理并非由应用代码直接实例化和管理，而是由运行时按需创建并托管。

如果您已熟悉AgentChat，需特别注意AgentChat的代理（如AssistantAgent）由应用直接创建，因此不受运行时直接管理。要在Core中使用AgentChat代理，您需要创建一个包装器Core代理来将消息委托给AgentChat代理，并由运行时管理该包装器代理。

实现一个代理#

要实现代理，开发者必须继承RoutedAgent类，并使用message_handler()装饰器为代理需要处理的每种消息类型实现对应的消息处理方法。例如，以下代理处理简单的MyMessageType消息类型，并打印接收到的消息：

from dataclasses import dataclass

from autogen_core import AgentId, MessageContext, RoutedAgent, message_handler


@dataclass
class MyMessageType:
    content: str


class MyAgent(RoutedAgent):
    def __init__(self) -> None:
        super().__init__("MyAgent")

    @message_handler
    async def handle_my_message_type(self, message: MyMessageType, ctx: MessageContext) -> None:
        print(f"{self.id.type} received message: {message.content}")

该代理仅处理MyMessageType类型消息，消息会被传递到handle_my_message_type方法。开发者可以通过使用message_handler()装饰器并为处理器函数中的message变量设置类型提示，来为不同消息类型定义多个消息处理器。如果更符合代理逻辑，您还可以在单个消息处理器函数中为message变量使用python类型联合。详见下一节关于消息与通信的内容。

使用AgentChat代理#

如果您拥有AgentChat代理并希望在Core API中使用它，可以创建一个包装器RoutedAgent来将消息委托给AgentChat代理。以下示例展示了如何为AgentChat中的AssistantAgent 创建包装器代理。

from autogen_agentchat.agents import AssistantAgent
from autogen_agentchat.messages import TextMessage
from autogen_ext.models.openai import OpenAIChatCompletionClient


class MyAssistant(RoutedAgent):
    def __init__(self, name: str) -> None:
        super().__init__(name)
        model_client = OpenAIChatCompletionClient(model="gpt-4o")
        self._delegate = AssistantAgent(name, model_client=model_client)

    @message_handler
    async def handle_my_message_type(self, message: MyMessageType, ctx: MessageContext) -> None:
        print(f"{self.id.type} received message: {message.content}")
        response = await self._delegate.on_messages(
            [TextMessage(content=message.content, source="user")], ctx.cancellation_token
        )
        print(f"{self.id.type} responded: {response.chat_message}")

关于如何使用模型客户端，请参阅模型客户端章节。

由于核心API是无预设立场的，您不需要使用AgentChat API来调用核心API。您可以实现自己的代理或使用其他代理框架。

注册代理类型#

为了使代理在运行时可用，开发者可以使用 register()类方法，该方法属于 BaseAgent类。注册过程会将一个由字符串唯一标识的代理类型，与一个工厂函数关联起来，该函数用于创建给定类的代理类型实例。工厂函数的作用是允许在需要时自动创建代理实例。

代理类型(AgentType)不同于代理类。在本例中，代理类型是AgentType("my_agent")或AgentType("my_assistant")，而代理类是Python类MyAgent或MyAssistantAgent。工厂函数预期会返回一个代理类的实例，在该实例上调用register()类方法。阅读代理身份与生命周期以了解更多关于代理类型和身份的信息。

备注

不同的代理类型可以注册为返回相同代理类的工厂函数。例如，在工厂函数中，可以使用构造函数参数的变化来创建同一代理类的不同实例。

要向 SingleThreadedAgentRuntime注册我们的代理类型，可以使用以下代码：

from autogen_core import SingleThreadedAgentRuntime

runtime = SingleThreadedAgentRuntime()
await MyAgent.register(runtime, "my_agent", lambda: MyAgent())
await MyAssistant.register(runtime, "my_assistant", lambda: MyAssistant("my_assistant"))

AgentType(type='my_assistant')

一旦代理类型被注册，我们就可以使用AgentId 向代理实例发送直接消息。运行时将在首次向该实例传递消息时创建该实例。

runtime.start()  # 在后台开始处理消息。
await runtime.send_message(MyMessageType("Hello, World!"), AgentId("my_agent", "default"))
await runtime.send_message(MyMessageType("Hello, World!"), AgentId("my_assistant", "default"))
await runtime.stop()  # 停止在后台处理消息。

my_agent received message: Hello, World!
my_assistant received message: Hello, World!
my_assistant responded: Hello! How can I assist you today?

备注

由于运行时管理着代理的生命周期，AgentId 仅用于与代理通信或检索其元数据（例如描述）。

运行单线程代理运行时#

上述代码片段使用 start() 启动后台任务来处理消息并将其传递给接收者的消息处理器。这是本地嵌入式运行时 SingleThreadedAgentRuntime 的一个特性。

要立即停止后台任务，请使用 stop() 方法：

runtime.start()
# ... 发送消息、发布消息等
await runtime.stop()  # 这会立即返回但不会取消
# 任何正在进行的消息处理。

您可以通过再次调用 start() 来恢复后台任务。

对于批量场景（如运行基准测试来评估代理），您可能希望等待后台任务在以下情况下自动停止：当没有未处理消息且没有代理正在处理消息时—— 可以认为批次已完成。您可以通过使用 stop_when_idle() 方法实现这一点：

runtime.start()
# ... 发送消息、发布消息等操作
await runtime.stop_when_idle()  # 这将阻塞直到运行时处于空闲状态。

要关闭运行时并释放资源，请使用 close() 方法：

await runtime.close()

其他运行时实现会有各自运行运行时的方式。