Skip to content

快速入门:通过 A2A 暴露远程智能体

Supported in ADKPythonExperimental

本快速入门涵盖了任何开发者最常见的起点:“如何暴露我的 ADK 智能体,以便其他智能体可以通过 A2A 使用它?”。这对于构建复杂的多智能体系统至关重要,其中不同的智能体需要协作和交互。

概述

本示例演示了智能体开发工具包(ADK)中的智能体到智能体(A2A)架构,展示了多个智能体如何协同工作来处理复杂任务。该示例实现了一个可以滚动骰子并检查数字是否为质数的智能体系统。

┌─────────────────┐    ┌──────────────────┐    ┌────────────────────┐
│   根智能体       │───▶│   滚动智能体     │    │   远程质数         │
│  (本地)         │    │   (本地)         │    │   智能体           │
│                 │    │                  │    │  (localhost:8001)  │
│                 │───▶│                  │◀───│                    │
└─────────────────┘    └──────────────────┘    └────────────────────┘

A2A 基础示例包含:

  • 根智能体root_agent):将任务委托给专门子智能体的主协调器
  • 滚动智能体roll_agent):处理骰子滚动操作的本地子智能体
  • 质数智能体prime_agent):检查数字是否为质数的远程 A2A 智能体,此智能体在单独的 A2A 服务器上运行

使用 ADK 服务器暴露你的智能体

ADK 提供了一个内置的 CLI 命令 adk api_server --a2a 来使用 A2A 协议暴露你的智能体。

a2a_basic 示例中,你首先需要通过 A2A 服务器暴露 check_prime_agent,以便本地根智能体可以使用它。

1. 使用代码暴露智能体

from google.adk.a2a.utils.agent_to_a2a import to_a2a

# 使你的智能体兼容 A2A
a2a_app = to_a2a(root_agent, port=8001)

to_a2a() 函数甚至会在幕后从 ADK 智能体中提取技能、能力和元数据,自动在内存中生成智能体卡片。这样,当使用 uvicorn 提供智能体端点服务时,标准的智能体卡片即可生效。

你也可以通过 agent_card 参数提供自己的智能体卡片。该值可以是一个 AgentCard 对象或指向智能体卡片 JSON 文件的路径。

使用 AgentCard 对象的示例:

from google.adk.a2a.utils.agent_to_a2a import to_a2a
from a2a.types import AgentCard

# 定义 A2A 智能体卡片
my_agent_card = AgentCard(
    name="file_agent",
    url="http://example.com",
    description="来自文件的测试智能体",
    version="1.0.0",
    capabilities={},
    skills=[],
    default_input_modes=["text/plain"],
    default_output_modes=["text/plain"],
    supports_authenticated_extended_card=False,
)
a2a_app = to_a2a(root_agent, port=8001, agent_card=my_agent_card)

使用 JSON 文件路径的示例:

from google.adk.a2a.utils.agent_to_a2a import to_a2a

# 从文件加载 A2A 智能体卡片
a2a_app = to_a2a(root_agent, port=8001, agent_card="/path/to/your/agent-card.json")

深入了解:to_a2a() 方法

当你调用 to_a2a() 时,ADK 会自动 handle 许多设置步骤:

  • A2aAgentExecutor 设置: 初始化一个 A2aAgentExecutor 作为 A2A 协议与你的 ADK 智能体之间的桥梁。如果你没有提供自定义的 Runner,它会自动创建一个默认的 Runner,该 Runner 由内存服务支持(用于素材、会话、记忆和凭据)。
  • 状态管理: 它创建一个 InMemoryTaskStore 来追踪 A2A 任务,并创建一个 InMemoryPushNotificationConfigStore 来处理推送通知。
  • 请求处理: 会创建一个 DefaultRequestHandler 来将传入的 A2A HTTP 请求路由到 A2aAgentExecutor 和状态存储。
  • Starlette 应用与智能体卡片: 它创建一个 Starlette 应用程序。在应用程序启动阶段,它会加载你提供的智能体卡片,或者使用 AgentCardBuilder 从你的智能体配置中自动构建一个。它还会挂载所有必要的 A2A API 路由。

现在让我们深入了解示例代码。

1. 获取示例代码

首先,确保你已经安装了必要的依赖项:

pip install google-adk[a2a]

你可以克隆并导航到 a2a_basic 示例

git clone https://github.com/google/adk-python.git

正如你将看到的,文件夹结构如下:

a2a_basic/
├── remote_a2a/
│   └── check_prime_agent/
│       ├── __init__.py
│       ├── agent.json
│       └── agent.py
├── README.md
├── __init__.py
└── agent.py # 本地根智能体

