Google ADK 2.0 深度解析：从单 Agent 到多智能体协作，Google 交出了怎样的工程化答卷？

前言：为什么 2026 年我们还需要另一个 Agent 框架？

2026 年的 AI Agent 生态，已经不是 2023-2024 年那个"调个 API 就能跑 Demo"的蛮荒时代了。随着 LLM 推理能力的整体提升，行业的关注点正在从"模型有多强"转向"系统有多稳"。一个不争的事实是：单 Agent 在真实生产环境中，能力是有上限的。

你让一个 Agent 干所有事情，它要么上下文爆炸（Context Overflow），要么工具调用链过长导致不可控，要么在遇到边界情况时完全失效。真实的企业场景——比如一个客服系统需要同时处理用户意图识别、信息检索、工单生成、情感分析、主管审核等多个环节——靠一个 Agent 打天下基本上是痴人说梦。

正是在这个背景下，Google 在 2026 年 4 月正式发布了 Agent Development Kit（ADK）2.0，将多智能体协作从理论推向了工程化落地的阶段。这个框架的出现，让很多团队看到了在生产环境中构建可靠 Agent 系统的可能性。

但问题来了：市面上的 Agent 框架已经多如牛毛了——LangGraph、AutoGen、CrewAI、Semantic Kernel……Google ADK 2.0 到底有什么不一样？它解决了什么别人没解决的问题？

这篇文章，我会从架构设计、核心原理、代码实战、工程落地四个维度，把 Google ADK 2.0 掰开揉碎讲清楚。我会给出大量可以直接运行的代码，也会对比它与其他主流框架的差异，让你在读完这篇文章后，能够判断 ADK 是否适合你的场景，以及如何用它构建生产级的多 Agent 系统。

一、从单 Agent 到多 Agent：为什么这是必然的演进？

1.1 单 Agent 的三大瓶颈

在深入 ADK 之前，我们先搞清楚为什么多 Agent 协作是刚需，而不是一个"听起来很酷但用不到"的技术概念。

瓶颈一：上下文膨胀（Context Explosion）

当一个 Agent 需要处理复杂任务时，它不得不把大量的中间状态、历史信息、工具返回结果都塞进上下文窗口。随着任务步骤增加，上下文 token 数量呈指数级增长，不仅消耗巨大成本，更会导致模型在长上下文中"遗忘"早期关键信息。举个例子，一个代码审查 Agent 需要同时检查 10 个文件的变更、历史代码规范、项目特定约定，如果所有信息都塞进一个 Agent 的上下文，那在第 8 个文件时它很可能已经忘了前 3 个文件的审查标准了。

瓶颈二：职责耦合（Responsibility Coupling）

一个通用的 Agent，在面对多样化的任务时必须具备多种能力：语言理解、工具调用、逻辑推理、代码生成、数据分析……但"什么都会"往往意味着"什么都不精"。用同一个 Agent 既做情感分析又做技术代码生成，prompt 工程师会告诉你这有多痛苦——两个任务的最优 system prompt 往往是相互冲突的。

瓶颈三：故障放大（Failure Amplification）

单 Agent 一旦出错，整个系统崩溃。没有中间检查点，没有人工回退机制，更没有局部重试的概念。在生产环境中，这基本上等同于灾难。

1.2 多 Agent 协作的核心哲学

多 Agent 系统的核心思想很简单：分而治之（Divide and Conquer）。

将复杂任务拆分为多个专用的子 Agent，每个 Agent 专注做好一件事，通过消息传递和协作机制将它们串联起来，形成一个整体。这个模式在分布式系统和微服务架构中已经被验证了几十年，现在不过是把它迁移到 AI Agent 领域罢了。

Google ADK 2.0 的出现，正是这一哲学的工程化实现。它提供了：

• 结构化的 Agent 抽象：定义 Agent 不再是写一个巨大的 prompt，而是构建一个层次化的组件体系
• 标准化的协作协议：A2A 协议让不同来源的 Agent 可以互操作
• 工程化的部署路径：从本地开发到 Cloud Run / Vertex AI 的完整闭环
• 生产级的可观测性：日志、追踪、回调机制一应俱全

二、Google ADK 2.0 核心架构解析

2.1 整体架构鸟瞰

Google ADK 的架构设计，可以用"三层 + 两协议"来概括。

┌─────────────────────────────────────────────────────────────┐
│                    应用层 (Application Layer)                │
│  单Agent / 多Agent协作 / Graph工作流 / 协作智能体             │
├─────────────────────────────────────────────────────────────┤
│                    框架层 (Framework Layer)                  │
│  上下文管理 | 工具系统 | 记忆系统 | 回调机制 | 部署适配       │
├─────────────────────────────────────────────────────────────┤
│                    运行时层 (Runtime Layer)                  │
│  LLM调用 | A2A协议 | MCP协议 | Vertex AI | Cloud Run        │
└─────────────────────────────────────────────────────────────┘

三层职责清晰：

• 应用层：开发者直接打交道的层，定义 Agent 行为和工作流
• 框架层：提供上下文管理、工具集成、错误处理等基础设施
• 运行时层：处理与外部系统的交互，包括 LLM 调用和协议通信

2.2 Agent 类型体系

ADK 定义了一套层次化的 Agent 类型体系，这是理解整个框架的关键。

2.2.1 BaseAgent：所有 Agent 的基类

BaseAgent 定义了所有 Agent 的公共接口，包括：

