Google ADK 2.0 深度解析：图工作流、协作智能体与A2A协议——从单Agent到企业级多智能体编排的完整实战指南

2026年5月19日，Google 正式发布 Agent Development Kit (ADK) Python 2.0 GA；6月30日，ADK Go 2.0 GA 紧随其后。这不只是一个版本号的跃迁——ADK 2.0 引入的 Graph Workflow 引擎、动态工作流和协作智能体架构，正在重新定义「生产级 AI Agent」的含义。

一、为什么需要 ADK 2.0？从 Agent 框架的三大痛点说起

2026年被称为 AI Agent 爆发元年。从 LangChain 到 CrewAI，从 AutoGen 到 OpenAI Agent SDK，框架多到让人选择困难。但如果你真正尝试把这些框架用到生产环境，很快会撞上三堵墙：

1.1 确定性之殇：Agent 的「薛定谔式执行」

大多数 Agent 框架的核心逻辑是「LLM 推理 → 工具调用 → LLM 再推理」的循环。这种模式在 Demo 阶段看起来很酷，但一旦进入生产，问题就来了：同样的输入，Agent 可能走完全不同的执行路径。你无法预测它会调用哪些工具、以什么顺序调用、在哪个环节卡住。

这对需要审计、合规、可回放的企业场景来说是致命的。你不能告诉审计部门「这次执行路径不确定，看 LLM 心情」。

1.2 编排之困：从「一问一答」到「多步协作」

单个 Agent 能做的事有限。现实世界的复杂任务——比如「分析客户需求 → 查询产品目录 → 生成报价 → 审批流转 → 发送通知」——需要多个专业 Agent 协作完成。但传统的 Agent 框架要么只支持简单的顺序编排，要么把多 Agent 协作搞成了一团乱麻的回调地狱。

1.3 规模之痛：从本地调试到云端部署

很多框架在本地跑得好好的，一旦要部署到生产环境，就要面对状态持久化、断点恢复、水平扩展、可观测性等一系列工程问题。大部分框架把这些留给了开发者自己解决。

ADK 2.0 的设计哲学就是直面这三堵墙。 它不是又一个「包装 LLM API」的框架，而是一套完整的 Agent 工程基础设施。

二、ADK 2.0 架构全景：五大核心模块

ADK 2.0 的架构可以拆解为五个核心模块，每个模块解决一个关键问题：

┌─────────────────────────────────────────────────┐
│                   ADK 2.0 Architecture           │
├─────────────┬─────────────┬─────────────────────┤
│  LLM Agent  │ Graph       │ Collaborative       │
│  (推理核心)  │ Workflow    │ Workflows           │
│             │ (图工作流)   │ (协作智能体)         │
├─────────────┴─────────────┴─────────────────────┤
│            Dynamic Workflows (动态工作流)         │
├─────────────────────────────────────────────────┤
│     A2A Protocol + MCP + Context Management      │
│           (通信协议 + 工具标准 + 上下文管理)       │
└─────────────────────────────────────────────────┘

2.1 LLM Agent：推理核心

ADK 的基础单元仍然是 LLM Agent。一个最简单的 Agent 定义如下：

from google.adk import Agent
from google.adk.tools import google_search

agent = Agent(
    name="researcher",
    model="gemini-flash-latest",
    instruction="你负责帮助用户深入研究各类课题。",
    tools=[google_search],
)

这看起来和 1.x 没什么区别，但 ADK 2.0 在底层做了一个关键改变：Agent 现在是 BaseNode 的子类，不再是独立的执行器。这意味着 Agent 被纳入了图工作流引擎的统一调度，可以作为图中的一个节点参与复杂的编排逻辑。

这个改变的影响是深远的。在 1.x 中，Agent 的执行是自驱动的——它自己决定何时调用工具、何时返回结果。在 2.0 中，Agent 的执行由图引擎驱动——图引擎决定何时激活这个 Agent、如何处理它的输出、以及在出错时如何重试。

2.2 Graph Workflow：图工作流引擎

