当你只有一个 AI 助手时，它是个工具。当你有多个 AI 助手相互协作时，你就有了一支团队。本文带你用 Google ADK 框架 + 阿里云 AgentRun，从零搭建一个多 Agent 协同系统。
为什么要多 Agent？
单 Agent 的局限性越来越明显。一个典型的场景：你让一个 Agent 同时完成"搜索资料 → 分析数据 → 写报告 → 画图表"，它往往顾此失彼。context window 被塞满，推理链条过长导致幻觉频发，一个环节出错整条链路崩溃。
多 Agent 协同的核心思想很简单：把复杂任务拆给多个专业 Agent，就像公司里有研发、测试、运维的分工。
Google 2025 年推出的 Agent Development Kit (ADK) 和 A2A 协议（Agent-to-Agent Protocol），正是为了解决这个问题——定义了一套多 Agent 之间如何发现彼此、如何通信、如何协作的标准。
A2A 协议的核心概念
A2A 协议的核心抽象是三个角色：
┌─────────────┐ 发现/调用 ┌─────────────┐
│ Client │ ──────────────→ │ Remote │
│ Agent │ ←────────────── │ Agent │
└─────────────┘ 返回结果 └─────────────┘
- Client Agent：发起调用的一方，类似"项目经理"
- Remote Agent：被调用的一方，类似"专业执行者"
- Agent Card：每个 Agent 的能力声明卡片（JSON 格式），类似"简历"
当你注册一个 Remote Agent 时，它会暴露一张 Agent Card，描述自己能做什么、需要什么参数、输出什么。Client Agent 读取这些 Card，按需调度。
这不就是微服务的 Agent 版本吗？确实如此。A2A 把微服务架构中"服务发现、RPC 调用、负载均衡"的思想搬到了 Agent 世界。
实战：搭建一个"氛围编程 + 代码审查"双 Agent 系统
场景设计
我们搭建两个 Agent：
| Agent | 角色 | 能力 |
|---|---|---|
| VibeCoder | 氛围编程专家 | 根据需求描述直接生成代码 |
| CodeReviewer | 代码审查专家 | 审查代码质量、安全性、性能 |
工作流：用户提需求 → VibeCoder 生成代码 → CodeReviewer 审查 → 返回最终结果
Step 1：定义 Agent Card
每个 Agent 需要一张能力声明卡片：
# vibe_coder_card.py
agent_card = {
"name": "VibeCoder",
"description": "氛围编程专家，根据自然语言需求生成代码",
"url": "https://your-agent-domain.com/vibecoder",
"capabilities": {
"streaming": True,
"pushNotifications": False
},
"skills": [
{
"id": "generate_code",
"name": "代码生成",
"description": "根据需求描述生成前端/后端代码",
"inputSchema": {
"type": "object",
"properties": {
"requirement": {
"type": "string", "description": "需求描述"},
"language": {
"type": "string", "enum": ["python", "javascript", "typescript"]},
"framework": {
"type": "string", "description": "目标框架"}
},
"required": ["requirement", "language"]
}
}
]
}
# code_reviewer_card.py
agent_card = {
"name": "CodeReviewer",
"description": "代码审查专家，检查代码质量和安全性",
"url": "https://your-agent-domain.com/codereviewer",
"capabilities": {
"streaming": True,
"pushNotifications": False
},
"skills": [
{
"id": "review_code",
"name": "代码审查",
"description": "审查代码并给出改进建议",
"inputSchema": {
"type": "object",
"properties": {
"code": {
"type": "string", "description": "待审查的代码"},
"language": {
"type": "string"},
"focus_areas": {
"type": "array",
"items": {
"enum": ["security", "performance", "readability", "architecture"]}
}
},
"required": ["code", "language"]
}
}
]
}
Step 2：实现 Agent 核心逻辑
# agent_core.py
from google.adk.agents import LlmAgent
from google.adk.tools import built_in_tools
from google.adk.sessions import SessionService
class VibeCoderAgent:
"""氛围编程 Agent"""
def __init__(self):
self.agent = LlmAgent(
name="VibeCoder",
model="qwen-plus", # 通义千问
instruction="""
你是一个专业的代码生成助手。请遵循以下原则：
1. 生成完整可运行的代码
2. 包含必要的注释和文档
3. 遵循目标语言的最佳实践
4. 输出时标注语言和文件名
""",
tools=[built_in_tools.code_executor]
)
async def generate(self, requirement: str, language: str, framework: str = None) -> dict:
prompt = f"""
需求：{requirement}
语言：{language}
框架：{framework or '无'}
请生成完整代码，包括：
- 项目结构说明
- 核心代码文件
- 依赖配置
- 运行说明
"""
response = await self.agent.run(prompt)
return {
"agent": "VibeCoder",
"code": response.text,
"language": language
}
class CodeReviewerAgent:
"""代码审查 Agent"""
def __init__(self):
self.agent = LlmAgent(
name="CodeReviewer",
model="qwen-plus",
instruction="""
你是一个严格的代码审查员。请从以下维度审查代码：
1. 安全性：是否存在注入、密钥泄露等风险
2. 性能：是否有明显的性能瓶颈
3. 可读性：命名、结构是否清晰
4. 架构：设计是否合理
使用评分制（1-10），并给出改进建议。
""",
tools=[built_in_tools.code_executor]
)
async def review(self, code: str, language: str, focus_areas: list = None) -> dict:
areas = focus_areas or ["security", "performance", "readability", "architecture"]
prompt = f"""
请审查以下 {
language} 代码，重点关注：{
', '.join(areas)}
```{
language}
{
code}
请按维度给出评分和改进建议。
"""
response = await self.agent.run(prompt)
return {
"agent": "CodeReviewer",
"review": response.text
}
### Step 3：编排协同流程
```python
# orchestrator.py
import asyncio
class MultiAgentOrchestrator:
"""多 Agent 协同编排器"""
def __init__(self):
self.vibe_coder = VibeCoderAgent()
self.code_reviewer = CodeReviewerAgent()
async def execute(self, requirement: str, language: str, framework: str = None):
print(f"[Orchestrator] 接收需求：{requirement}")
# 阶段1：生成代码
print("[Orchestrator] → 调度 VibeCoder 生成代码...")
code_result = await self.vibe_coder.generate(requirement, language, framework)
print(f"[Orchestrator] ✓ 代码生成完成 ({len(code_result['code'])} 字符)")
# 阶段2：审查代码
print("[Orchestrator] → 调度 CodeReviewer 审查代码...")
review_result = await self.code_reviewer.review(
code_result['code'],
language,
["security", "performance", "readability"]
)
print("[Orchestrator] ✓ 代码审查完成")
# 阶段3：整合结果
return {
"requirement": requirement,
"code": code_result["code"],
"review": review_result["review"],
"summary": self._summarize(review_result["review"])
}
def _summarize(self, review_text: str) -> str:
# 提取评分和关键建议
return review_text[:500] # 简化版摘要
# 使用示例
async def main():
orchestrator = MultiAgentOrchestrator()
result = await orchestrator.execute(
requirement="创建一个 FastAPI 接口，接受用户查询并返回数据库搜索结果",
language="python",
framework="fastapi"
)
print(f"\n=== 最终结果 ===\n{result['code']}\n\n=== 审查意见 ===\n{result['review']}")
if __name__ == "__main__":
asyncio.run(main())
部署到生产：为什么选择 Serverless Agent 平台？
到这里你可能会问：写了这么多代码，部署在哪？自己搭服务吗？
自己部署会遇到一个很现实的问题：Agent 是长时运行任务，不像普通 API 请求几百毫秒就能返回。一个多 Agent 协同流程可能要跑几分钟甚至更久。用传统服务器部署，你需要处理：
- 长连接管理（WebSocket / SSE）
- 并发任务调度（同时跑 10 个 Agent 流程怎么办）
- GPU 资源弹性（推理时用 GPU，空闲时别浪费钱）
这就是为什么我选择了 阿里云 AgentRun（函数计算 FC + 百炼的 Agent 平台）：
- Serverless 免运维：不用管服务器，Agent 跑在函数计算上
- GPU 弹性伸缩：推理时自动分配 GPU 算力，空闲时缩到 0
- 内置模型接入：直接用百炼的通义千问等大模型，不用自己搭模型服务
- API 集成：部署完直接拿 API 凭证，在业务代码里调用
把上面的代码部署到 AgentRun 上，只需要三步：
- 在 AgentRun 控制台选择"Google ADK 多 Agent 协同"模板
- 上传上面写的 Agent Card 和核心代码
- 点击部署，拿 API 端点
不需要配服务器，不需要装 CUDA，不需要写 Kubernetes YAML。
多 Agent 协同的最佳实践
在实际项目中踩过一些坑，分享几条经验：
1. Agent 粒度要适中。别拆太细——一个 Agent 只做一件事（比如"验证邮箱格式"），调用开销会很高。也别太粗——一个 Agent 包揽所有事，就失去了协同的意义。参考标准：一个 Agent 的职责能用一句话说清楚，且无需内部多步骤决策。
2. 错误处理要分级。Agent A 调用 Agent B 失败了怎么办？建议三层兜底：
- L1：重试（幂等任务）
- L2：降级（用规则引擎代替）
- L3：透传失败信息给用户
3. 监控多 Agent 的"端到端延迟"。单 Agent 的延迟可能是 2 秒，但经过 3 个 Agent 串行调用就变成 6 秒。用 AgentRun 的链路追踪，可以清楚看到每个 Agent 的耗时分布。
4. A2A 协议的 Agent Card 要写得诚实。别夸大能力——Agent Card 是"服务契约"，夸大了会导致 Client Agent 做出错误调度决策。
总结
多 Agent 协同不是噱头，是解决复杂任务工程的必然方向。Google ADK + A2A 协议提供了一个标准化的框架，而 Serverless Agent 平台（如阿里云 AgentRun）解决了部署和运维的难题。
如果你也想试试，可以从官方的"氛围编程专家"模板开始——这是一个基于 ADK 的多 Agent 协同 Demo，展示了上面说的完整流程。
本文中的示例代码已经内置在 阿里云 AgentRun 的 Google ADK 模板 中，可以直接免费体验：点击跳转到免费体验地址
