{ "cells": [ { "cell_type": "markdown", "metadata": {}, "source": [ "# 代理与代理运行时\n\n在本节及后续章节中，我们将聚焦AutoGen的核心概念：\n代理、代理运行时、消息与通信——\n这些是构建多代理应用的基石组件。\n\n```{note}\n核心API设计秉持无预设立场且高度灵活的原则，因此有时您可能会觉得颇具挑战性。\n如果您正在构建一个交互式、可扩展且分布式的多代理系统，并希望完全掌控所有工作流，请继续阅读。\n若您仅希望快速实现功能，可参阅[AgentChat API](../../agentchat-user-guide/index.md)。\n```\n" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "AutoGen中的代理是由基础接口{py:class}`~autogen_core.Agent`定义的实体。\n它具有{py:class}`~autogen_core.AgentId`类型的唯一标识符，\n以及{py:class}`~autogen_core.AgentMetadata`类型的元数据字典。\n\n大多数情况下，您可以从更高层级的{py:class}`~autogen_core.RoutedAgent`类派生子类，\n通过{py:meth}`~autogen_core.message_handler`装饰器和`message`变量的正确类型提示，\n将消息路由到指定的消息处理器。\n代理运行时是AutoGen中代理的执行环境。\n\n类似于编程语言的运行时环境，\n代理运行时提供了必要的基础设施来促进代理间通信、\n管理代理生命周期、实施安全边界，并支持监控与调试。\n\n对于本地开发，开发者可使用可嵌入Python应用的\n{py:class}`~autogen_core.SingleThreadedAgentRuntime`。\n\n```{note}\n代理并非由应用代码直接实例化和管理，\n而是由运行时按需创建并托管。\n\n如果您已熟悉[AgentChat](../../agentchat-user-guide/index.md)，\n需特别注意AgentChat的代理（如{py:class}`~autogen_agentchat.agents.AssistantAgent`）\n由应用直接创建，因此不受运行时直接管理。要在Core中使用AgentChat代理，\n您需要创建一个包装器Core代理来将消息委托给AgentChat代理，\n并由运行时管理该包装器代理。\n```\n" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## 实现一个代理\n\n要实现代理，开发者必须继承{py:class}`~autogen_core.RoutedAgent`类，\n并使用{py:meth}`~autogen_core.message_handler`装饰器为代理需要处理的每种消息类型\n实现对应的消息处理方法。\n例如，\n以下代理处理简单的`MyMessageType`消息类型，并打印接收到的消息：\n" ] }, { "cell_type": "code", "execution_count": 11, "metadata": {}, "outputs": [], "source": [ "from dataclasses import dataclass\n", "\n", "from autogen_core import AgentId, MessageContext, RoutedAgent, message_handler\n", "\n", "\n", "@dataclass\n", "class MyMessageType:\n", " content: str\n", "\n", "\n", "class MyAgent(RoutedAgent):\n", " def __init__(self) -> None:\n", " super().__init__(\"MyAgent\")\n", "\n", " @message_handler\n", " async def handle_my_message_type(self, message: MyMessageType, ctx: MessageContext) -> None:\n", " print(f\"{self.id.type} received message: {message.content}\")" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "该代理仅处理`MyMessageType`类型消息，消息会被传递到`handle_my_message_type`方法。开发者可以通过使用{py:meth}`~autogen_core.message_handler`装饰器并为处理器函数中的`message`变量设置类型提示，来为不同消息类型定义多个消息处理器。如果更符合代理逻辑，您还可以在单个消息处理器函数中为`message`变量使用[python类型联合](https://docs.python.org/3/library/typing.html#typing.Union)。\n详见下一节关于[消息与通信](./message-and-communication.ipynb)的内容。\n" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## 使用AgentChat代理\n\n如果您拥有[AgentChat](../../agentchat-user-guide/index.md)代理并希望在Core API中使用它，\n可以创建一个包装器{py:class}`~autogen_core.RoutedAgent`来将消息委托给AgentChat代理。\n以下示例展示了如何为AgentChat中的{py:class}`~autogen_agentchat.agents.AssistantAgent`\n创建包装器代理。\n" ] }, { "cell_type": "code", "execution_count": 12, "metadata": {}, "outputs": [], "source": [ "from autogen_agentchat.agents import AssistantAgent\n", "from autogen_agentchat.messages import TextMessage\n", "from autogen_ext.models.openai import OpenAIChatCompletionClient\n", "\n", "\n", "class MyAssistant(RoutedAgent):\n", " def __init__(self, name: str) -> None:\n", " super().__init__(name)\n", " model_client = OpenAIChatCompletionClient(model=\"gpt-4o\")\n", " self._delegate = AssistantAgent(name, model_client=model_client)\n", "\n", " @message_handler\n", " async def handle_my_message_type(self, message: MyMessageType, ctx: MessageContext) -> None:\n", " print(f\"{self.id.type} received message: {message.content}\")\n", " response = await self._delegate.on_messages(\n", " [TextMessage(content=message.content, source=\"user\")], ctx.cancellation_token\n", " )\n", " print(f\"{self.id.type} responded: {response.chat_message}\")" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "关于如何使用模型客户端，请参阅[模型客户端](../components/model-clients.ipynb)章节。\n\n由于核心API是无预设立场的，\n您不需要使用AgentChat API来调用核心API。\n您可以实现自己的代理或使用其他代理框架。\n" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## 注册代理类型\n\n为了使代理在运行时可用，开发者可以使用\n{py:meth}`~autogen_core.BaseAgent.register`类方法，该方法属于\n{py:class}`~autogen_core.BaseAgent`类。\n注册过程会将一个由字符串唯一标识的代理类型，\n与一个工厂函数关联起来，\n该函数用于创建给定类的代理类型实例。\n工厂函数的作用是允许在需要时自动创建代理实例。\n\n代理类型({py:class}`~autogen_core.AgentType`)不同于代理类。在本例中，\n代理类型是`AgentType(\"my_agent\")`或`AgentType(\"my_assistant\")`，而代理类是Python类`MyAgent`或`MyAssistantAgent`。\n工厂函数预期会返回一个代理类的实例，\n在该实例上调用{py:meth}`~autogen_core.BaseAgent.register`类方法。\n阅读[代理身份与生命周期](../core-concepts/agent-identity-and-lifecycle.md)\n以了解更多关于代理类型和身份的信息。\n\n```{note}\n不同的代理类型可以注册为返回相同代理类的工厂函数。例如，在工厂函数中，\n可以使用构造函数参数的变化\n来创建同一代理类的不同实例。\n```\n\n要向\n{py:class}`~autogen_core.SingleThreadedAgentRuntime`注册我们的代理类型，\n可以使用以下代码：\n" ] }, { "cell_type": "code", "execution_count": 13, "metadata": {}, "outputs": [ { "data": { "text/plain": [ "AgentType(type='my_assistant')" ] }, "execution_count": 13, "metadata": {}, "output_type": "execute_result" } ], "source": [ "from autogen_core import SingleThreadedAgentRuntime\n", "\n", "runtime = SingleThreadedAgentRuntime()\n", "await MyAgent.register(runtime, \"my_agent\", lambda: MyAgent())\n", "await MyAssistant.register(runtime, \"my_assistant\", lambda: MyAssistant(\"my_assistant\"))" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "一旦代理类型被注册，我们就可以使用{py:class}`~autogen_core.AgentId`\n向代理实例发送直接消息。\n运行时将在首次向该实例传递消息时创建该实例。\n" ] }, { "cell_type": "code", "execution_count": 14, "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "my_agent received message: Hello, World!\n", "my_assistant received message: Hello, World!\n", "my_assistant responded: Hello! How can I assist you today?\n" ] } ], "source": [ "runtime.start() # 在后台开始处理消息。\n", "await runtime.send_message(MyMessageType(\"Hello, World!\"), AgentId(\"my_agent\", \"default\"))\n", "await runtime.send_message(MyMessageType(\"Hello, World!\"), AgentId(\"my_assistant\", \"default\"))\n", "await runtime.stop() # 停止在后台处理消息。" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "```{note}\n由于运行时管理着代理的生命周期，{py:class}`~autogen_core.AgentId`\n仅用于与代理通信或检索其元数据（例如描述）。\n```\n" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## 运行单线程代理运行时\n\n上述代码片段使用 {py:meth}`~autogen_core.SingleThreadedAgentRuntime.start` 启动后台任务\n来处理消息并将其传递给接收者的消息处理器。\n这是本地嵌入式运行时 {py:class}`~autogen_core.SingleThreadedAgentRuntime` 的一个特性。\n\n要立即停止后台任务，请使用 {py:meth}`~autogen_core.SingleThreadedAgentRuntime.stop` 方法：\n" ] }, { "cell_type": "code", "execution_count": 15, "metadata": {}, "outputs": [], "source": [ "runtime.start()\n", "# ... 发送消息、发布消息等\n", "await runtime.stop() # 这会立即返回但不会取消\n", "# 任何正在进行的消息处理。" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "您可以通过再次调用 {py:meth}`~autogen_core.SingleThreadedAgentRuntime.start` 来恢复后台任务。\n\n对于批量场景（如运行基准测试来评估代理），\n您可能希望等待后台任务在以下情况下自动停止：\n当没有未处理消息且没有代理正在处理消息时——\n可以认为批次已完成。\n您可以通过使用 {py:meth}`~autogen_core.SingleThreadedAgentRuntime.stop_when_idle` 方法实现这一点：\n" ] }, { "cell_type": "code", "execution_count": 16, "metadata": {}, "outputs": [], "source": [ "runtime.start()\n", "# ... 发送消息、发布消息等操作\n", "await runtime.stop_when_idle() # 这将阻塞直到运行时处于空闲状态。" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "要关闭运行时并释放资源，请使用 {py:meth}`~autogen_core.SingleThreadedAgentRuntime.close` 方法：\n" ] }, { "cell_type": "code", "execution_count": 17, "metadata": {}, "outputs": [], "source": [ "await runtime.close()" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "其他运行时实现会有各自运行运行时的方式。\n" ] } ], "metadata": { "kernelspec": { "display_name": ".venv", "language": "python", "name": "python3" }, "language_info": { "codemirror_mode": { "name": "ipython", "version": 3 }, "file_extension": ".py", "mimetype": "text/x-python", "name": "python", "nbconvert_exporter": "python", "pygments_lexer": "ipython3", "version": "3.12.7" } }, "nbformat": 4, "nbformat_minor": 2 }