• name: Agent 的唯一标识
• model: 所使用的 LLM（可以是 Gemini、GPT-4、Claude 等）
• instruction: 系统提示词（Agent 的行为定义）
• description: Agent 的描述（用于自动路由和协作）
• tools: Agent 可调用的工具列表
• sub_agents: 子 Agent 列表（用于多 Agent 协作）

from google.adk import Agent

# 最基础的 Agent 定义
base_agent = Agent(
    name="base_assistant",
    model="gemini-2.5-flash",
    instruction="你是一个有帮助的助手。",
    description="基础助手"
)

2.2.2 LlmAgent：LLM 驱动的智能体

LlmAgent 是 ADK 中最常用的 Agent 类型，由 LLM 驱动推理和决策。你可以把它理解为"有脑子"的 Agent——它能理解自然语言、进行复杂推理、调用工具、生成内容。

from google.adk.agents import LlmAgent

# 创建一个搜索 Agent
search_agent = LlmAgent(
    name="web_searcher",
    model="gemini-2.5-flash",
    instruction="""
    你是一个专业的网络研究员。
    当用户提出需要实时信息的问题时，使用 google_search 工具。
    搜索结果要准确、全面，并注明信息来源。
    如果搜索结果不足以回答问题，明确告知用户。
    """,
    description="网络搜索研究员",
    tools=["google_search"]
)

# 创建一个代码审查 Agent
code_reviewer = LlmAgent(
    name="code_reviewer", 
    model="gemini-2.5-pro",
    instruction="""
    你是一个严格的代码审查员。
    审查重点：
    1. 代码安全性（SQL注入、XSS、敏感信息泄露等）
    2. 性能问题（N+1查询、低效算法、内存泄漏等）
    3. 代码可读性和可维护性
    4. 是否遵循项目编码规范
    
    每次审查结果必须包含：
    - 问题严重程度（Critical/Major/Minor）
    - 具体代码位置
    - 修复建议
    """,
    description="代码审查专家",
    tools=["code_execution", "file_system"]
)

2.2.3 WorkflowAgent：流程控制型智能体

WorkflowAgent 是 ADK 2.0 引入的核心新类型，专门用于编排复杂的工作流程。它不依赖 LLM 做决策，而是按照预定义的逻辑执行任务序列。

ADK 2.0 提供了三种内置的 WorkflowAgent：

SequentialAgent（顺序执行）：按顺序执行一系列子 Agent，上一个完成才执行下一个。

from google.adk.agents.workflow import SequentialAgent

# 顺序执行的数据处理流水线
data_pipeline = SequentialAgent(
    name="data_processing_pipeline",
    description="数据处理流水线",
    sub_agents=[
        data_extractor,    # Step 1: 从数据库提取原始数据
        data_cleaner,      # Step 2: 数据清洗
        data_transformer,  # Step 3: 数据转换
        data_loader        # Step 4: 加载到目标系统
    ]
)

ParallelAgent（并行执行）：同时执行多个子 Agent，适用于独立任务的并发处理。

from google.adk.agents.workflow import ParallelAgent

# 并行处理多路数据源
multi_source_processor = ParallelAgent(
    name="multi_source_processor",
    description="多数据源并行处理器",
    sub_agents=[
        github_data_fetcher,    # 从 GitHub 获取数据
        twitter_data_fetcher,   # 从 Twitter 获取数据
        news_data_fetcher,      # 从新闻源获取数据
        reddit_data_fetcher     # 从 Reddit 获取数据
    ]
)
# 四个数据源同时请求，总耗时 = max(t1, t2, t3, t4) 而不是 t1+t2+t3+t4

LoopAgent（循环执行）：重复执行直到满足退出条件。

from google.adk.agents.workflow import LoopAgent

# 循环优化直到达到质量阈值
quality_loop = LoopAgent(
    name="quality_improvement_loop",
    description="质量改进循环",
    max_iterations=10,  # 最多迭代10次
    exit_condition="""
    当改进质量评分达到 8.0/10 时退出，
    或者达到最大迭代次数时退出
    """,
    sub_agents=[code_generator, quality_evaluator, refactorer]
)

2.3 上下文管理：ADK 的杀手锏

ADK 相比其他框架最值得称道的特性之一，就是它的上下文管理机制。ADK 不是简单地把所有历史对话和工具返回结果拼接成字符串塞给 LLM，而是将上下文视为"一等公民"来管理。

ADK 的上下文模型包含以下组件：

# ADK 的上下文组件
class Context:
    session: Session           # 当前会话状态
    memory: Memory             # 长期记忆（跨会话持久化）
    tools_output: list         # 工具调用结果
    artifacts: list            # 工作产物（生成的文件、数据等）
    token_usage: dict          # Token 使用统计

ADK 上下文管理的核心策略：

1. 自动过滤（Auto-filtering）：自动识别并移除与当前任务无关的上下文内容
2. 渐进式摘要（Progressive Summarization）：对较旧的对话轮次自动生成摘要，避免上下文膨胀
3. 延迟加载（Lazy Loading）：大型工作产物（如文件、代码库）只在需要时才加载到上下文
4. Token 预算控制（Budget Control）：实时追踪 token 使用量，在接近限制时主动压缩上下文

# 自定义上下文管理策略
from google.adk.context import ContextConfig

agent = LlmAgent(
    name="context_aware_agent",
    model="gemini-2.5-pro",
    instruction="你是一个数据分析师",
    context_config=ContextConfig(
        max_tokens=100000,          # 最大上下文长度
        summary_threshold=0.7,       # 超过 70% 时触发摘要
        auto_filter=True,            # 自动过滤无关信息
        lazy_load_artifacts=True     # 延迟加载大型产物
    )
)