这是 ADK 2.0 最核心的新特性。图工作流允许你将 Agent 逻辑定义为一个由节点（Nodes）和边（Edges）组成的有向图，将 AI 推理能力与确定性代码逻辑编织在一起。

核心概念：

• Node（节点）：执行单元，可以是 LLM Agent、函数、工具调用、人工审批步骤等
• Edge（边）：节点之间的连接，定义执行流向
• Route（路由）：基于条件的分支逻辑，支持静态路由和动态路由

图工作流 vs 传统编排的区别：

传统编排（如 LangChain 的 SequentialChain）本质上是线性的——A → B → C。图工作流支持任意拓扑——分支、并行、循环、条件跳转，甚至可以在运行时动态修改图结构。

from google.adk import Workflow
from google.adk.workflow import node
from typing import Any

# 定义一个数据验证节点
@node(name="validate_input")
def validate(node_input: dict) -> dict:
    if not node_input.get("email"):
        raise ValueError("缺少必要字段: email")
    return {"valid": True, "data": node_input}

# 定义一个风险评估节点
@node(name="risk_assessment")
async def assess_risk(ctx: Any, node_input: dict) -> dict:
    # 调用 LLM 进行风险评估
    result = await ctx.run_llm(
        prompt=f"评估以下客户数据的风险等级: {node_input}",
        model="gemini-pro-latest"
    )
    return {"risk_level": result}

# 定义一个审批节点（人工介入）
@node(name="human_approval")
def approval(node_input: dict) -> dict:
    # 暂停工作流，等待人工审批
    return {"approved": None}  # 需要人工确认

# 组装图工作流
workflow = Workflow(
    name="customer_onboarding",
    edges=[
        ("START", "validate_input"),
        ("validate_input", "risk_assessment"),
        ("risk_assessment", "human_approval"),
        ("human_approval", "END"),
    ],
)

图工作流的四大优势：

1. 可预测性：执行路径在图定义时就确定了（静态图）或在运行时可审计（动态图）
2. 容错性：每个节点可以独立配置重试策略（RetryConfig），失败的节点不会导致整个流程崩溃
3. 可观测性：图引擎自动记录每个节点的输入、输出、耗时，天然适配分布式追踪
4. 可恢复性：支持断点恢复（Resume），已成功的节点在重试时会被跳过

2.3 Dynamic Workflows：动态工作流

静态图工作流适合流程固定的场景，但现实世界中很多流程是动态的——比如「如果风险评估为高风险，需要增加额外的审批步骤」或「根据客户类型选择不同的处理流程」。

ADK 2.0 的动态工作流解决了这个问题。它允许你用编程语言的全部能力来构建工作流，而不是被图的结构所限制。

from google.adk import Context, Workflow
from google.adk.workflow import node

@node(name="fetch_data")
async def fetch(node_input: str) -> dict:
    # 获取数据
    return {"records": [...], "page": 1, "total": 100}

@node(rerun_on_resume=True)
async def process_pages(ctx: Context, node_input: dict) -> dict:
    """动态工作流：分页处理数据，支持断点恢复"""
    page = 1
    all_results = []

    while True:
        # 获取当前页数据
        data = await ctx.run_node(fetch, node_input=f"page={page}")

        # 处理当前页
        result = await ctx.run_node(
            process_batch,
            node_input=data["records"]
        )
        all_results.extend(result)

        # 判断是否还有下一页
        if data["page"] >= data["total"]:
            break

        page += 1

    return {"processed": len(all_results), "results": all_results}

# 动态工作流的定义方式
root_agent = Workflow(
    name="data_pipeline",
    edges=[("START", process_pages)],
)

动态工作流的关键特性：

• 自动检查点：每个节点执行成功后自动记录检查点，恢复时跳过已完成的节点
• 原生异步：支持 async/await，充分利用 I/O 并发
• 编程语言原生控制流：while、for、if-else 都可以直接使用
• 状态持久化：工作流状态可以持久化到数据库，支持长时间运行的任务

