Agent 和 Agent 运行时#

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

注意

核心 API 旨在保持中立和灵活性。因此，有时您可能会觉得它具有挑战性。如果您正在构建交互式、可扩展和分布式的多 agent 系统，并且想要完全控制所有工作流程，请继续。如果您只想快速运行某些内容，您可以查看 AgentChat API。

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

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

与编程语言的运行时环境类似，agent 运行时提供必要的 инфраструктура 以促进 agent 之间的通信、管理 agent 生命周期、强制执行安全边界以及支持监视和调试。

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

注意

Agent 不是由应用程序代码直接实例化和管理的。而是由运行时在需要时创建并由运行时管理。

如果您已经熟悉 AgentChat，请务必注意，AgentChat 的 agent（例如 AssistantAgent）由应用程序创建，因此不由运行时直接管理。要在 Core 中使用 AgentChat agent，您需要创建一个包装 Core agent，该 agent 将消息委托给 AgentChat agent，并让运行时管理该包装 agent。

实现 Agent#

要实现一个 agent，开发人员必须子类化 RoutedAgent 类，并为 agent 预期处理的每种消息类型实现一个消息处理方法，使用 message_handler() 装饰器。例如，以下 agent 处理一个简单的消息类型 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}")

此 agent 仅处理 MyMessageType，并且消息将传递到 handle_my_message_type 方法。开发人员可以通过使用 message_handler() 装饰器并为处理程序函数中的 message 变量设置类型提示来为不同的消息类型设置多个消息处理程序。如果更适合 agent 的逻辑，您还可以利用 python 类型联合来处理一个消息处理程序函数中的 message 变量。请参阅下一节中的消息和通信。

使用 AgentChat Agent#

如果您有 AgentChat agent 并且想在 Core API 中使用它，您可以创建一个包装 RoutedAgent，该 agent 将消息委托给 AgentChat agent。以下示例显示了如何为 AgentChat 中的 AssistantAgent 创建一个包装 agent。

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}")

有关如何使用模型客户端的信息，请参阅模型客户端部分。

由于 Core API 是中立的，因此您不需要使用 AgentChat API 即可使用 Core API。您可以实现自己的 agent 或使用另一个 agent 框架。

注册 Agent 类型#

为了使 agent 可用于运行时，开发人员可以使用 register() 类方法从 BaseAgent 类。注册过程将 agent 类型（由字符串唯一标识）与工厂函数相关联，该工厂函数创建给定类的 agent 类型的实例。工厂函数用于允许在需要时自动创建 agent 实例。

Agent 类型 (AgentType) 与 agent 类不同。在此示例中，agent 类型为 AgentType("my_agent") 或 AgentType("my_assistant")，agent 类为 Python 类 MyAgent 或 MyAssistantAgent。工厂函数应返回在其上调用 register() 类方法的 agent 类的实例。阅读 Agent 身份和生命周期以了解有关 agent 类型和身份的更多信息。

注意

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

要使用 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()

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