这个上下文管理机制，是 ADK 在处理长任务、多轮协作时表现优于 LangGraph 等竞品的核心技术原因。

三、ADK 2.0 新特性：Graph 工作流深度剖析

3.1 为什么要引入 Graph 工作流？

在 ADK 1.x 版本中，多 Agent 协作主要通过 sub_agents 列表和简单的顺序/并行执行来实现。这种方式对于线性流程和简单分支是够用的，但面对复杂的业务场景——比如一个有多个检查点、多个回退路径、多个并行分支的工作流——就力不从心了。

Graph 工作流是 ADK 2.0 最重量级的新特性，它引入了基于有向图的工作流定义，让你可以精确描述复杂任务之间的依赖关系和流转条件。

3.2 Graph 工作流基础

from google.adk.agents.graph import Graph, State

# 定义一个完整的代码发布工作流
release_workflow = Graph(
    name="software_release_workflow",
    start="code_check",      # 起始状态
    end="deployment_done",   # 结束状态
    states={
        # 节点定义
        "code_check": State(
            type="agent",
            agent=code_checker,
            description="检查代码规范和安全"
        ),
        "unit_tests": State(
            type="agent",
            agent=unit_test_runner,
            description="运行单元测试"
        ),
        "integration_tests": State(
            type="agent",
            agent=integration_test_runner,
            description="运行集成测试"
        ),
        "security_scan": State(
            type="agent",
            agent=security_scanner,
            description="安全扫描"
        ),
        "approval": State(
            type="human_approval",   # 人工审批节点
            description="人工审批发布"
        ),
        "deployment_done": State(
            type="terminal",           # 终态节点
            description="发布完成"
        )
    },
    # 边定义（状态转换）
    transitions=[
        # 代码检查通过 → 运行单元测试
        {"from": "code_check", "to": "unit_tests", "condition": "pass"},
        # 代码检查失败 → 直接结束（不发布）
        {"from": "code_check", "to": "deployment_done", "condition": "fail"},
        # 单元测试通过 → 并行运行集成测试和安全扫描
        {"from": "unit_tests", "to": "integration_tests", "condition": "pass"},
        {"from": "unit_tests", "to": "security_scan", "condition": "pass"},  # 并行分支
        # 集成测试和安全扫描都通过 → 人工审批
        {"from": "integration_tests", "to": "approval", "condition": "all_pass"},
        {"from": "security_scan", "to": "approval", "condition": "all_pass"},
        # 审批通过 → 发布
        {"from": "approval", "to": "deployment_done", "condition": "approved"},
        # 审批拒绝 → 结束
        {"from": "approval", "to": "deployment_done", "condition": "rejected"},
    ]
)

3.3 动态路由与条件分支

Graph 工作流的核心能力在于条件路由——根据当前状态动态决定下一个执行节点。

from google.adk.agents.graph import Graph, State, Condition

# 智能客服工作流：根据用户意图动态路由
customer_service_workflow = Graph(
    name="intelligent_customer_service",
    start="intent_classification",
    end="case_closed",
    states={
        "intent_classification": State(type="agent", agent=intent_classifier),
        "refund_processing": State(type="agent", agent=refund_agent),
        "technical_support": State(type="agent", agent=tech_support_agent),
        "sales_consultation": State(type="agent", agent=sales_agent),
        "escalation": State(type="agent", agent=human_escalation_agent),
        "sentiment_check": State(type="agent", agent=sentiment_analyzer),
        "case_closed": State(type="terminal")
    },
    transitions=[
        # 意图分类后根据类型路由到对应 Agent
        {"from": "intent_classification", "to": "refund_processing", 
         "condition": "intent=='refund'"},
        {"from": "intent_classification", "to": "technical_support", 
         "condition": "intent=='technical'"},
        {"from": "intent_classification", "to": "sales_consultation", 
         "condition": "intent=='sales'"},
        
        # 所有处理路径都需要经过情感分析
        {"from": "refund_processing", "to": "sentiment_check"},
        {"from": "technical_support", "to": "sentiment_check"},
        {"from": "sales_consultation", "to": "sentiment_check"},
        
        # 情感极负面 → 升级人工处理
        {"from": "sentiment_check", "to": "escalation", 
         "condition": "sentiment=='very_negative'"},
        # 情感正常 → 关闭工单
        {"from": "sentiment_check", "to": "case_closed", 
         "condition": "sentiment!='very_negative'"},
        
        # 人工处理完也走情感检查
        {"from": "escalation", "to": "sentiment_check"}
    ]
)

3.4 并行分支的同步与汇合

Graph 工作流支持复杂的并行分支，但关键是**汇合点（Join）**的处理——当多个并行分支都有输出时，如何决定下一步？

# 带汇合的多源数据聚合工作流
data_aggregation_workflow = Graph(
    name="multi_source_data_aggregation",
    start="fetch_all_sources",
    end="final_report",
    states={
        "fetch_all_sources": State(type="parallel", 
                                    agents=[github_fetcher, twitter_fetcher, 
                                           news_fetcher, reddit_fetcher]),
        "aggregate": State(type="agent", agent=data_aggregator),
        "deduplicate": State(type="agent", agent=deduplicator),
        "enrich": State(type="agent", agent=data_enricher),
        "final_report": State(type="terminal")
    },
    transitions=[
        # 所有数据源fetch完成后 → 聚合
        {"from": "fetch_all_sources", "to": "aggregate", "condition": "all_complete"},
        # 聚合 → 去重
        {"from": "aggregate", "to": "deduplicate"},
        # 去重 → 丰富
        {"from": "deduplicate", "to": "enrich"},
        # 丰富 → 生成报告
        {"from": "enrich", "to": "final_report"}
    ]
)