2.4 Collaborative Workflows：协作智能体

当任务复杂度超过单个 Agent 的能力范围时，你需要多个专业 Agent 协作。ADK 2.0 的协作工作流提供了两种协作模式：

模式一：Coordinator + Sub-Agents（协调器模式）

from google.adk import Agent

# 定义专业子Agent
researcher = Agent(
    name="researcher",
    model="gemini-pro-latest",
    instruction="你是一个研究员，负责收集和分析信息。",
    tools=[search_tool, database_tool],
)

writer = Agent(
    name="writer",
    model="gemini-pro-latest",
    instruction="你是一个技术作家，负责将研究结果转化为文档。",
    tools=[markdown_tool],
)

reviewer = Agent(
    name="reviewer",
    model="gemini-pro-latest",
    instruction="你是一个审核员，负责审查文档质量。",
    tools=[grammar_check_tool],
)

# 定义协调器
coordinator = Agent(
    name="coordinator",
    model="gemini-pro-latest",
    instruction="你负责协调研究、写作和审核流程。先让研究员收集信息，再让作家撰写文档，最后让审核员审查。",
    sub_agents=[researcher, writer, reviewer],
)

模式二：Peer-to-Peer（对等协作）

在对等模式下，Agent 之间可以直接通信，不需要中间协调器。这适合需要频繁交互的场景，比如辩论、协商等。

协作工作流的核心机制：

1. 上下文隔离：每个子 Agent 有自己的上下文空间，不会互相干扰
2. 消息传递：Agent 之间通过结构化消息通信，而不是共享内存
3. 转移控制：协调器可以将控制权转移给子 Agent，子 Agent 完成后自动返回
4. 模式配置：可以配置协作模式为 auto（自动选择）、parallel（并行执行）等

2.5 Context Management：上下文管理

ADK 2.0 引以为豪的一个特性是其上下文管理机制。与简单的字符串拼接不同，ADK 把上下文当作「源代码」来管理：

# ADK 的上下文管理策略
context_strategies = {
    "session": "会话上下文 - 当前对话的完整状态",
    "memory": "记忆上下文 - 跨会话的长期记忆",
    "artifacts": "工作产物 - 文件、图片等二进制数据（延迟加载）",
    "tool_outputs": "工具输出 - 过滤无关事件，保留关键结果",
}

# 自动优化
auto_optimizations = [
    "过滤无关事件",
    "总结旧对话轮次",
    "延迟加载 Artifacts",
    "跟踪 Token 使用量",
]

这意味着即使 Agent 处理了大量数据，上下文窗口也不会被无用信息填满。每个 Token 都「物尽其用」。

三、A2A 协议：Agent 之间的「HTTP」

ADK 2.0 深度集成了 Agent-to-Agent (A2A) 协议，这是 Google 主导的开放标准，定义了 Agent 之间如何发现、通信和协作。

3.1 A2A vs MCP：互补而非竞争

很多人混淆 A2A 和 MCP（Model Context Protocol）。简单来说：

MCP 解决的是「Agent 如何使用工具」的问题，A2A 解决的是「Agent 如何与其他 Agent 对话」的问题。两者是互补关系。

3.2 A2A 的三个核心角色

A2A 协议定义了三个核心角色：

1. Client Agent：发起请求的 Agent（消费者）
2. Remote Agent：提供服务的 Agent（提供者）
3. A2A Server：承载 Remote Agent 的服务端点

工作流程：

Client Agent                    A2A Server (Remote Agent)
    │                                │
    │  1. Agent Card Discovery       │
    │  ────────────────────────────→ │
    │                                │
    │  2. Create Task                │
    │  ────────────────────────────→ │
    │                                │
    │  3. Task Status Updates        │
    │  ←──────────────────────────── │
    │                                │
    │  4. Get Result                 │
    │  ←──────────────────────────── │

