Google ADK：代码优先的AI智能体开发框架实战指南

1. 项目概述：为什么我们需要一个“代码优先”的AI智能体框架？

如果你和我一样，在过去一两年里尝试过构建基于大语言模型的AI应用，大概率经历过这样的场景：一开始兴致勃勃，用LangChain或者AutoGen快速搭了个原型，感觉智能体（Agent）无所不能。但随着项目深入，你开始头疼了——工具调用逻辑像黑盒，调试起来全靠打印日志；想实现一个复杂的多智能体协作流程，代码结构迅速变得混乱不堪；好不容易写完了，却发现部署到生产环境又是一堆兼容性问题。这时候你可能会想，要是有个框架能像写普通Python服务一样，用清晰的代码结构、可测试的模块和标准的部署流程来构建智能体，那该多好。

这就是Google开源的Agent Development Kit（ADK）要解决的核心问题。它不是一个试图用配置文件或声明式语言“魔法般”生成智能体的工具，而是一个彻头彻尾的“代码优先”Python框架。它的设计哲学非常明确：将软件工程的最佳实践——模块化、可测试性、版本控制、清晰的抽象——引入到AI智能体的开发中。这意味着，你构建的不仅仅是一个能回答问题的聊天机器人，而是一个结构清晰、易于维护和扩展的软件系统，只不过这个系统的核心组件是能够理解、推理和调用工具的AI智能体。

ADK的定位很务实：它不绑定任何特定的云服务（虽然与Google Cloud的集成很顺畅），也不强制你使用某一家的大模型（尽管对Gemini的优化做得最好）。它提供了一套基础构件，让你能以编程的方式定义智能体的行为、工具、记忆以及它们之间的协作关系。无论是构建一个简单的、能联网搜索的客服助手，还是一个由多个专业智能体（如分析员、执行员、审核员）组成的复杂决策系统，ADK都试图让这个过程变得可控、可预测。接下来，我们就深入拆解一下，如何用ADK从零开始，构建一个真正可用的智能体系统。

2. 核心设计理念与架构拆解：ADK如何让智能体开发“工程化”？

2.1 “代码优先”意味着什么？

很多智能体框架强调“低代码”或“无代码”，通过YAML或图形界面进行配置。ADK反其道而行之，坚持“代码优先”。这背后的逻辑是：当智能体逻辑变得复杂时，配置文件的表达能力会迅速达到瓶颈，而代码的灵活性、可组合性和可调试性是无可替代的。

举个例子：假设你想让一个智能体在调用“发送邮件”工具前，必须先经过一个“风险审核”智能体的批准。在纯配置驱动的框架里，实现这种带条件的、动态的工作流可能非常棘手，甚至需要框架提供特定的语法支持。而在ADK中，这就是一段清晰的Python逻辑：你可以定义一个父级协调员智能体（Coordinator），在它的执行逻辑中，先调用审核子智能体，根据其返回结果决定是否调用邮件发送工具。整个决策流程、状态管理、错误处理，都可以用你熟悉的Python代码来编写和测试。

这种设计带来了几个直接好处：

1. 可测试性：你可以像测试普通Python函数一样，为智能体的工具调用逻辑、多智能体间的消息传递编写单元测试和集成测试。
2. 版本控制友好：你的智能体逻辑就是.py文件，Git可以完美地管理其变更历史、分支和合并。
3. 强大的IDE支持：代码补全、类型提示、跳转到定义、实时语法检查……这些现代软件开发中提升效率的利器，在ADK开发中都能用上。
4. 无缝集成现有代码库：你的业务逻辑、数据访问层、第三方API客户端，都可以直接作为模块导入，并被智能体作为工具调用，无需为了适配框架而进行大幅重构。

2.2 核心架构组件解析

ADK的架构围绕几个核心抽象构建，理解它们之间的关系是高效使用的关键。