四、工具系统：MCP 集成与自定义工具

4.1 为什么工具系统如此重要？

没有工具的 Agent 只是一个语言模型——能说不能做。工具系统是 Agent 与外部世界交互的桥梁。ADK 提供了一套完整的工具框架，支持：

• 内置工具：开箱即用的 Google Search、代码执行、文件系统、计算器等
• MCP 工具：通过 Model Context Protocol 接入任何 MCP 服务器
• 自定义工具：用 Python 函数定义任意功能的工具

4.2 MCP 协议深度集成

MCP（Model Context Protocol）在 2026 年已经成为 AI Agent 开发的事实标准。ADK 对 MCP 提供了原生支持，这意味着你可以直接复用 MCP 生态中数以千计的工具。

from google.adk.tools.mcp import MCPTool

# 连接到文件系统 MCP 服务器
filesystem_mcp = MCPTool(
    server_name="filesystem",
    server_config={
        "command": "npx",
        "args": ["-y", "@modelcontextprotocol/server-filesystem", "/workspace"]
    },
    tools=["read_file", "write_file", "list_directory", "create_directory"]
)

# 连接到 GitHub MCP 服务器
github_mcp = MCPTool(
    server_name="github",
    server_config={
        "command": "npx",
        "args": ["-y", "@modelcontextprotocol/server-github"]
    },
    tools=["create_issue", "list_issues", "get_file_contents", "create_branch"]
)

# 连接到 Slack MCP 服务器
slack_mcp = MCPTool(
    server_name="slack",
    server_config={
        "command": "npx",
        "args": ["-y", "@modelcontextprotocol/server-slack"]
    },
    tools=["send_message", "list_channels", "search_messages"]
)

# 在 Agent 中组合使用多个 MCP 工具
full_stack_agent = LlmAgent(
    name="devops_assistant",
    model="gemini-2.5-pro",
    instruction="""
    你是一个全栈 DevOps 助手，能够：
    1. 读取和分析代码文件
    2. 管理 GitHub 仓库和 issue
    3. 向 Slack 频道发送通知
    4. 执行 shell 命令
    
    当你需要执行这些操作时，明确告诉用户你将使用的工具。
    """,
    tools=[filesystem_mcp, github_mcp, slack_mcp]
)

4.3 自定义工具：从函数到 Agent 能力

from google.adk.tools import FunctionTool

# 定义一个数据库查询工具
def query_database(query: str, database: str = "production") -> str:
    """
    执行 SQL 查询（只支持 SELECT，防止注入）
    
    Args:
        query: SQL 查询语句（必须以 SELECT 开头）
        database: 目标数据库名称
    
    Returns:
        查询结果的 JSON 字符串
    """
    if not query.strip().upper().startswith("SELECT"):
        return "错误：只支持 SELECT 查询"
    
    # 实际应用中连接真实数据库
    # 这里用模拟数据演示
    return f'{{"status": "success", "rows": 42, "data": [...], "query": "{query}"}}'

db_tool = FunctionTool(
    name="query_database",
    description="""执行只读 SQL 查询。
    仅支持 SELECT 语句，其他语句会被拒绝。
    适用于数据分析、报表生成等场景。""",
    parameters={
        "type": "object",
        "properties": {
            "query": {
                "type": "string",
                "description": "SQL SELECT 查询语句",
                "example": "SELECT * FROM users WHERE created_at > '2026-01-01'"
            },
            "database": {
                "type": "string",
                "description": "目标数据库名称",
                "enum": ["production", "staging", "analytics"],
                "default": "production"
            }
        },
        "required": ["query"]
    },
    function=query_database,
    safety_filter=True  # 自动过滤危险操作
)

# 定义一个 API 调用工具
def call_external_api(
    endpoint: str,
    method: str = "GET",
    headers: dict = None,
    body: dict = None
) -> str:
    """
    调用外部 REST API
    
    Args:
        endpoint: API 端点 URL
        method: HTTP 方法（GET/POST/PUT/DELETE）
        headers: 请求头
        body: 请求体（用于 POST/PUT）
    
    Returns:
        API 响应的 JSON 字符串
    """
    import json
    if headers is None:
        headers = {"Content-Type": "application/json"}
    
    # 实际应用中发起真实 HTTP 请求
    return json.dumps({"status_code": 200, "body": "response data"})

api_tool = FunctionTool(
    name="call_api",
    description="调用外部 REST API，支持 GET/POST/PUT/DELETE 方法",
    parameters={
        "type": "object",
        "properties": {
            "endpoint": {"type": "string", "description": "API 端点"},
            "method": {
                "type": "string",
                "enum": ["GET", "POST", "PUT", "DELETE"],
                "default": "GET"
            },
            "headers": {"type": "object"},
            "body": {"type": "object"}
        },
        "required": ["endpoint"]
    },
    function=call_external_api
)

五、A2A 协议：让不同框架的 Agent 可以对话

5.1 A2A 协议是什么？

如果说 MCP 解决的是"Agent 如何调用工具"的问题，那 A2A（Agent-to-Agent）协议解决的就是"Agent 如何与其他 Agent 对话"的问题。