Agent Card 是 A2A 的核心概念——它是一个 JSON 描述文件，类似于 API 的 OpenAPI Spec，描述了一个 Agent 的能力、端点、认证方式等信息。Client Agent 通过读取 Agent Card 来发现和理解 Remote Agent。

3.3 在 ADK 中使用 A2A

ADK 2.0 提供了开箱即用的 A2A 支持。将一个 ADK Agent 暴露为 A2A 服务只需要几行代码：

# 暴露 Agent 为 A2A 服务
from google.adk.a2a import A2AServer

# 已有的 ADK Agent
my_agent = Agent(
    name="financial_advisor",
    model="gemini-pro-latest",
    instruction="你是一个金融顾问。",
    tools=[portfolio_tool, market_data_tool],
)

# 一行代码暴露为 A2A 服务
server = A2AServer(agent=my_agent, port=8080)
server.start()

消费远程 Agent 同样简单：

from google.adk.a2a import A2AClient

# 连接到远程 Agent
remote_agent = A2AClient(url="https://finance-agent.example.com/a2a")

# 委托任务
task = await remote_agent.create_task(
    message="分析苹果公司最近一个季度的财报"
)

# 获取结果
result = await task.get_result()

3.4 A2A 的实际应用场景

场景一：跨组织协作

你的公司有一个内部的代码审查 Agent，合作伙伴公司有一个安全扫描 Agent。通过 A2A，你的代码审查 Agent 可以自动将代码提交给合作伙伴的安全扫描 Agent 进行检查，无需人工介入。

场景二：微服务架构中的 Agent 网络

在一个电商平台中，订单处理 Agent、库存管理 Agent、物流 Agent 各自独立部署，通过 A2A 协议通信。每个 Agent 可以独立扩展、独立更新，互不影响。

场景三：多语言 Agent 互操作

你的核心业务逻辑用 Python 写的 ADK Agent，但有一个遗留系统是 Java 实现的。通过 A2A 协议，Python Agent 可以无缝调用 Java Agent，反之亦然。

四、代码实战：从零构建一个多 Agent 协作系统

让我们通过一个完整的实战案例来展示 ADK 2.0 的能力。我们要构建一个「技术博客自动生成系统」，包含三个 Agent：

1. Research Agent：搜索技术热点
2. Writer Agent：撰写文章
3. Editor Agent：审查和优化文章

4.1 环境准备

# 创建虚拟环境
python3 -m venv .venv
source .venv/bin/activate

# 安装 ADK 2.0
pip install "google-adk>=2.0"

# 设置 API Key
export GOOGLE_API_KEY="your-api-key"

4.2 定义工具

# tools.py
from google.adk.tools import FunctionTool

def web_search(query: str) -> str:
    """搜索互联网获取最新信息"""
    # 这里集成实际的搜索 API
    import requests
    response = requests.get(
        "https://api.search.example.com/search",
        params={"q": query, "count": 5}
    )
    return response.json()

def save_article(title: str, content: str) -> str:
    """保存文章到文件系统"""
    filename = f"articles/{title.replace(' ', '_')}.md"
    with open(filename, "w") as f:
        f.write(content)
    return f"文章已保存到 {filename}"

search_tool = FunctionTool(web_search)
save_tool = FunctionTool(save_article)

4.3 定义 Agent

# agents.py
from google.adk import Agent, Workflow
from google.adk.workflow import node
from tools import search_tool, save_tool

# Research Agent
researcher = Agent(
    name="researcher",
    model="gemini-flash-latest",
    instruction="""你是一个技术研究员。你的任务是：
    1. 根据给定的主题搜索最新的技术动态
    2. 提取关键信息和数据
    3. 整理成结构化的研究报告
    输出格式：主题、核心发现、技术细节、参考来源""",
    tools=[search_tool],
)

# Writer Agent
writer = Agent(
    name="writer",
    model="gemini-pro-latest",
    instruction="""你是一个资深技术博客作家。你的任务是：
    1. 根据研究报告撰写深度技术文章
    2. 文章结构：背景介绍 → 核心概念 → 代码示例 → 实战指南 → 总结展望
    3. 字数要求：5000-10000字
    4. 风格：程序员视角，实用主义，有独特观点
    5. 包含完整的代码示例""",
    tools=[save_tool],
)