智能体（Agent）：这是最核心的抽象。一个智能体封装了与大模型交互的指令、可用的工具列表、记忆（会话历史）以及行为逻辑。ADK提供了LlmAgent作为基础实现，它负责处理与大模型的对话、解析模型响应、调度工具执行。你可以通过继承BaseAgent来创建完全自定义的智能体，实现更复杂的决策逻辑。

工具（Tool）：智能体能力的延伸。一个工具本质上是一个Python函数，附加上描述其功能和参数的元数据。ADK的工具生态非常丰富：

• 预构建工具：如google_search，开箱即用。
• 函数即工具：用@tool装饰器将任何Python函数转化为工具。
• OpenAPI工具：可以直接加载一个服务的OpenAPI规范，自动生成对应的工具集，让智能体能够操作该服务的所有API。
• MCP工具：支持Model Context Protocol，这是一种新兴的、标准化的工具提供方式，未来可能成为工具互操作性的关键。

引擎（Engine）：智能体系统的运行时环境。它管理智能体的生命周期、处理输入输出、维护会话状态，并提供了将智能体部署为服务的桥梁。AgentEngine是核心引擎，它不仅能运行本地智能体，还能通过集成A2A协议，与远程的、可能由不同语言实现的智能体进行通信。

会话（Session）：代表一次与智能体的交互上下文。它保存了完整的对话历史、工具调用记录和自定义的会话状态。ADK一个强大的新特性是“回滚”（Rewind），它允许你将会话状态重置到某次调用之前，这对于调试复杂的、状态依赖的交互流程极其有用。你可以精确地回到错误发生前的那一刻，修改输入或智能体逻辑，然后重新执行，而无需从头开始。

服务与部署：ADK内置了基于FastAPI的Web服务器，可以将你的智能体快速暴露为HTTP端点。更重要的是，它提供了与Google Cloud Vertex AI Agent Engine和Cloud Run的一键式集成，让从开发到生产的部署路径非常顺畅。AgentEngineSandboxCodeExecutor这类新组件，则展示了如何安全地执行智能体生成的代码，这是实现“代码解释器”类高级能力的基础。

3. 从零开始：构建你的第一个ADK智能体

理论说得再多，不如动手写一行代码。让我们从一个最简单的例子开始，逐步增加复杂度，体会ADK的开发流程。

3.1 环境准备与安装

首先，确保你有一个Python 3.10+的环境。创建一个新的虚拟环境是良好的习惯。

python -m venv adk-env source adk-env/bin/activate # Linux/macOS # 或 adk-env\Scripts\activate # Windows

安装ADK。对于大多数用户，直接从PyPI安装稳定版即可。

pip install google-adk

注意：ADK会安装一些依赖，如Google的通用AI库google.generativeai。如果你计划使用Gemini模型，还需要额外设置API密钥。通常通过环境变量GOOGLE_API_KEY来配置。如果你需要用到最新的、尚未发布的特性，可以安装开发版（pip install git+https://github.com/google/adk-python.git@main），但生产环境请务必使用稳定版。

3.2 定义单智能体：一个联网搜索助手

我们来构建一个经典的例子：一个能回答实时问题的智能体，当它不知道答案时，会自动使用Google搜索。