主智能体 (a2a_basic/agent.py)

  • roll_die(sides: int):用于滚动骰子的函数工具
  • roll_agent:专门从事骰子滚动的本地智能体
  • prime_agent:远程 A2A 智能体配置
  • root_agent:具有委托逻辑的主协调器

远程质数检查智能体 (a2a_basic/remote_a2a/check_prime_agent/)

  • agent.py:质数检查服务的实现
  • agent.json:A2A 智能体的智能体卡片
  • check_prime(nums: list[int]):质数检查算法

2. 启动远程质数智能体服务器

为了展示你的 ADK 智能体如何通过 A2A 消费 (Consuming) 远程智能体,你首先需要启动一个服务器,它将托管质数智能体(位于 check_prime_agent 目录下)。

# 启动 A2A 服务器,在 8001 端口上提供 check_prime_agent 服务
adk api_server --a2a --port 8001 contributing/samples/a2a_basic/remote_a2a
使用 --log_level debug 查看详细日志

要启用调试级别的日志记录,你可以在 adk api_server 命令中添加 --log_level debug。这将在测试智能体时提供更丰富的调试信息。

为什么使用 8001 端口?

在本地测试时,暴露智能体(远程质数智能体)的 A2A 服务器端口必须与消费智能体的端口不同。默认情况下,adk web 的端口是 8000,因此我们将 A2A 服务器设置为 8001 端口。

执行后,你应该看到类似以下内容:

INFO:     Started server process [56558]
INFO:     Waiting for application startup.
INFO:     Application startup complete.
INFO:     Uvicorn running on http://127.0.0.1:8001 (Press CTRL+C to quit)

3. 查看远程智能体所需的智能体卡片 (agent-card.json)

A2A 协议要求每个智能体必须有一个描述其功能的智能体卡片。

如果你想要在你的智能体中消费 (Consuming) 远程 A2A 智能体,你应该确认对方提供了一个智能体卡片(agent-card.json)。

在示例中,check_prime_agent 已经提供了一个智能体卡片:

a2a_basic/remote_a2a/check_prime_agent/agent-card.json
{
  "capabilities": {},
  "defaultInputModes": ["text/plain"],
  "defaultOutputModes": ["application/json"],
  "description": "专门用于检查数字是否为质数的智能体。它可以高效地确定单个数字或列表的质数属性。",
  "name": "check_prime_agent",
  "skills": [
    {
      "id": "prime_checking",
      "name": "Prime Number Checking",
      "description": "使用高效的数学算法检查列表中的数字是否为质数",
      "tags": ["mathematical", "computation", "prime", "numbers"]
    }
  ],
  "url": "http://localhost:8001/a2a/check_prime_agent",
  "version": "1.0.0"
}

4. 运行主(消费)智能体

# 在另一个终端中运行 adk web 服务器
adk web contributing/samples/

工作原理

主智能体使用 RemoteA2aAgent() 来消费远程智能体。

a2a_basic/agent.py
from google.adk.agents.remote_a2a_agent import AGENT_CARD_WELL_KNOWN_PATH
from google.adk.agents.remote_a2a_agent import RemoteA2aAgent

# 配置远程 A2A 智能体
prime_agent = RemoteA2aAgent(
    name="prime_agent",
    description="处理质数检查任务的智能体。",
    agent_card=(
        f"http://localhost:8001/a2a/check_prime_agent{AGENT_CARD_WELL_KNOWN_PATH}"
    ),
    use_legacy=False,
)

注意

设置 use_legacy=False 会启用 A2A 扩展

随后,你只需将 prime_agent 注册为 root_agent 的子智能体即可:

root_agent = Agent(
    model="gemini-2.0-flash",
    name="root_agent",
    instruction="...",
    sub_agents=[roll_agent, prime_agent],
)

高级配置:自定义转换器与拦截器

在内部,RemoteA2aAgent 在 A2A 协议格式与 ADK 的原生 Event 系统之间执行转换。你可以通过 config 参数传入 A2aRemoteAgentConfig 来自定义此行为。

转换器

  • a2a_message_converter:转换 A2A 消息。
  • a2a_task_converter:转换 A2A 任务。
  • a2a_part_converter:底层 Parts 转换。

请求拦截器

  • before_request:请求处理前执行。
  • after_request:请求处理后执行。

交互示例

一旦你的主智能体和远程智能体都在运行,你就可以观察到它们之间的 A2A 调用:

质数检查:

用户:7 是质数吗?
机器人:是的,7 是质数。

组合操作:

用户:滚动一个 10 面骰子并检查它是否是质数
机器人:我为你滚动了一个 8。
机器人:8 不是质数。

下一步