# Editor Agent
editor = Agent(
    name="editor",
    model="gemini-pro-latest",
    instruction="""你是一个技术编辑。你的任务是：
    1. 检查文章的技术准确性
    2. 优化文章结构和可读性
    3. 修正语法和表达问题
    4. 确保代码示例的正确性
    输出修改后的完整文章""",
)

4.4 组装工作流

# workflow.py
from google.adk import Workflow
from google.adk.workflow import node
from agents import researcher, writer, editor

# 定义工作流
blog_workflow = Workflow(
    name="blog_generator",
    edges=[
        ("START", "research"),
        ("research", "write"),
        ("write", "edit"),
        ("edit", "END"),
    ],
    agents={
        "research": researcher,
        "write": writer,
        "edit": editor,
    },
)

# 运行工作流
async def generate_blog(topic: str):
    result = await blog_workflow.run(
        input={"topic": topic},
        config={
            "max_retries": 3,
            "timeout_seconds": 300,
        }
    )
    return result

4.5 运行和监控

# main.py
import asyncio
from workflow import generate_blog

async def main():
    topic = "Google ADK 2.0 图工作流深度解析"

    print(f"开始生成博客: {topic}")

    result = await generate_blog(topic)

    print(f"生成完成!")
    print(f"标题: {result['title']}")
    print(f"字数: {result['word_count']}")
    print(f"保存路径: {result['file_path']}")

if __name__ == "__main__":
    asyncio.run(main())

五、ADK 2.0 vs 竞品框架：深度对比

5.1 ADK vs LangChain/LangGraph

5.2 ADK vs CrewAI

5.3 ADK vs OpenAI Agent SDK

六、ADK 2.0 的部署与运维

6.1 本地开发

# 使用 ADK CLI 启动开发服务器
adk web

# 这会启动一个 Web 界面，可以：
# - 可视化查看 Agent 图结构
# - 实时测试 Agent 交互
# - 查看执行日志和追踪

6.2 部署到 Google Cloud

# 一键部署到 Agent Runtime
adk deploy agent-runtime \
    --project=my-gcp-project \
    --region=us-central1 \
    --agent=my_agent

# 或部署到 Cloud Run
adk deploy cloud-run \
    --project=my-gcp-project \
    --region=us-central1 \
    --agent=my_agent

6.3 可观测性

ADK 2.0 内置了完整的可观测性支持：

• Logging：结构化日志，自动记录 Agent 决策过程
• Metrics：Token 使用量、延迟、错误率等关键指标
• Traces：分布式追踪，完整还原 Agent 执行链路

# 配置可观测性
from google.adk.observability import TracingConfig

config = TracingConfig(
    exporter="cloud_trace",  # 或 "jaeger", "zipkin"
    sample_rate=1.0,
    include_tool_outputs=True,
)

七、从 1.x 迁移到 2.0：必须知道的 Breaking Changes

如果你正在使用 ADK 1.x，迁移到 2.0 需要注意以下几点：

7.1 Event Schema 变更

ADK 2.0 在 Event 对象中新增了 node_info 和 output 字段。如果你有自定义的 Session 存储（如 SQL 数据库），需要更新数据库 Schema。

# 需要新增的字段
class Event:
    node_info: dict  # 新增：图节点信息
    output: Any      # 新增：工作流输出
    # ... 原有字段

7.2 Agent 执行方式变更

Agent 现在是 BaseNode 的子类，不再是独立执行器。如果你重写了 _run_async_impl() 或 generate_content()，需要迁移到 Callback 机制。

# 1.x 的方式（已废弃）
class MyAgent(Agent):
    async def _run_async_impl(self, ctx, input):
        # 自定义逻辑
        pass