# search_assistant.py import os from google.adk.agents import LlmAgent from google.adk.tools import google_search from google.adk.sessions import InMemorySessionService # 1. 设置API密钥（假设你已申请Gemini API Key） os.environ['GOOGLE_API_KEY'] = 'YOUR_API_KEY_HERE' # 2. 定义智能体 search_agent = LlmAgent( name="web_researcher", model="gemini-2.0-flash-exp", # 选用一个合适的Gemini模型 instruction="""你是一个专业的网络研究员。你的目标是准确、高效地回答用户的问题。 遵循以下步骤： 1. 首先，理解用户问题的核心。 2. 如果你已经拥有足够的知识来给出准确答案，请直接回答。 3. 如果问题涉及实时信息、最新事件、具体数据或你不确定的内容，你必须使用`google_search`工具进行查询。 4. 根据搜索得到的信息，组织一个清晰、有条理的回答，并注明信息来源（如果适用）。 永远保持友好和乐于助人。""", description="一个能够使用Google搜索来获取实时信息的研究助手。", tools=[google_search], # 将工具赋予智能体 ) # 3. 创建会话并运行 session_service = InMemorySessionService() session = session_service.create_session(agent=search_agent) try: # 第一次提问，可能不需要搜索 response = session.send_message("什么是量子计算？") print(f"用户: 什么是量子计算？") print(f"助手: {response.messages[-1].content}\n") # 第二次提问，需要实时信息，会触发搜索 response = session.send_message("今天纽约的天气怎么样？") print(f"用户: 今天纽约的天气怎么样？") # 查看响应和工具调用历史 for msg in response.messages: if msg.tool_calls: print(f"[工具调用] {msg.tool_calls}") if msg.content: print(f"助手: {msg.content}") finally: session_service.delete_session(session.id)

代码解读与实操要点：

1. 模型选择：model参数指定使用哪个大模型。ADK默认与Gemini API集成，你可以使用gemini-2.0-flash、gemini-2.0-pro等。确保你的API密钥有权限访问所选模型。
2. 指令设计：instruction是智能体的“人格”和“行为准则”。写得越具体、越有步骤性，智能体的表现就越可控。这里我们明确规定了使用搜索工具的条件。
3. 工具集成：tools=[google_search]这一行就将搜索能力赋予了智能体。当模型认为需要搜索时，它会生成一个结构化的工具调用请求，ADK引擎会捕获这个请求，执行真正的搜索函数，并将结果返回给模型，由模型整合成最终回答。
4. 会话管理：InMemorySessionService在内存中管理会话，适合开发和测试。在生产环境中，你可能需要将其替换为持久化的服务（如数据库存储）。

运行这个脚本，你会看到智能体对于第一个常识性问题可能直接回答，而对于第二个需要实时信息的问题，则会先显示工具调用记录，再给出包含搜索结果的回答。

3.3 构建多智能体系统：任务分解与协作

单智能体能力有限。复杂任务通常需要分解，由多个各司其职的智能体协作完成。ADK通过sub_agents属性优雅地支持了这种层级结构。

假设我们要构建一个“旅行规划系统”，它包含一个协调员、一个信息收集员和一个预算分析师。

