构建一个使用 Google ADK、A2A 和 Cloud Run 的多代理深度研究工具

发布: 1周前 (2025年12月30日 GMT+8 11:29)

9 min read

原文: Dev.to

Source: Dev.to

介绍

Research 是一个含义丰富的词。它不仅仅是 Googling 一个关键词。它包括阅读论文、核实事实、寻找那张完美的图表，并把所有信息合成为一个连贯的整体。

让单个 AI 代理顺序完成所有这些工作并不高效。它们会产生幻觉、卡住，且速度肯定很慢。

TL;DR: 想要代码吗？查看 Deep Research Agent code on GitHub。

我想要一个系统，能够接受一个主题——比如“循环神经网络的历史”——并生成一份全面且带插图的报告。同时，我也想学习如何从零构建一个 Deep Research Tool。

第一次尝试？一个单一循环。它先进行研究，然后寻找图片，接着检查自己的工作。耗时极长。

于是我问：

我能让它更快吗？

在本文中，我们将构建一个 Parallel Research Squad（并行研究小队）。我们不再让单个代理完成所有任务，而是启动三个专门的代理同时运行，并由一个中心 Orchestrator（编排器）进行协调。我们将使用：

Google’s Agent Development Kit (ADK) 作为“大脑”
Agent‑to‑Agent (A2A) Protocol 进行通信
Google Cloud Run 实现无限扩展

第 1 部分 – 代理设计模式

我们已经不再只是编写提示词；我们在进行 系统工程。为了构建一个稳健的系统，我们采用了三种关键设计模式：

1. 编排者模式

我们不使用一个决定一切的 “上帝代理”，而是拥有一个中心 编排者——可以把它看作总编辑。它不直接撰写文章；它把报道任务分配给记者，管理状态，处理错误，并确保最终产品按时交付。

2. 并行化

这是我们的提速技巧。大多数代理框架都是顺序执行的（步骤 A → 步骤 B → 步骤 C）。但 “阅读 ArXiv 论文” 与 “搜索图片” 是相互独立的任务。通过并行运行，它们的总延迟被压缩为最慢那个任务的时长，而不是所有任务时长的累加。

3. 评估‑优化器

我们不相信第一稿。系统中包含一个评审代理。编排者将研究成果发送给评审，评审返回严格的通过/未通过评分并提供反馈。如果未通过，编排者会回环（优化器）以弥补缺陷。

第 2 部分 – 速度需求（并行执行）

AI 代理最大的瓶颈是延迟。等待模型“思考”以及浏览网页会耗费时间。

使用 ADK 我们实现了 ParallelAgent。这不仅仅是一个概念；它是框架中的原语，帮助我们处理异步的复杂性。ParallelAgent 会并行运行，编排器会等所有代理完成后才继续。这是并行化彼此不依赖的代理的简便方式。

# orchestrator/app/agent.py
from google.adk.agents import ParallelAgent

# “小队”一起运行
research_squad = ParallelAgent(
    name="research_squad",
    description=(
        "并行运行 researcher、academic scholar 和 asset gatherer。"
    ),
    sub_agents=[researcher, academic_scholar, asset_gatherer],
)

这一次改动将我们的总处理时间缩短了 60 %。当 Scholar 正在阅读一篇密集的 PDF 时，Asset Gatherer 已经在验证图像 URL 了。

A2A 握手

Source: …

第 3 部分 – 通用语言（A2A 协议）

这些代理如何交流？它们是独立的微服务。Researcher 可能运行在高内存实例上，而 Orchestrator 则运行在一个小型实例上。

我们使用 Agent‑to‑Agent (A2A) 协议，这是一种基于 JSON‑RPC 的 AI 代理标准化 API。

为什么使用 A2A？

解耦 – Orchestrator 不需要了解 Researcher 的实现细节，只需知道它的位置。
互操作性 – 你可以用 Python 编写 Researcher，用 Go 编写 Judge。只要它们遵循 A2A，就能协同工作。
服务发现 – 在开发阶段我们将代理映射到 localhost 端口；在生产环境则映射到 Cloud Run URL（或其他服务网格）。

TL;DR 实现清单

定义子代理（researcher、academic_scholar、asset_gatherer、judge）。
将它们包装在 ParallelAgent 中（research_squad）。
创建 Orchestrator，其职责是：
- 将主题发送给 research_squad。
- 接收合并后的输出。
- 将结果传递给 judge。
- 如果 judge 评估失败，则触发 Optimizer 循环。
将每个代理部署为独立的 Cloud Run 服务，并公开 A2A 端点。
配置服务发现（环境变量或服务注册表）。

通过这些模式，你可以把慢速、单体的 “research bot” 转变为快速、可扩展的 Parallel Research Squad，可靠地产出高质量、带插图的报告。祝构建愉快！

# orchestrator/app/agent.py
from google.adk.agents.remote_a2a_agent import RemoteA2aAgent

# The Orchestrator calls the remote Scholar service
academic_scholar = RemoteA2aAgent(
    name="academic_scholar",
    # In prod, this is an internal Cloud Run URL
    agent_card="http://scholar-service:8000/.well-known/agent.json",
    description="Searches for academic papers."
)

Scaling Graph

第 4 部分 – 基础设施即超级能力（Cloud Run）

我们将在 Google Cloud Run 上部署此系统。这为我们提供了“杂货店”扩展模型。

“杂货店”模型

想象一家只有一个收银通道的杂货店。如果有 50 个人出现，排队会一直延伸到门外。

在我们的系统中，每个代理就是一个收银通道。

单体：一个通道。50 个请求 = 50× 等待时间。
在 Cloud Run 上的微服务：50 个请求 = Cloud Run 会立即启动 50 个 Researcher 实例。所有人同时完成结账。

零实例运行

当没有人使用应用时，我们的 实例数为 0。费用为 $0。这对成本效益高的 AI 应用至关重要。

注意： 当 Cloud Run 服务未被使用时，它会自动缩减至零实例，这意味着下一个请求需要一次冷启动。您可以通过使用健康检查来保持服务“热”。

第5部分 – 前端（Next.js + 实时）

我们不想要一个 CLI 工具；我们想要一个产品。

我们构建了一个 Next.js 前端，连接到 Orchestrator。因为我们了解架构，所以可以将其可视化。当 research_squad 启动时，我们的前端会并排显示三个脉冲指示灯。你实际上可以看到并行正在发生。

它营造出一种“活性”和透明感，提升用户信任。

结论

通过将我们的单体系统拆分为 Parallel Research Squad，我们构建了一个：

更快：并行执行将等待时间缩短了 >50%。
更好：专门的代理（Scholar、Gatherer）能够比单一的通才完成更深入的工作。
可扩展：在 Cloud Run 上的微服务可以处理无限的负载。

想自己动手构建吗？

构建一个使用 Google ADK、A2A 和 Cloud Run 的多代理深度研究工具

介绍