代理和代理运行时

在本节和下一节中，我们将重点介绍 AutoGen 的核心概念：代理、代理运行时、消息和通信——多代理应用程序的基础构建块。

注意

核心 API 被设计为无偏见且灵活的。因此有时，您可能会觉得它具有挑战性。如果您正在构建一个交互式、可扩展和分布式多代理系统，并希望完全控制所有工作流，请继续。如果您只想快速运行某些东西，可以查看AgentChat API。

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

在大多数情况下，您可以从更高级别的类RoutedAgent中子类化您的代理，这使您能够将消息路由到使用message_handler()装饰器和message变量的正确类型提示指定的消息处理程序。代理运行时是 AutoGen 中代理的执行环境。

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

对于本地开发，开发人员可以使用SingleThreadedAgentRuntime，它可以嵌入到 Python 应用程序中。

注意

代理不由应用程序代码直接实例化和管理。相反，它们在需要时由运行时创建并由运行时管理。

如果您已经熟悉AgentChat，需要注意的是，AgentChat 的代理，例如AssistantAgent是由应用程序创建的，因此不由运行时直接管理。要在核心中使用 AgentChat 代理，您需要创建一个封装 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变量设置类型提示来为不同的消息类型设置多个消息处理程序。如果更适合代理的逻辑，您还可以利用python typing union来处理一个消息处理程序函数中的message变量。请参阅下一节关于消息和通信的内容。

使用 AgentChat 代理

如果您有一个AgentChat代理并希望在核心 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()  # Start processing messages in the background.
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()  # Stop processing messages in the background.

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()
# ... Send messages, publish messages, etc.
await runtime.stop()  # This will return immediately but will not cancel
# any in-progress message handling.

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

对于批处理场景，例如运行基准测试以评估代理，您可能希望在没有未处理消息且没有代理正在处理消息时自动停止后台任务——批处理可能被认为是完整的。您可以使用stop_when_idle()方法实现此目的

runtime.start()
# ... Send messages, publish messages, etc.
await runtime.stop_when_idle()  # This will block until the runtime is idle.

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

await runtime.close()

其他运行时实现将有自己的运行方式。