# travel_planner_system.py from google.adk.agents import LlmAgent from google.adk.tools import tool from datetime import datetime import asyncio # 首先，定义一些自定义工具（模拟） @tool def fetch_flight_info(destination: str, date: str) -> str: """模拟查询航班信息。返回航班号和价格。""" # 这里应该是调用真实API的代码 return f"找到航班：CA123 前往{destination}，日期{date}，价格 ¥1200。" @tool def fetch_hotel_info(city: str, check_in: str, nights: int) -> str: """模拟查询酒店信息。""" return f"找到酒店：{city}市中心酒店，入住{check_in}，{nights}晚，总价 ¥800。" @tool def calculate_budget(flight_cost: float, hotel_cost: float, daily_expense: float, days: int) -> str: """计算旅行总预算。""" total = flight_cost + hotel_cost + (daily_expense * days) return f"估算总预算：机票¥{flight_cost} + 酒店¥{hotel_cost} + {days}天日常开销¥{daily_expense*days} = 总计¥{total}。" # 定义子智能体 info_gatherer = LlmAgent( name="信息收集员", model="gemini-2.0-flash-exp", instruction="你负责收集具体的旅行信息，如航班和酒店。根据用户需求调用相应的查询工具，并返回原始数据。确保信息的准确性和时效性。", description="专门负责查询航班、酒店等具体信息的智能体。", tools=[fetch_flight_info, fetch_hotel_info], ) budget_analyst = LlmAgent( name="预算分析师", model="gemini-2.0-flash-exp", instruction="你负责财务部分。根据信息收集员提供的航班、酒店价格，以及用户给出的每日开销预估和天数，计算总预算。给出清晰的预算 breakdown。", description="负责计算和分析旅行预算的智能体。", tools=[calculate_budget], ) # 定义父级协调员智能体 travel_coordinator = LlmAgent( name="旅行规划协调员", model="gemini-2.0-pro-exp", # 协调员可以用更强一点的模型 instruction="""你是旅行规划系统的总指挥。你的工作流程是： 1. 与用户沟通，明确旅行目的地、时间、人数、预算偏好等核心需求。 2. 将‘信息查询’子任务派发给[信息收集员]。 3. 将收集到的信息，连同用户的每日开销预估，派发给[预算分析师]进行核算。 4. 综合所有信息，为用户生成一份完整的、个性化的旅行规划建议，包括行程概览、重要信息和总预算。 确保最终回答友好、完整、条理清晰。""", description="协调信息收集和预算分析，生成最终旅行规划的智能体。", sub_agents=[info_gatherer, budget_analyst], # 关键：指定子智能体 ) # 使用系统 async def main(): from google.adk.sessions import InMemorySessionService session_service = InMemorySessionService() session = session_service.create_session(agent=travel_coordinator) user_request = "我想下个月15号去上海玩3天，两个人，对住宿要求中等，帮我规划一下，并估算一下预算，假设每天吃喝玩乐大概500元。" print(f"用户: {user_request}\n") response = await session.asend_message(user_request) # 打印整个交互过程，可以看到多智能体的协作 for i, msg in enumerate(response.messages): role = msg.role.capitalize() if msg.tool_calls: print(f"[步骤{i}] {role} 调用了工具: {msg.tool_calls}") if msg.content and msg.role == 'model': # 主要看模型的回复 print(f"[最终规划] {msg.content}") elif msg.content: print(f"[内部消息] {role}: {msg.content[:100]}...") # 截断内部长消息 session_service.delete_session(session.id) if __name__ == "__main__": asyncio.run(main())

多智能体协作的核心机制：

1. 任务分解：协调员智能体（travel_coordinator）并不直接拥有查询工具或计算工具。它通过分析用户请求，理解到需要“查询信息”和“计算预算”两个子任务。
2. 内部调度：当协调员决定将一个子任务交给某个子智能体时，ADK引擎会在内部创建一个“子会话”。协调员将任务描述作为消息发送给子智能体（如“请查询上海下月15号左右的航班信息”）。
3. 子智能体执行：子智能体在其自己的会话上下文中运行，调用它自己的工具，生成结果。
4. 结果汇总：子智能体的执行结果（可能是文本或结构化数据）被返回给协调员。协调员收集所有子任务的结果。
5. 综合输出：协调员利用所有子结果，结合自己的指令，生成最终回复给用户。

在这个过程中，用户只与协调员交互，完全感知不到背后多个智能体的分工与协作。这种架构使得系统非常模块化：你可以独立修改信息收集员的工具，或替换一个更擅长分析的预算分析师，而不会影响协调员和其他部分。

实操心得：设计多智能体系统时，关键在于清晰定义每个智能体的“单一职责”和它们之间的“契约”（即输入输出的数据格式）。协调员的指令（instruction）是系统的“总控程序”，写得好坏直接决定了协作是否顺畅。建议先用流程图画出你期望的协作流程，再转化为各个智能体的指令。

4. 进阶实战：工具确认流、会话回滚与生产部署

4.1 为危险操作添加“人工确认”环节

智能体自动调用工具虽然强大，但也存在风险，比如发送邮件、删除数据、支付订单等。ADK提供了工具确认流，可以在工具执行前插入一个确认步骤，这个确认可以由另一个智能体（“守护智能体”）完成，或者在开发阶段，直接由开发者控制。