A2A 是 Google 在 2025 年联合多家厂商推出的开放协议，它的核心设计理念是：

• 基于标准协议：HTTP + SSE + JSON-RPC，企业 IT 基础设施友好
• 安全协作：支持企业级身份验证和授权
• 任务传递：Agent 之间可以委托任务、传递上下文、共享结果
• 双向通信：支持同步和异步两种交互模式

5.2 ADK 中的 A2A 实现

from google.adk.a2a import A2AServer, A2AClient

# 定义一个支持 A2A 协议的 Agent
class WeatherA2AAgent(A2AServer):
    """天气查询 Agent，提供 A2A 接口"""
    
    def __init__(self):
        self.name = "weather_agent"
        self.description = "提供全球城市天气查询服务"
        self.capabilities = ["weather_query", "forecast"]
    
    async def handle_task(self, task):
        """处理来自其他 Agent 的任务请求"""
        if task.type == "weather_query":
            city = task.params.get("city")
            return await self.get_weather(city)
        elif task.type == "forecast":
            city = task.params.get("city")
            days = task.params.get("days", 7)
            return await self.get_forecast(city, days)
    
    async def get_weather(self, city: str) -> dict:
        """获取城市当前天气"""
        # 实际应用中调用天气 API
        return {
            "city": city,
            "temperature": 28,
            "condition": "多云",
            "humidity": 65,
            "wind_speed": 12
        }

# 在 ADK Agent 中调用其他 A2A Agent
async def call_weather_agent(city: str):
    """通过 A2A 协议调用外部天气 Agent"""
    client = A2AClient(
        url="http://weather-service:8080/a2a",
        auth_token="your_auth_token"
    )
    
    result = await client.send_task({
        "type": "weather_query",
        "params": {"city": city}
    })
    return result

5.3 MCP vs A2A：搞清楚它们的区别

很多开发者会把 MCP 和 A2A 搞混，这是一个非常常见的问题。下面用一个表格说清楚：

六、协作智能体：Multi-Agent 协作的工程化实践

6.1 主从模式（Master-Slave）

最经典的多 Agent 协作模式，一个主 Agent 负责任务分解和协调，多个子 Agent 负责具体执行。

from google.adk.agents import LlmAgent

# 子 Agent：接待员
greeter = LlmAgent(
    name="greeter",
    model="gemini-2.5-flash",
    instruction="""
    你是接待员，负责问候用户并快速了解用户需求。
    收集的信息包括：
    1. 用户姓名
    2. 问题类型（技术支持/产品咨询/投诉建议）
    3. 问题简述
    
    不要尝试解决问题，只负责收集信息并礼貌转交。
    """,
    description="接待员"
)

# 子 Agent：技术支持
tech_support = LlmAgent(
    name="tech_support",
    model="gemini-2.5-pro",
    instruction="""
    你是一个专业的技术支持工程师。
    当接到用户的技术问题时：
    1. 仔细分析问题症状
    2. 按照"收集信息 → 定位问题 → 给出方案 → 确认解决"的流程处理
    3. 如果需要执行命令，使用 code_execution 工具
    4. 复杂问题分步骤解决，每步确认用户理解
    
    回答要专业、耐心、清晰。
    """,
    description="技术支持工程师",
    tools=["code_execution", "google_search"]
)

# 子 Agent：投诉处理
complaint_handler = LlmAgent(
    name="complaint_handler",
    model="gemini-2.5-flash",
    instruction="""
    你是一个投诉处理专员。
    处理原则：
    1. 首先表达歉意和同理心
    2. 认真倾听用户的诉求
    3. 承诺合理的处理方案
    4. 记录投诉详情以便后续跟进
    
    永远保持冷静和专业，不与用户争辩。
    """,
    description="投诉处理专员"
)

# 主 Agent：协调者
coordinator = LlmAgent(
    name="customer_service_coordinator",
    model="gemini-2.5-pro",
    instruction="""
    你是客服中心的协调者，管理 greeter、tech_support 和 complaint_handler 三个子 Agent。
    
    工作流程：
    1. 让 greeter 接待用户并收集基本信息
    2. 根据用户问题类型，选择 tech_support 或 complaint_handler 处理
    3. 监督处理过程，必要时介入
    4. 处理完成后生成服务总结
    
    重要：子 Agent 的输出会汇总到你的上下文中，你可以综合判断。
    """,
    description="客服协调者",
    sub_agents=[greeter, tech_support, complaint_handler]
)

6.2 动态路由模式

from google.adk.routing import DynamicRouter

# 专家库：多个专业 Agent
experts = {
    "legal": LlmAgent(
        name="legal_expert",
        model="gemini-2.5-pro",
        instruction="你是一个法律顾问，专长：合同法、劳动法、知识产权法",
        description="法律专家"
    ),
    "finance": LlmAgent(
        name="finance_expert",
        model="gemini-2.5-pro",
        instruction="你是一个财务顾问，专长：税务筹划、财务报表分析、投资建议",
        description="财务专家"
    ),
    "engineering": LlmAgent(
        name="engineering_expert",
        model="gemini-2.5-pro",
        instruction="你是一个技术专家，专长：系统架构、代码审查、性能优化",
        description="技术专家"
    ),
    "hr": LlmAgent(
        name="hr_expert",
        model="gemini-2.5-flash",
        instruction="你是一个人力资源专家，专长：招聘、绩效管理、员工关系",
        description="HR专家"
    )
}

