1. 项目概述为什么我们需要一个“代码优先”的AI智能体框架如果你和我一样在过去一两年里尝试过构建基于大语言模型的AI应用大概率经历过这样的场景一开始兴致勃勃用LangChain或者AutoGen快速搭了个原型感觉智能体Agent无所不能。但随着项目深入你开始头疼了——工具调用逻辑像黑盒调试起来全靠打印日志想实现一个复杂的多智能体协作流程代码结构迅速变得混乱不堪好不容易写完了却发现部署到生产环境又是一堆兼容性问题。这时候你可能会想要是有个框架能像写普通Python服务一样用清晰的代码结构、可测试的模块和标准的部署流程来构建智能体那该多好。这就是Google开源的Agent Development KitADK要解决的核心问题。它不是一个试图用配置文件或声明式语言“魔法般”生成智能体的工具而是一个彻头彻尾的“代码优先”Python框架。它的设计哲学非常明确将软件工程的最佳实践——模块化、可测试性、版本控制、清晰的抽象——引入到AI智能体的开发中。这意味着你构建的不仅仅是一个能回答问题的聊天机器人而是一个结构清晰、易于维护和扩展的软件系统只不过这个系统的核心组件是能够理解、推理和调用工具的AI智能体。ADK的定位很务实它不绑定任何特定的云服务虽然与Google Cloud的集成很顺畅也不强制你使用某一家的大模型尽管对Gemini的优化做得最好。它提供了一套基础构件让你能以编程的方式定义智能体的行为、工具、记忆以及它们之间的协作关系。无论是构建一个简单的、能联网搜索的客服助手还是一个由多个专业智能体如分析员、执行员、审核员组成的复杂决策系统ADK都试图让这个过程变得可控、可预测。接下来我们就深入拆解一下如何用ADK从零开始构建一个真正可用的智能体系统。2. 核心设计理念与架构拆解ADK如何让智能体开发“工程化”2.1 “代码优先”意味着什么很多智能体框架强调“低代码”或“无代码”通过YAML或图形界面进行配置。ADK反其道而行之坚持“代码优先”。这背后的逻辑是当智能体逻辑变得复杂时配置文件的表达能力会迅速达到瓶颈而代码的灵活性、可组合性和可调试性是无可替代的。举个例子假设你想让一个智能体在调用“发送邮件”工具前必须先经过一个“风险审核”智能体的批准。在纯配置驱动的框架里实现这种带条件的、动态的工作流可能非常棘手甚至需要框架提供特定的语法支持。而在ADK中这就是一段清晰的Python逻辑你可以定义一个父级协调员智能体Coordinator在它的执行逻辑中先调用审核子智能体根据其返回结果决定是否调用邮件发送工具。整个决策流程、状态管理、错误处理都可以用你熟悉的Python代码来编写和测试。这种设计带来了几个直接好处可测试性你可以像测试普通Python函数一样为智能体的工具调用逻辑、多智能体间的消息传递编写单元测试和集成测试。版本控制友好你的智能体逻辑就是.py文件Git可以完美地管理其变更历史、分支和合并。强大的IDE支持代码补全、类型提示、跳转到定义、实时语法检查……这些现代软件开发中提升效率的利器在ADK开发中都能用上。无缝集成现有代码库你的业务逻辑、数据访问层、第三方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 githttps://github.com/google/adk-python.gitmain但生产环境请务必使用稳定版。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( nameweb_researcher, modelgemini-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(agentsearch_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)代码解读与实操要点模型选择model参数指定使用哪个大模型。ADK默认与Gemini API集成你可以使用gemini-2.0-flash、gemini-2.0-pro等。确保你的API密钥有权限访问所选模型。指令设计instruction是智能体的“人格”和“行为准则”。写得越具体、越有步骤性智能体的表现就越可控。这里我们明确规定了使用搜索工具的条件。工具集成tools[google_search]这一行就将搜索能力赋予了智能体。当模型认为需要搜索时它会生成一个结构化的工具调用请求ADK引擎会捕获这个请求执行真正的搜索函数并将结果返回给模型由模型整合成最终回答。会话管理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信息收集员, modelgemini-2.0-flash-exp, instruction你负责收集具体的旅行信息如航班和酒店。根据用户需求调用相应的查询工具并返回原始数据。确保信息的准确性和时效性。, description专门负责查询航班、酒店等具体信息的智能体。, tools[fetch_flight_info, fetch_hotel_info], ) budget_analyst LlmAgent( name预算分析师, modelgemini-2.0-flash-exp, instruction你负责财务部分。根据信息收集员提供的航班、酒店价格以及用户给出的每日开销预估和天数计算总预算。给出清晰的预算 breakdown。, description负责计算和分析旅行预算的智能体。, tools[calculate_budget], ) # 定义父级协调员智能体 travel_coordinator LlmAgent( name旅行规划协调员, modelgemini-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(agenttravel_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())多智能体协作的核心机制任务分解协调员智能体travel_coordinator并不直接拥有查询工具或计算工具。它通过分析用户请求理解到需要“查询信息”和“计算预算”两个子任务。内部调度当协调员决定将一个子任务交给某个子智能体时ADK引擎会在内部创建一个“子会话”。协调员将任务描述作为消息发送给子智能体如“请查询上海下月15号左右的航班信息”。子智能体执行子智能体在其自己的会话上下文中运行调用它自己的工具生成结果。结果汇总子智能体的执行结果可能是文本或结构化数据被返回给协调员。协调员收集所有子任务的结果。综合输出协调员利用所有子结果结合自己的指令生成最终回复给用户。在这个过程中用户只与协调员交互完全感知不到背后多个智能体的分工与协作。这种架构使得系统非常模块化你可以独立修改信息收集员的工具或替换一个更擅长分析的预算分析师而不会影响协调员和其他部分。实操心得设计多智能体系统时关键在于清晰定义每个智能体的“单一职责”和它们之间的“契约”即输入输出的数据格式。协调员的指令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_modeConfirmationMode.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邮件助手, modelgemini-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( agentemail_agent, confirmation_providerCliConfirmationProvider() ) 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( nameproduction_assistant, modelgemini-2.0-flash-exp, instructionYou are a helpful production assistant., tools[google_search], ) # 创建FastAPI应用 app create_app(agentagent, 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应用容器化毫无特殊之处。创建一个DockerfileFROM 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的控制台中进行统一管理和部署。部署注意事项密钥管理永远不要将API密钥硬编码在代码中。使用环境变量、Cloud Secret Manager等安全服务来管理。会话存储生产环境不能使用InMemorySessionService。你需要实现或使用一个持久化的会话服务将会话数据存储在数据库如Firestore、Cloud SQL中。超时与重试在云函数或Run中注意设置合理的请求超时时间。对于可能长时间运行的智能体任务考虑采用异步任务队列模式。监控与日志确保集成应用的日志记录并配置监控告警关注智能体的延迟、错误率和工具调用成功率。5. 常见问题、排查技巧与生态资源5.1 开发与调试常见问题速查表问题现象可能原因排查步骤与解决方案安装失败提示依赖冲突Python版本不兼容或与其他包冲突。1. 确认Python版本3.10。2. 在新虚拟环境中安装。3. 尝试先安装pip install google-adk --no-deps再手动安装核心依赖。运行时报API key not found未正确设置Gemini API密钥。1. 检查环境变量GOOGLE_API_KEY是否已设置且有效。2. 在代码中通过os.environ[GOOGLE_API_KEY] key设置仅限开发。3. 如果使用Google Cloud确认已启用相关API且服务账号有权限。智能体不调用工具1. 指令未明确要求。2. 工具描述不清晰。3. 模型认为不需要。1. 在instruction中明确写出“当你需要XX信息时请使用YY工具”。2. 优化工具的description和参数描述使其更易被模型理解。3. 在开发UI中查看模型中间推理过程判断问题所在。工具调用参数错误模型对参数格式理解有误。1. 使用tool装饰器时为函数参数添加明确的类型注解如str,int,List[str]。2. 在工具描述中举例说明参数格式。3. 考虑使用Pydantic模型来定义复杂的参数结构。多智能体协作卡住协调员指令不清晰或子智能体返回结果格式不符合协调员预期。1. 为每个子智能体定义清晰的输入输出规范。2. 在协调员指令中明确说明如何解析和使用子智能体的返回结果。3. 使用开发UI的单步调试功能观察消息在智能体间的传递过程。部署后性能差/延迟高1. 模型端点慢。2. 网络延迟。3. 工具调用外部API慢。1. 考虑使用更快的模型变体如Flash版。2. 将服务部署在离你的用户或所调用的API较近的区域。3. 为耗时的工具调用实现异步或缓存机制。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提供了一条连贯的路径。