from google.adk.agents import LlmAgent from google.adk.tools import tool, ConfirmationMode from google.adk.confirmation import UserConfirmationProvider import getpass # 定义一个需要确认的危险工具 @tool(confirmation_mode=ConfirmationMode.REQUIRED) # 关键：标记此工具需要确认 def send_email(to: str, subject: str, body: str) -> str: """向指定收件人发送电子邮件。这是一个高风险操作，需要确认。""" # 模拟发送逻辑 print(f"[模拟] 发送邮件给 {to}， 主题：{subject}") print(f"正文：{body}") return f"邮件已成功发送至 {to}。" # 创建一个使用该工具的智能体 email_agent = LlmAgent( name="邮件助手", model="gemini-2.0-flash-exp", instruction="你是一个邮件起草助手。根据用户要求起草邮件内容，并在发送前请求确认。", tools=[send_email], ) # 创建一个用户确认提供者（在实际应用中，这里可以接入一个审批系统或管理界面） class CliConfirmationProvider(UserConfirmationProvider): async def confirm(self, tool_name: str, tool_args: dict) -> bool: print(f"\n⚠️ 工具调用等待确认：") print(f" 工具: {tool_name}") print(f" 参数: {tool_args}") answer = input("是否允许执行？(yes/no): ").strip().lower() return answer == 'yes' # 在会话中使用 async def send_email_with_confirmation(): from google.adk.sessions import InMemorySessionService session_service = InMemorySessionService() # 创建会话时注入确认提供者 session = session_service.create_session( agent=email_agent, confirmation_provider=CliConfirmationProvider() ) response = await session.asend_message("给老板发一封邮件，主题是‘项目周报’，内容问好并附上本周进度总结。") print(f"助手回复: {response.messages[-1].content}") session_service.delete_session(session.id) # 运行 import asyncio asyncio.run(send_email_with_confirmation())

当你运行这段代码时，智能体会先起草好邮件内容，并在调用send_email工具前暂停。控制台会打印出待执行的工具和参数，等待你输入“yes”确认。只有确认后，工具才会真正执行。这种机制为生产环境中的自动化流程提供了至关重要的安全阀。

4.2 利用“会话回滚”进行高效调试

调试智能体，尤其是涉及多轮复杂交互和工具调用的场景，非常耗时。ADK的“回滚”功能允许你将会话状态回退到之前的任意一步。

# 假设我们有一个会话，已经进行了几轮交互 session_id = session.id history = session.get_history() print(f"当前会话历史消息数：{len(history)}") # 假设我们在最新一轮交互中发现了错误，想回到第三步之后的状态 # 首先，找到第三步消息的ID。在实际中，你可能需要根据内容或顺序来定位。 target_message_id = history[2].id # 索引2代表第三条消息 # 回滚到该消息之前的状态 rewound_session = session_service.rewind_session(session_id, target_message_id) print(f"回滚后会话历史消息数：{len(rewound_session.get_history())}") # 现在，rewound_session 的状态就回到了执行第三条消息之前。 # 你可以修改接下来的用户输入，或者甚至替换智能体的工具，然后重新执行。 new_response = await rewound_session.asend_message("请换一种方式解释刚才的概念。")

这个功能在以下场景中不可或缺：

• 工具调用错误：智能体用错误参数调用了工具，你可以回滚到调用前，修正提示词或工具描述，然后重试。
• 模型输出不佳：模型的回答偏离了方向，你可以回滚并尝试换一种问法。
• 测试不同分支：对于决策型智能体，你可以回滚到决策点，输入不同的条件，观察智能体如何走向不同的分支。

它本质上是为智能体开发提供了“时间旅行调试器”，极大地提升了迭代效率。

4.3 部署到生产环境：从本地服务到云上托管

开发完成后，你需要将智能体部署为一个可被其他系统调用的服务。ADK让这一步变得非常简单。

第一步：创建FastAPI应用

ADK内置了与FastAPI的集成。创建一个app.py文件：