# 路由器：根据问题类型智能选择 Agent
router = DynamicRouter(
    agents=experts,
    default_agent="engineering",  # 默认路由到技术专家
    routing_prompt="""
    分析用户的问题，判断最适合的专家类型：
    - 法律相关问题 → legal
    - 财务/投资相关问题 → finance  
    - 技术/代码/系统相关问题 → engineering
    - 人事/招聘/管理相关问题 → hr
    
    如果问题涉及多个领域，选择最主要的领域。
    """
)

# 全科医生 Agent：拥有所有专家的访问权限
generalist = LlmAgent(
    name="general_consultant",
    model="gemini-2.5-pro",
    instruction="""
    你是一个全科顾问。当用户提出问题时：
    1. 快速判断问题领域
    2. 使用路由工具选择最合适的专家
    3. 整合专家回复，给出综合建议
    4. 如果专家建议有冲突，进行二次分析
    """,
    tools=[router]
)

6.3 辩论模式（Agent Debate）

# 两个 Agent 进行辩论，主 Agent 做最终裁决
pro_agent = LlmAgent(
    name="pro_arguer",
    model="gemini-2.5-pro",
    instruction="""
    你是正方辩手。针对用户提出的论点，从正方角度进行论证。
    论证要求：
    1. 论点清晰，论据充分
    2. 预判反方可能的反驳
    3. 提供数据和案例支撑
    
    你的目标是通过论证说服"裁判"。
    """
)

con_agent = LlmAgent(
    name="con_arguer",
    model="gemini-2.5-pro",
    instruction="""
    你是反方辩手。针对用户提出的论点，从反方角度进行论证。
    论证要求：
    1. 论点清晰，论据充分
    2. 反驳正方可能提出的论点
    3. 提供数据和案例支撑
    
    你的目标是通过论证说服"裁判"。
    """
)

judge_agent = LlmAgent(
    name="debate_judge",
    model="gemini-2.5-pro",
    instruction="""
    你是辩论裁判。在收到正方和反方的论点后：
    1. 评估双方论点的逻辑严密性
    2. 评估论据的充分性
    3. 指出双方的弱点
    4. 给出平衡且有深度的裁决
    
    裁决应该有理有据，不偏不倚。
    """,
    sub_agents=[pro_agent, con_agent]
)

七、开发工具链：从本地到生产的完整闭环

7.1 Agents CLI：命令行开发利器

ADK 提供了强大的 CLI 工具，让 Agent 开发可以像传统软件开发一样规范化。

# 安装
pip install google-adk

# 创建新 Agent 项目
adk init my_agent_project
cd my_agent_project

# 创建一个新 Agent
adk create agent --name code_review --model gemini-2.5-pro

# 运行本地开发服务器（带热重载）
adk web

# 运行特定 Agent
adk run --agent code_reviewer

# 测试 Agent
adk test --agent code_reviewer --input "审查这个函数的性能"

# 评估 Agent 性能
adk eval --agent code_reviewer --dataset ./test_cases.jsonl

# 部署到 Cloud Run
adk deploy --platform cloudrun --project my-gcp-project

# 部署到 Vertex AI
adk deploy --platform vertex --project my-gcp-project

7.2 内置 Web UI：可视化调试

adk web 命令启动一个本地 Web 服务器，提供：

• 实时对话界面：与 Agent 交互测试
• Token 使用监控：实时显示上下文 token 消耗
• 工具调用追踪：可视化展示每次工具调用的输入输出
• 上下文窗口：查看完整的上下文构建过程
• 重放功能：回放任意历史对话步骤

7.3 本地开发配置

# agent.yaml - Agent 项目配置文件
agent:
  name: production_agent
  model: gemini-2.5-pro
  description: 生产环境数据分析 Agent
  
context:
  max_tokens: 100000
  summary_threshold: 0.8
  auto_filter: true
  
tools:
  - google_search
  - code_execution
  - filesystem
  
deployment:
  platform: cloudrun
  region: asia-east1
  min_instances: 1
  max_instances: 10
  concurrency: 10
  
observability:
  tracing: true
  logging: structured
  metrics: true

7.4 Cloud Run 部署实战

# Dockerfile
FROM python:3.11-slim

WORKDIR /app

# 安装依赖
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

# 安装 ADK
RUN pip install google-adk

# 复制 Agent 代码
COPY agents/ ./agents/
COPY agent.yaml .

# 健康检查
HEALTHCHECK --interval=30s --timeout=10s --start-period=5s \
    CMD curl -f http://localhost:8080/health || exit 1

# 启动
CMD ["adk", "serve", "--port=8080", "--host=0.0.0.0"]

# 构建并推送镜像
gcloud builds submit --tag gcr.io/$PROJECT_ID/adk-agent:v1.0.0

# 部署到 Cloud Run
gcloud run deploy adk-agent \
    --image gcr.io/$PROJECT_ID/adk-agent:v1.0.0 \
    --platform managed \
    --region asia-east1 \
    --min-instances 1 \
    --max-instances 10 \
    --concurrency 10 \
    --memory 2Gi \
    --cpu 2 \
    --allow-unauthenticated \
    --set-env-vars GOOGLE_CLOUD_PROJECT=$PROJECT_ID

7.5 Vertex AI Agent Engine 部署

from google.cloud import agent_engine

