代理和代理运行环境#

在本节及接下来的部分中，我们将重点介绍 AutoGen 的核心概念：代理、代理运行时、消息和通信 – 这些是多代理应用程序的基础构建模块。

注意

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

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

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

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

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

注意

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

如果你已经熟悉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变量设置类型提示，为不同的消息类型设置多个消息处理程序。如果更适合代理的逻辑，您还可以在一个消息处理函数中利用python typing union来处理message变量。请参阅下一节关于消息和通信的内容。

使用 AgentChat 代理#

如果你有一个AgentChat代理并希望在Core API中使用它，你可以创建一个封装器RoutedAgent，将消息委托给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.content}")

关于如何使用模型客户端，请参见模型客户端部分。

由于Core API是无偏见的，使用Core API并不强制要求使用AgentChat API。你可以实现自己的代理或使用其他代理框架。

注册代理类型#

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

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

注意

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

要将我们的代理类型注册到 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()

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