# app.py from fastapi import FastAPI from google.adk.servers.fastapi import create_app from google.adk.agents import LlmAgent from google.adk.tools import google_search import os os.environ['GOOGLE_API_KEY'] = 'YOUR_API_KEY' # 定义你的智能体 agent = LlmAgent( name="production_assistant", model="gemini-2.0-flash-exp", instruction="You are a helpful production assistant.", tools=[google_search], ) # 创建FastAPI应用 app = create_app(agent=agent, path="/chat") # 智能体服务将挂在 /chat 端点 # 你可以添加其他自定义API路由 @app.get("/health") def health_check(): return {"status": "healthy"}

第二步：本地运行测试

pip install "uvicorn[standard]" uvicorn app:app --reload --host 0.0.0.0 --port 8080

现在，你的智能体就有了一个标准的HTTP API。你可以用curl或Postman向http://localhost:8080/chat发送POST请求（消息体格式参考ADK文档）进行交互。

第三步：容器化与部署

ADK应用就是标准的FastAPI应用，容器化毫无特殊之处。创建一个Dockerfile：

FROM python:3.11-slim WORKDIR /app COPY requirements.txt . RUN pip install --no-cache-dir -r requirements.txt COPY . . CMD ["uvicorn", "app:app", "--host", "0.0.0.0", "--port", "8080"]

构建并推送到容器仓库后，你可以将其部署到任何支持容器的平台，比如Google Cloud Run，这是一个无服务器容器平台，非常适合智能体这种请求量波动大的服务。

gcloud run deploy my-adk-agent \ --image gcr.io/YOUR_PROJECT/my-adk-agent:latest \ --platform managed \ --region us-central1 \ --allow-unauthenticated # 根据你的安全需求调整

第四步：进阶部署 - Vertex AI Agent Engine

对于更复杂、需要与Google云服务深度集成，或希望使用托管版本来处理扩展、监控等问题的场景，可以部署到Vertex AI Agent Engine。这需要一些额外的配置，但ADK提供了相应的集成模块，可以将你的智能体定义打包成符合Agent Engine规范的格式，从而在Google Cloud的控制台中进行统一管理和部署。

部署注意事项：

1. 密钥管理：永远不要将API密钥硬编码在代码中。使用环境变量、Cloud Secret Manager等安全服务来管理。
2. 会话存储：生产环境不能使用InMemorySessionService。你需要实现或使用一个持久化的会话服务，将会话数据存储在数据库（如Firestore、Cloud SQL）中。
3. 超时与重试：在云函数或Run中，注意设置合理的请求超时时间。对于可能长时间运行的智能体任务，考虑采用异步任务队列模式。
4. 监控与日志：确保集成应用的日志记录，并配置监控告警，关注智能体的延迟、错误率和工具调用成功率。

5. 常见问题、排查技巧与生态资源

5.1 开发与调试常见问题速查表

5.2 充分利用ADK生态

• 官方文档：https://google.github.io/adk-docs是你的第一站，包含了从概念到API的完整指南。
• 示例仓库：https://github.com/google/adk-samples提供了大量从简单到复杂的示例代码，是学习的最佳实践库。
• 社区仓库：https://github.com/google/adk-python-community这里有社区贡献的各种第三方工具集成、部署脚本和扩展组件。如果你实现了某个通用服务的工具（如Slack、Notion），可以考虑贡献到这里。
• 开发UI：运行adk ui命令可以启动一个本地Web界面，用于可视化地测试、调试你的智能体，观察完整的思维链和工具调用流程，这对调试复杂逻辑至关重要。
• 评估框架：使用adk eval命令，你可以基于定义的评估集对智能体的表现进行自动化测试和评分，确保智能体行为的稳定性和准确性。

构建AI智能体应用，正从早期的“玩具项目”阶段走向真正的“生产级”系统。ADK通过其代码优先、工程化友好的设计，为开发者提供了搭建这类系统所需的坚实脚手架。它不试图隐藏复杂性，而是通过清晰的抽象和模块化设计，让你能够掌控复杂性。从定义一个简单的工具，到编排一个多智能体协作网络，再到将其安全、可靠地部署到云端，ADK提供了一条连贯的路径。