# Vertex AI 部署配置
deployment_config = {
    "display_name": "production_code_review_agent",
    "description": "生产环境代码审查 Agent",
    "agent_uri": "./agents/code_reviewer",
    "runtime_version": "google-adk-2.0",
    "machine_type": "e2-standard-4",
    "min_instance_count": 1,
    "max_instance_count": 10,
    "autoscaling_metric": {
        "metric_name": "agent_request_count",
        "target_value": 100
    },
    "environment_variables": {
        "GEMINI_API_KEY": "$(GEMINI_API_KEY)",  # 从 Secret Manager 读取
        "LOG_LEVEL": "INFO"
    },
    "oauth_config": {
        "service_account_email": "agent-sa@$PROJECT.iam.gserviceaccount.com"
    }
}

# 创建部署
deployment = agent_engine.create(**deployment_config)

# 等待部署就绪
deployment.wait()

# 获取端点
endpoint = deployment.endpoint
print(f"Agent 已部署到: {endpoint}")

八、与其他框架的深度对比

8.1 ADK vs LangGraph

LangGraph 是目前最成熟的图结构 Agent 框架，由 LangChain 团队维护。

选 ADK 的场景：你在 Google Cloud 生态中，需要快速部署生产级 Agent
选 LangGraph 的场景：你需要最大的灵活性，愿意投入时间深入学习

8.2 ADK vs CrewAI

CrewAI 以"扮演角色"的多 Agent 理念著称，上手非常简单。

选 ADK 的场景：需要复杂工作流和生产级部署
选 CrewAI 的场景：快速做原型，场景简单

8.3 ADK vs AutoGen

AutoGen 来自 Microsoft，在代码生成和自动化场景中有深厚积累。

九、最佳实践：从 Demo 到生产的必经之路

9.1 Agent 设计原则

原则一：单一职责

每个 Agent 应该只负责一个明确的职责。不要试图让一个 Agent 既做情感分析又做代码生成——prompt 冲突会让两者都做不好。

# ❌ 反面典型：一个 Agent 做所有事
super_agent = LlmAgent(
    name="super_agent",
    instruction="你既能写代码、又能做情感分析、又能翻译语言、又能写诗..."
    # 这基本上等同于什么都不会
)

# ✅ 正确做法：职责分离
code_agent = LlmAgent(name="coder", instruction="你是一个专业的Python程序员...")
sentiment_agent = LlmAgent(name="sentiment", instruction="你是一个情感分析专家...")
translator = LlmAgent(name="translator", instruction="你是一个专业翻译...")

原则二：清晰的 Agent 描述

ADK 的 Agent 描述（description）字段不只是文档，它会被用于自动路由和协作。

# ✅ 好的 description：清晰说明了 Agent 的能力和边界
code_reviewer = LlmAgent(
    name="security_reviewer",
    model="gemini-2.5-pro",
    instruction="你是代码安全审查专家...",
    description="专注于安全审查，仅处理 Python/Go/Java 代码，"
                "不处理前端代码或数据库 schema 变更，"
                "发现严重漏洞会自动触发人工审批流程"
)

# ❌ 差的 description：说了等于没说
code_reviewer = LlmAgent(
    name="reviewer",
    instruction="...",
    description="审查代码"
)

原则三：分层设计

# 第一层：专家 Agent（细粒度，专一职责）
python_expert = LlmAgent(name="python_expert", ...)
security_expert = LlmAgent(name="security_expert", ...)
performance_expert = LlmAgent(name="performance_expert", ...)

# 第二层：领域协调者（聚合专家，提供领域级服务）
backend_coordinator = LlmAgent(
    name="backend_coordinator",
    sub_agents=[python_expert, security_expert, performance_expert],
    instruction="管理后端代码审查流程..."
)

# 第三层：系统协调者（最高层，统筹全局）
system_coordinator = LlmAgent(
    name="system_coordinator",
    sub_agents=[backend_coordinator, frontend_coordinator, devops_coordinator],
    instruction="统筹整个系统质量审查..."
)

9.2 错误处理与容错

from google.adk.errors import AgentError, ToolExecutionError, ContextOverflowError
from google.adk.callbacks import Callback
import asyncio

class ProductionCallbacks(Callback):
    """生产级回调：记录、告警、自动恢复"""
    
    def __init__(self, alert_service):
        self.alert_service = alert_service
        self.retry_count = {}
    
    def on_agent_start(self, agent, input_data):
        print(f"[START] {agent.name}")
    
    def on_agent_end(self, agent, output_data):
        print(f"[END] {agent.name}")
    
    def on_tool_call(self, tool_name, input_params):
        print(f"[TOOL] {tool_name}: {input_params}")
    
    def on_error(self, agent, error: Exception):
        print(f"[ERROR] {agent.name}: {error}")
        
        # 分类处理
        if isinstance(error, ContextOverflowError):
            # 上下文溢出：触发自动摘要
            agent.trigger_context_compaction()
        elif isinstance(error, ToolExecutionError):
            # 工具执行失败：自动重试
            self._retry_tool_call(agent, tool_name, input_params)
        elif isinstance(error, AgentError):
            # Agent 错误：记录并可能升级
            self.alert_service.send_alert(
                level="warning",
                message=f"Agent {agent.name} 遇到错误: {error}",
                context={"agent": agent.name, "error": str(error)}
            )
    
    def _retry_tool_call(self, agent, tool_name, params):
        """指数退避重试"""
        max_retries = 3
        for attempt in range(max_retries):
            try:
                return agent.retry_tool(tool_name, params)
            except ToolExecutionError:
                if attempt < max_retries - 1:
                    wait_time = 2 ** attempt
                    print(f"重试 {tool_name}，等待 {wait_time}s...")
                    asyncio.sleep(wait_time)
                else:
                    raise