# 2.0 的方式
class MyAgent(Agent):
    def __init__(self):
        super().__init__(
            before_agent_callback=self.before_run,
            after_agent_callback=self.after_run,
        )

    async def before_run(self, ctx):
        # 前置逻辑
        pass

    async def after_run(self, ctx, result):
        # 后置逻辑
        pass

7.3 错误处理变更

ADK 2.0 引入了自动重试机制。如果你在工具代码中有 try...except 块，需要确保异常能够正确传播到框架层。

# 不推荐（会阻止框架的自动重试）
def my_tool(input):
    try:
        result = do_something(input)
    except Exception as e:
        return f"Error: {e}"

# 推荐（让框架处理异常）
def my_tool(input):
    result = do_something(input)  # 异常自动传播
    return result

八、ADK 生态系统与社区

8.1 五语言 SDK

ADK 2.0 是目前唯一提供五种语言 SDK 的 Agent 框架：

8.2 模型支持

ADK 原生支持 Gemini，同时通过适配器支持多种模型：

• Google：Gemini, Gemma
• Anthropic：Claude
• 本地模型：Ollama, vLLM, LiteLLM
• 企业级：Google Cloud Agent Platform, Apigee AI Gateway

8.3 Agents CLI

ADK 提供了 agents-cli 工具，支持：

• adk init：初始化 Agent 项目
• adk web：启动本地开发界面
• adk deploy：一键部署
• adk evaluate：运行评估测试
• adk debug：调试 Agent 执行

8.4 社区

• GitHub：活跃的开源社区，定期发布更新
• Community Calls：核心工程团队的定期技术分享
• 中文文档：adk.wiki 提供完整中文翻译

九、ADK 2.0 的适用场景与局限性

9.1 最佳适用场景

1. 企业级 Agent 应用：需要审计、合规、可恢复的生产环境
2. 多 Agent 协作系统：复杂的多步骤业务流程
3. 跨组织 Agent 互操作：需要标准化通信协议的场景
4. 长期运行任务：需要断点恢复和状态持久化的场景
5. 多语言技术栈：团队使用不同编程语言的场景

9.2 当前局限性

1. Google 生态绑定：虽然支持多种模型，但最佳体验仍在 Google Cloud
2. 社区成熟度：相比 LangChain，社区规模和生态丰富度还有差距
3. 文档覆盖：部分高级特性的文档还不够完善
4. TypeScript/Java/Kotlin 2.0 尚未发布：目前只有 Python 和 Go 有 2.0 版本

十、总结与展望

ADK 2.0 的发布标志着 Agent 框架从「能用」到「能用在生产」的转变。它的图工作流引擎解决了确定性问题，协作智能体解决了编排问题，A2A 协议解决了互操作问题，一键部署解决了运维问题。

几个关键判断：

1. 图工作流将成为标配：ADK 2.0 的图工作流理念会被其他框架借鉴，成为 Agent 编排的标准范式
2. A2A 协议有望普及：随着更多框架和平台支持 A2A，Agent 之间的互操作将不再是问题
3. 多语言 SDK 是差异化优势：对于大型企业来说，能够用自己熟悉的语言构建 Agent 是一个重要的决策因素
4. Google Cloud 生态整合：ADK 与 Google Cloud 的深度整合将吸引更多企业用户

给开发者的建议：

• 如果你在构建生产级 Agent 应用，ADK 2.0 值得认真评估
• 如果你已经在用 LangChain，可以关注 ADK 的图工作流理念，但不必急于迁移
• 如果你是 Google Cloud 用户，ADK 2.0 几乎是必选项
• 如果你需要跨语言 Agent 互操作，ADK 是目前最好的选择

AI Agent 的时代才刚刚开始。ADK 2.0 不是终点，而是通往更智能、更可靠、更开放的 Agent 生态的一个重要里程碑。

参考资源：

• ADK 官方文档：https://adk.dev
• ADK 中文文档：https://adk.wiki
• ADK Python GitHub：https://github.com/google/adk-python
• ADK Go GitHub：https://github.com/google/adk-go
• A2A 协议官网：https://a2a-protocol.org