一、为什么你需要关注多 Agent 协同
单 Agent 的能力边界正在被快速触及。
一个典型的例子：你想让 AI 帮你做一次竞品分析。这需要先搜索公开资料、抓取网页、提取关键信息、对比分析、最后生成报告。如果只用一个 Agent，Prompt 会变得极其复杂，上下文窗口很快爆满，而且每一步的可靠性会随着链路拉长急剧下降。
Google 在 2025 年提出的 A2A（Agent-to-Agent）协议 正是为了解决这个问题。它的核心思路很简单：把复杂任务拆成多个专业 Agent，让它们通过标准协议相互通信、协作完成。
听起来很美好，但落地面临三个现实问题：
- 基础设施：多 Agent 意味着多个运行时，手动管理服务器不现实
- 模型接入：每个 Agent 可能需要不同的模型，统一接入成本高
- 状态管理：Agent 之间传数据、存中间结果，需要可靠的存储层
阿里云的 函数计算 AgentRun 恰好在这三个问题上给出了一个开箱即用的答案。它基于 Serverless 架构，内置百炼大模型集成，可选配 NAS 文件存储，而且原生支持 Google ADK 框架的多 Agent 协同。
本文带你在 AgentRun 上跑通一个完整的 A2A 多 Agent 工作流。不聊概念，直接上实操。
二、环境准备
2.1 开通服务
访问 AgentRun 控制台（通过链接可体验：AgentRun 控制台体验），完成以下准备：
- 开通函数计算 FC 服务
- 开通百炼模型服务（用于调用通义千问等大模型）
- （可选）开通文件存储 NAS，用于 Agent 间共享数据
2.2 本地开发环境
确保本地安装了以下工具：
# 安装 Serverless Devs 工具
npm install -g @serverless-devs/s
# 配置阿里云凭证
s config add --AccessKeyID <your-ak> --AccessKeySecret <your-sk>
三、场景设计：多 Agent 竞品分析系统
我们要搭建的系统包含三个专业 Agent：
┌──────────────┐ A2A ┌──────────────┐ A2A ┌──────────────┐
│ 搜索 Agent │ ────────→ │ 分析 Agent │ ────────→ │ 报告 Agent │
│ (资料收集) │ 搜索结果 │ (对比分析) │ 分析结果 │ (生成报告) │
└──────────────┘ └──────────────┘ └──────────────┘
│ │ │
└─────────────────────────┼─────────────────────────┘
│
┌──────┴──────┐
│ NAS 存储 │
│ (共享中间结果)│
└─────────────┘
每个 Agent 职责单一：
- 搜索 Agent：接收用户关键词，调用搜索引擎 API，返回结构化结果
- 分析 Agent：接收搜索结果，调用大模型做竞品对比分析
- 报告 Agent：接收分析结果，生成格式化的 Markdown 报告
四、Agent 开发
4.1 搜索 Agent
搜索 Agent 的核心逻辑：接收查询关键词，调用搜索 API，格式化输出。
# search_agent/main.py
from google.adk.agents import Agent
from google.adk.tools import ToolContext
import requests
import json
import os
def web_search(query: str, tool_context: ToolContext) -> str:
"""执行网络搜索并返回结构化结果"""
# 实际场景中使用搜索 API（如 Bing Search API）
# 这里展示核心逻辑结构
results = [
{
"title": f"关于"{
query}"的搜索结果",
"snippet": "模拟搜索摘要信息...",
"url": "https://example.com/result"
}
]
# 存入 NAS 共享路径，供下游 Agent 读取
output_path = "/mnt/nas/search_results.json"
os.makedirs(os.path.dirname(output_path), exist_ok=True)
with open(output_path, "w", encoding="utf-8") as f:
json.dump({
"query": query, "results": results}, f, ensure_ascii=False)
tool_context.actions.skip_summarization = True
return json.dumps(results, ensure_ascii=False, indent=2)
search_agent = Agent(
name="search_agent",
model="qwen-plus", # 百炼提供的通义千问模型
instruction="""你是一个专业的搜索 Agent。
当用户提出搜索需求时，调用 web_search 工具执行搜索。
然后将结果传递给下游的分析 Agent。""",
tools=[web_search],
)
4.2 分析 Agent
分析 Agent 读取搜索结果，调用大模型做深度对比分析。
# analysis_agent/main.py
from google.adk.agents import Agent
import json
import os
def load_search_results() -> str:
"""从 NAS 读取搜索 Agent 的结果"""
input_path = "/mnt/nas/search_results.json"
if not os.path.exists(input_path):
return "未找到搜索结果，请先运行搜索 Agent。"
with open(input_path, "r", encoding="utf-8") as f:
data = json.load(f)
return json.dumps(data, ensure_ascii=False, indent=2)
def save_analysis(analysis_text: str) -> str:
"""将分析结果存入 NAS"""
output_path = "/mnt/nas/analysis_result.md"
os.makedirs(os.path.dirname(output_path), exist_ok=True)
with open(output_path, "w", encoding="utf-8") as f:
f.write(analysis_text)
return "分析结果已保存"
analysis_agent = Agent(
name="analysis_agent",
model="qwen-plus",
instruction="""你是一个专业的竞品分析 Agent。
1. 读取搜索 Agent 返回的结果
2. 从产品定位、核心功能、定价策略、技术架构四个维度做对比分析
3. 输出结构化的分析结论
4. 将最终结果保存到 NAS 供报告 Agent 使用""",
tools=[load_search_results, save_analysis],
)
4.3 报告 Agent
# report_agent/main.py
from google.adk.agents import Agent
import os
def load_analysis() -> str:
"""读取分析 Agent 的输出"""
input_path = "/mnt/nas/analysis_result.md"
if not os.path.exists(input_path):
return "分析结果尚未生成"
with open(input_path, "r", encoding="utf-8") as f:
return f.read()
report_agent = Agent(
name="report_agent",
model="qwen-plus",
instruction="""你是一个专业的报告生成 Agent。
1. 读取分析 Agent 的输出
2. 生成一份格式规范的 Markdown 报告
3. 报告包含：摘要、详细对比、结论与建议三部分
4. 确保输出排版清晰，适合直接分享""",
tools=[load_analysis],
)
五、部署到 AgentRun
5.1 项目结构
multi-agent-analysis/
├── search_agent/
│ ├── main.py
│ └── requirements.txt
├── analysis_agent/
│ ├── main.py
│ └── requirements.txt
├── report_agent/
│ ├── main.py
│ └── requirements.txt
└── s.yaml # Serverless Devs 配置
5.2 部署配置
# s.yaml
edition: 4.0.0
name: multi-agent-analysis
services:
search:
component: fc
props:
region: cn-hangzhou
serviceName: adk-multi-agent
functionName: search-agent
runtime: python3.9
codeUri: ./search_agent
handler: main.search_agent
memorySize: 1024
# 挂载 NAS 实现 Agent 间文件共享
nasConfig:
groupId: 0
userId: 0
mountPoints:
- serverAddr: "xxxxx.cn-hangzhou.nas.aliyuncs.com"
mountDir: /mnt/nas
analysis:
component: fc
props:
region: cn-hangzhou
serviceName: adk-multi-agent
functionName: analysis-agent
runtime: python3.9
codeUri: ./analysis_agent
handler: main.analysis_agent
memorySize: 2048
nasConfig:
groupId: 0
userId: 0
mountPoints:
- serverAddr: "xxxxx.cn-hangzhou.nas.aliyuncs.com"
mountDir: /mnt/nas
report:
component: fc
props:
region: cn-hangzhou
serviceName: adk-multi-agent
functionName: report-agent
runtime: python3.9
codeUri: ./report_agent
handler: main.report_agent
memorySize: 1024
nasConfig: ...
5.3 一键部署
# 部署全部 Agent
s deploy
# 部署结果示例：
# search-agent: deployed ✓ (URL: https://xxx.cn-hangzhou.fc.aliyuncs.com/...)
# analysis-agent: deployed ✓
# report-agent: deployed ✓
部署完成后，三个 Agent 各自获得一个独立的 HTTP 端点，通过 A2A 协议互相调用，无需额外配置服务发现或负载均衡——Serverless 架构天然处理了这些问题。
六、成本分析
多 Agent 系统的传统部署方式需要至少 3 台 ECS 实例（或一个 K8s 集群），基础成本每月几百到上千元，而且不管有没有请求都要付费。
AgentRun 的 Serverless 模式按实际调用计费：
- 无请求时：不产生计算费用（NAS 存储除外，约 ¥0.35/GB/月）
- 低频调用（日均几十次）：每月成本可能不到 ¥10
- 高频调用：弹性伸缩，按需付费，不会因流量骤增而宕机
参考吉利汽车在 AgentRun 上的实际案例：算力成本优化了 33%。
如果你想亲自体验 AgentRun 的多 Agent 开发能力，可以通过链接注册并开通：体验 AgentRun 的多 Agent 开发能力
七、局限与注意事项
- 框架锁定：Google ADK 和 PydanticAI 是当前支持的两个主要框架，如果你团队用的是 LangGraph 或 CrewAI，需要评估迁移成本
- NAS 性能：Agent 间通过 NAS 共享文件是当前推荐的方案，适合中等数据量的场景；高频、小数据量的通信场景建议等待原生消息队列支持
- 冷启动：函数计算存在冷启动延迟（通常在百毫秒级），对实时性要求极高的场景需要预热策略
八、总结
本文从零搭建了一个基于 AgentRun + Google ADK 的多 Agent 竞品分析系统。核心收获：
- A2A 协议让 Agent 间通信标准化，降低了系统复杂度
- Serverless 架构免去了运维负担，按需付费成本更可控
- NAS 共享存储解决了 Agent 间的状态传递问题
- 百炼大模型提供了开箱即用的模型能力，无需自行部署
AI Agent 正在从"能用"走向"好用"，而多 Agent 协同是其中最关键的一块拼图。AgentRun 让这块拼图的落地门槛降低了不少。