9.3 生产级监控

from google.adk.observability import MetricsCollector
import time

class ProductionMetrics(MetricsCollector):
    """生产级指标收集"""
    
    def collect(self, agent_name, event_type, data):
        # 记录关键指标
        metrics = {
            "agent": agent_name,
            "event": event_type,
            "timestamp": time.time(),
            "token_usage": data.get("token_count", 0),
            "latency_ms": data.get("duration_ms", 0),
            "success": data.get("success", False)
        }
        
        # 发送到监控系统（Prometheus/Grafana）
        self.prometheus_client.write(metrics)
        
        # 触发成本告警
        daily_cost = self.calculate_daily_cost()
        if daily_cost > self.cost_limit:
            self.alert_service.send_alert(
                level="critical",
                message=f"日成本 ${daily_cost} 超过限额 ${self.cost_limit}"
            )

十、适用场景分析与选型建议

10.1 强烈推荐使用 ADK 的场景

场景一：Google Cloud 生态深度集成

如果你的基础设施在 Google Cloud 上（BigQuery、GCS、Vertex AI、Spanner 等），ADK 与这些服务的集成是无缝的，可以节省大量对接工作。

from google.adk.integrations import BigQueryTool, VertexAIModel

# 直接使用 BigQuery 工具
bq_agent = LlmAgent(
    name="data_analyst",
    model="gemini-2.5-pro",
    tools=[BigQueryTool(project_id="my-gcp-project")],
    instruction="你是一个数据分析专家，使用 BigQuery 分析数据..."
)

# 使用 Vertex AI 上的企业模型
enterprise_agent = LlmAgent(
    model=VertexAIModel(
        endpoint="us-central1-aiplatform-enterprise@latest",
        max_tokens=8192
    )
)

场景二：复杂多 Agent 协作系统

当你的系统需要 5 个以上的 Agent 协作，且它们之间有复杂的依赖关系时，ADK 的 Graph 工作流是最佳选择。

场景三：A2A 协议互操作需求

如果你需要让自己的 Agent 与其他厂商的 Agent（比如 Salesforce 的 Einstein Agent、Microsoft 的 Copilot Agent）通信，A2A 协议是唯一的选择。

场景四：企业级部署和安全合规

ADK 在 Google Cloud 上的部署原生支持 OAuth 2.0、服务账号、VPC 访问控制等企业安全机制，对于有合规要求的团队来说是加分项。

10.2 不适合使用 ADK 的场景

场景一：快速原型验证

如果你的目标是"今天想个点子，明天就跑个 Demo"，CrewAI 或 AutoGen 的上手速度更快。ADK 的完整功能需要更多的配置工作。

场景二：极度轻量级嵌入

如果你的 Agent 需要嵌入到一个非常轻量的应用（比如一个小程序或浏览器插件），ADK 的依赖包体积可能是个问题。

场景三：非 Google 生态但需要本地部署

如果你的基础设施是纯本地部署（on-premise），且没有使用 Google Cloud，ADK 的优势会被大幅削弱。

十一、未来展望：ADK 的路线图与行业趋势

11.1 ADK 的技术演进方向

根据 Google ADK 团队的公开路线图，以下特性值得关注：

1. 多模态 Agent 原生支持：图像、视频、音频输入输出的 Agent 能力将更深度地整合进 ADK
2. 自适应工作流：Graph 工作流将支持根据 Agent 执行结果动态调整下一步，而不仅仅是条件分支
3. 联邦学习支持：跨组织的 Agent 协作，在保护数据隐私的前提下共享推理能力
4. 更丰富的预置工具集：预计推出更多垂直领域的预置 Agent（如金融分析 Agent、医疗 Agent）

11.2 多 Agent 系统的行业趋势

2026 年的多 Agent 系统，正在经历从"能用"到"好用"的转变：

• 标准化：A2A 和 MCP 正在成为行业事实标准，跨框架互操作将逐步实现
• 自动化编排：未来 AI Agent 系统将能根据任务复杂度自动决定是否需要多 Agent 协作
• 可解释性：多 Agent 协作过程的可解释性将成为焦点，"黑盒协作"难以满足企业合规需求
• 安全边界：Agent 越权问题（Agentic AI Security）将受到更多关注，工具调用的安全审核会变得更严格

总结：ADK 2.0 到底值不值得用？

Google ADK 2.0 不是又一个"Hello World 级别的玩具框架"。它的设计体现了 Google 对 AI Agent 工程化落地的深度思考——从上下文管理到 Graph 工作流，从 MCP 集成到 A2A 协议，每一项特性都直击当前 Agent 开发的痛点。

我认为 ADK 2.0 最核心的价值主张是：让多 Agent 系统从"手工小作坊"变成"工业化流水线"。

对于在 Google 生态中工作的团队，ADK 几乎是目前最完整、最工程化的选择。它可能不是上手最简单的，也可能不是最灵活的，但它在"开箱即用的完整性"上做到了相当高的水准。

当然，框架只是工具，真正的核心仍然是：你打算解决什么问题，以及你如何拆解这个问题。多 Agent 不是银弹，用不好反而会增加系统复杂度和维护成本。在动手之前，先想清楚你的场景是否真的需要多 Agent 协作。

参考链接：

• Google ADK GitHub：https://github.com/google/adk-python
• ADK 官方文档：https://google.github.io/adk-docs/
• ADK 中文文档：https://adk.wiki/
• A2A 协议规范：https://google.github.io/A2A/
• MCP 协议：https://modelcontextprotocol.io/