代理和代理运行时#

在本节和下一节中，我们关注 AutoGen 的核心概念：代理、代理运行时、消息和通信。这里您不会找到任何 AI 模型或工具，只有构建多代理应用程序的基础构建块。

AutoGen 中的代理是由基类 autogen_core.base.BaseAgent 定义的实体。它具有类型为 autogen_core.base.AgentId 的唯一标识符，类型为 autogen_core.base.AgentMetadata 的元数据字典，以及用于处理消息的方法autogen_core.base.BaseAgent.on_message()。

代理运行时是 AutoGen 中代理的执行环境。类似于编程语言的运行时环境，代理运行时提供必要的基础设施来促进代理之间的通信，管理代理生命周期，执行安全边界，并支持监控和调试。对于本地开发，开发人员可以使用 SingleThreadedAgentRuntime，它可以嵌入到 Python 应用程序中。

备注

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

实现代理#

要实现代理，开发人员必须继承 BaseAgent 类并实现 on_message() 方法。当代理收到消息时，会调用此方法。例如，以下代理处理一个简单的消息类型并打印它收到的消息：

from dataclasses import dataclass

from autogen_core.base import AgentId, BaseAgent, MessageContext


@dataclass
class MyMessageType:
    content: str


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

    async def on_message(self, message: MyMessageType, ctx: MessageContext) -> None:
        print(f"Received message: {message.content}")  # type: ignore

这个代理只处理 MyMessageType 消息。要处理多种消息类型，开发人员可以继承 RoutedAgent 类，它提供了一个易于使用的 API 来为不同的消息类型实现不同的消息处理程序。请参阅下一节message and communication。

注册代理类型#

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

Agent type（AgentType）与Agent class不同。在这个例子中，代理类型是 AgentType("my_agent")，代理类是 Python 类 MyAgent。工厂函数预期返回调用 register() 类方法的代理类的实例。阅读Agent Identity and Lifecycles以了解更多关于代理类型和身份的信息。

备注

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

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

from autogen_core.application import SingleThreadedAgentRuntime

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

AgentType(type='my_agent')

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

agent_id = AgentId("my_agent", "default")
runtime.start()  # Start processing messages in the background.
await runtime.send_message(MyMessageType("Hello, World!"), agent_id)
await runtime.stop()  # Stop processing messages in the background.

Received message: Hello, World!

备注

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

运行单线程代理运行时#

上面的代码片段使用 runtime.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.

您也可以使用以下方法直接逐个处理消息，而无需后台任务：

await runtime.process_next()

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