Google ADK：代码优先的AI Agent开发框架，构建可维护的智能体应用

1. 项目概述为什么我们需要一个“代码优先”的Agent框架如果你和我一样在过去一两年里尝试过构建AI Agent应用大概率经历过这样的场景一开始兴致勃勃用LangChain或者AutoGen这类流行框架快速搭了个原型感觉Agent能理解指令、调用工具简直无所不能。但随着项目深入你开始头疼了——想给Agent加个自定义的状态管理框架的抽象层太厚不知道从哪下手。想把Agent部署到生产环境发现它和你的Web服务框架格格不入。想写个单元测试却发现Agent的每次调用都充满了不确定性测试用例根本写不下去。这就是为什么当我看到Google开源的Agent Development KitADK时会感到眼前一亮。它没有试图做一个“大而全”的魔法黑盒而是选择了一条更“工程师友好”的路代码优先。ADK的核心哲学很明确把Agent当成软件来开发。这意味着你可以用熟悉的Python类、函数、设计模式来构建Agent享受完整的类型提示、依赖注入、模块化设计和可测试性。它不是一个试图替代你整个应用架构的“框架”而是一个可以无缝嵌入到你现有Python项目中的“工具包”。简单来说ADK适合这样的你你已经不满足于仅仅“调通”一个Agent对话而是希望构建可靠、可维护、可扩展的智能体应用并且你希望对这个系统的每一个环节都有清晰的控制力。无论是构建一个简单的客服助手还是一个由数十个专业Agent协同工作的复杂业务系统ADK提供的这套基于软件工程理念的范式都能让你事半功倍。2. 核心设计理念当软件工程遇见智能体ADK的设计并非凭空而来它深刻反映了从“Prompt工程”到“Agent工程”的范式转变。早期的Agent开发更像是一种实验我们通过精心设计的Prompt和有限的工具调用引导大模型完成特定任务。但当任务变得复杂、系统需要长期运行和维护时这种方式的短板就暴露无遗逻辑散落在Prompt字符串里、状态难以追踪、错误处理机制薄弱、几乎无法进行自动化测试。ADK的解决方案是引入成熟的软件工程原则。让我们拆解一下它的几个核心设计思想2.1 明确的抽象与分层ADK没有把“Agent”变成一个模糊的、无所不包的概念。相反它进行了清晰的职责分离Agent代理 定义了核心的“角色”、指令instruction以及它可以使用的工具Tools列表。它负责决策“做什么”。Engine引擎 负责执行。它接收Agent的决策调用模型处理工具的执行并管理整个会话Session的状态流转。你可以把Engine看作Agent的“运行时环境”。Session会话 封装了一次完整交互的上下文。所有的消息历史、工具调用记录、自定义状态都保存在Session对象中。这为实现会话持久化、回滚Rewind、以及复杂的多轮交互逻辑提供了基础。Tool工具 Agent能力的延伸。ADK对工具的支持非常开放可以是简单的Python函数、基于OpenAPI规范的HTTP接口、甚至是来自Model Context ProtocolMCP服务器的工具。这种分层带来的最大好处是可测试性。你可以单独测试一个Tool函数是否工作正常可以Mock一个Engine来测试Agent的决策逻辑也可以序列化一个Session状态用于调试或复现问题。这和我们测试一个普通的Web服务或业务逻辑模块没有本质区别。2.2 状态管理的主动权交还给开发者很多框架将状态管理完全黑盒化开发者难以介入。ADK则通过Session对象将状态的读写权完全开放。你可以在Session中存储任意自定义的数据结构。例如你正在构建一个旅行规划Agent需要记住用户已经选择的航班和酒店。在其他框架中你可能需要把这些信息绞尽脑汁地塞进对话历史里。而在ADK中你可以这样做from google.adk.sessions import Session # 假设在某个工具调用后你获得了航班信息 def book_flight_tool(session: Session, destination: str): # ... 调用订票API的逻辑 ... flight_info {airline: XX, flight_no: AB123, date: 2024-01-01} # 将关键信息存入session的自定义状态中 if not hasattr(session.state, trip_plan): session.state.trip_plan {} session.state.trip_plan[flight] flight_info return fFlight to {destination} booked successfully. # 在后续的Agent指令或工具中可以直接读取 def suggest_hotel(session: Session): flight_date session.state.trip_plan.get(flight, {}).get(date) # 基于航班日期推荐酒店...这种模式使得Agent具备了“记忆”并且这份记忆的结构完全由你定义清晰可控。2.3 真正的模型无关与部署无关虽然ADK由Google推出并且与Gemini模型和Vertex AI平台集成得非常好但它从架构上就避免了绑定。Agent对象在定义时指定model参数这个参数可以是一个Gemini模型名也可以是一个符合特定接口的任何LLM客户端。理论上你可以通过实现相应的适配器轻松接入OpenAI、Anthropic或本地部署的模型。部署方面同样如此。ADK应用本质上是一个Python对象网络。你可以把它封装成一个简单的CLI脚本一个FastAPI应用或者打包进Docker容器。官方文档详细介绍了如何部署到Cloud Run或Vertex AI Agent Engine但你也完全可以把它部署在你自己的Kubernetes集群、服务器less函数或任何你能运行Python的地方。这种灵活性对于企业级应用至关重要。3. 从零开始构建你的第一个ADK Agent理论说得再多不如动手实践。让我们从一个最简单的“Hello World” Agent开始逐步增加复杂度体会ADK的开发流程。3.1 环境搭建与安装首先确保你的Python环境在3.10及以上。创建一个新的虚拟环境是良好的习惯。python -m venv adk-env source adk-env/bin/activate # Linux/macOS # 或 adk-env\Scripts\activate # Windows安装ADK的稳定版pip install google-adk同时你需要配置Gemini API的访问权限。前往 Google AI Studio 获取一个API密钥并将其设置为环境变量export GOOGLE_API_KEY你的API密钥 # Linux/macOS # 或 set GOOGLE_API_KEY你的API密钥 # Windows注意 在生产环境中请务必使用安全的密钥管理服务如Google Cloud Secret Manager切勿将密钥硬编码在代码中或提交到版本控制系统。3.2 定义单智能体一个会搜索的助手我们的第一个目标是创建一个能回答实时问题的助手当它不确定时会使用谷歌搜索。# search_assistant.py import asyncio from google import genai from google.adk.agents import Agent from google.adk.tools import google_search from google.adk.sessions import Session from google.adk.engines import AgentEngine # 1. 初始化客户端ADK内部会使用这个客户端 client genai.Client() # 2. 定义Agent search_agent Agent( nameSearchAssistant, modelgemini-2.0-flash-exp, # 指定使用的模型 instruction你是一个乐于助人的助手。你的知识截止于2024年初。 当用户询问关于近期事件、实时信息或你不确定的事情时你必须使用谷歌搜索工具来获取最新、准确的信息。 在提供答案时请引用你的信息来源。, description一个能够使用谷歌搜索来回答实时问题的助手。, tools[google_search] # 注入工具 ) # 3. 创建引擎和会话 engine AgentEngine(agentsearch_agent, clientclient) session Session() # 4. 运行交互循环 async def main(): print(Search Assistant 已启动。输入‘退出’或‘quit’来结束对话。) while True: try: user_input input(\n你: ) if user_input.lower() in [退出, quit]: break # 将用户输入添加到会话中 session.add_human_message(user_input) # 引擎执行Agent思考、决定是否调用工具、生成回复 response await engine.run(sessionsession) # 打印Agent的最终回复 print(f\n助手: {response.messages[-1].text}) except KeyboardInterrupt: break except Exception as e: print(f\n发生错误: {e}) if __name__ __main__: asyncio.run(main())运行这个脚本python search_assistant.py你就可以和一个能联网搜索的助手对话了。问它“今天纽约的天气怎么样”或者“特斯拉最新的财报是什么时候发布的”观察它如何自动触发google_search工具。关键点解析Agent类 这是智能体的“大脑”定义。instruction是其核心行为准则相当于一个系统Prompt但被结构化地管理起来。google_search工具 这是ADK提供的一个预构建工具。当Agent在思考过程中认为需要搜索时引擎会暂停文本生成转而执行这个工具函数并将工具返回的结果作为上下文继续生成回复。AgentEngine.run 这是核心的执行循环。它内部处理了与模型的通信、工具调用的调度、以及会话状态的更新。3.3 构建自定义工具赋予Agent独特能力预构建工具虽好但真正的力量在于创建自定义工具。假设我们要为Agent添加一个“计算器”工具和一个“查询数据库”工具。# custom_tools.py from typing import Annotated from pydantic import BaseModel, Field from google.adk.tools import tool # 1. 使用Pydantic定义工具输入的结构化模式 class CalculatorInput(BaseModel): expression: Annotated[str, Field(description一个合法的数学表达式例如3 5 * (2 - 1))] class QueryDBInput(BaseModel): question: Annotated[str, Field(description用自然语言描述的数据查询问题例如‘上个月销售额最高的产品是什么’)] # 2. 使用tool装饰器创建工具函数 tool(args_schemaCalculatorInput) def calculator_tool(expression: str) - str: 一个安全的计算器工具。它可以评估基本的数学表达式。 # 警告在实际生产中直接使用eval是危险的这里仅为演示。 # 应使用ast.literal_eval或专门的数学表达式解析库如numexpr。 try: # 极其简单的安全过滤生产环境需要更严格的检查 if any(c for c in expression if c.isalpha() and c not in pi ): return 错误表达式中包含不被允许的字符。 result eval(expression, {__builtins__: {}}, {}) return f计算结果: {result} except Exception as e: return f计算错误: {e} tool(args_schemaQueryDBInput) async def query_database_tool(question: str, session) - str: 一个模拟的数据库查询工具。根据自然语言问题返回模拟数据。 # 注意这个工具接收了额外的session参数。ADK会自动注入当前会话。 # 在实际应用中这里会连接你的数据库解析问题执行SQL等。 # 我们可以利用session中的状态来使查询更智能。 user_id getattr(session.state, user_id, unknown_user) # 模拟一个基于问题的查询逻辑 if 销售额 in question and 最高 in question: return f[模拟数据] 用户{user_id}根据查询‘{question}’上个月销售额最高的产品是‘智能音箱Pro’销售额为$125,430。 elif 订单 in question: return f[模拟数据] 用户{user_id}您最近有3笔待处理订单。 else: return f[模拟数据] 已根据您的问题‘{question}’执行查询返回了通用结果集。现在更新你的Agent使用这些自定义工具from custom_tools import calculator_tool, query_database_tool my_agent Agent( nameBusinessAnalyst, modelgemini-2.0-flash-exp, instruction你是一个业务分析助手。可以帮助用户计算数据也可以查询模拟的业务数据库来回答问题。, tools[calculator_tool, query_database_tool] # 注入自定义工具 )工具开发心得结构化输入 使用Pydantic模型定义输入参数这能让大模型更准确地理解如何调用工具并自动进行参数验证和类型转换。清晰的描述 工具函数的docstring和Field(description...)非常重要它们是模型理解工具用途的主要依据。描述要具体、准确。会话注入 工具函数可以声明session参数。ADK引擎会自动将当前会话对象注入进来这使得工具可以访问和修改会话状态实现有状态的工具交互。安全性 像calculator_tool中的eval是极度危险的示例。在实际开发中必须避免执行任意用户输入。应使用受限制的解析器或沙箱环境。3.4 实现多智能体协作分工与调度单个Agent能力有限复杂的任务需要分工协作。ADK通过sub_agents属性优雅地支持了多Agent系统。想象一个客服场景一个“协调员”Agent负责接待用户根据问题类型将任务分派给“技术支持”或“订单查询”专员Agent。# multi_agent_system.py from google.adk.agents import LlmAgent from google.adk.tools import tool from typing import Literal # 1. 定义专业Agent tech_support_agent LlmAgent( nameTechSupport, modelgemini-2.0-flash-exp, instruction你是技术支持专家擅长解决软件安装、错误代码和网络连接问题。请用清晰、步骤化的方式回答。如果问题超出范围请如实告知。, description处理技术类问题的专家。 ) order_agent LlmAgent( nameOrderSpecialist, modelgemini-2.0-flash-exp, instruction你是订单查询专家可以处理订单状态查询、退货申请、物流跟踪等问题。你需要先验证用户身份如订单号或邮箱。, description处理订单相关问题的专家。 ) # 2. 定义一个工具让协调员可以“呼叫”专家 tool def transfer_to_agent( target_agent_name: Literal[TechSupport, OrderSpecialist], user_query: str ) - str: 将用户问题转接给指定的专家坐席。调用此工具后对话将交由该专家主导。 # 这个工具本身不执行复杂逻辑它主要是一个信号。 # 在实际的ADK多Agent引擎中这个“转移”动作会由引擎根据会话状态和规则来处理。 # 这里返回一个结构化信息供引擎解析。 return fTRANSFER_REQUEST:{target_agent_name}:{user_query} # 3. 定义协调员Agent并将专家作为其子Agent coordinator LlmAgent( nameCoordinator, modelgemini-2.5-flash, # 使用能力更强的模型做路由决策 instruction你是客服系统的总协调员。首先友好地问候用户。 然后分析用户的问题 - 如果涉及软件错误、无法安装、连接问题 - 调用工具转接给 TechSupport。 - 如果涉及订单状态、退货、物流 - 调用工具转接给 OrderSpecialist。 - 如果是简单问候或无法归类的问题由你自己直接回答。 在转接时请简要总结用户问题作为工具参数。, description客服入口负责问题分类和路由。, tools[transfer_to_agent], # 协调员拥有路由工具 sub_agents[tech_support_agent, order_agent] # 声明下属Agent ) # 4. 使用一个支持多Agent的引擎来运行例如 AgentEngine 配置了相应的路由逻辑 # 这里省略了引擎的具体配置代码它需要能理解TRANSFER_REQUEST信号并切换活跃的Agent。在这个设计中coordinator作为根Agent。当它决定调用transfer_to_agent工具时一个设计良好的多Agent引擎会捕获这个信号暂停coordinator的会话然后创建一个新的或切换到对应的子Agent会话并将用户问题传递过去。子Agent处理完毕后控制权可以返回给协调员或直接结束。多Agent设计要点明确分工 每个子Agent应有清晰、单一的职责边界。路由逻辑 协调员Agent的路由决策能力至关重要。可以通过精细的Instruction、示例对话few-shot甚至微调来提升其分类准确率。状态共享与隔离 子Agent之间是共享会话状态还是完全隔离这取决于业务逻辑。ADK的Session对象可以设计为全局共享也可以为每个子对话创建分支。编排复杂性 多Agent系统引入了编排Orchestration的复杂性。除了简单的树状结构还可能需要循环、条件分支、并行执行等。ADK的底层提供了构建这些复杂工作流的组件但需要开发者进行更多的设计和编码。4. 进阶实战开发、调试与部署全流程构建出Agent只是第一步让它成为一个健壮的应用还需要开发、调试、评估和部署等一系列工程化实践。4.1 利用开发UI进行交互式调试ADK提供了一个内置的开发UI这对于调试Agent行为来说是一个神器。它让你可以像使用ChatGPT界面一样与你的Agent对话但同时能实时看到底层的工具调用、模型请求/响应、会话状态等所有细节。启动开发UI非常简单。假设你的主Agent定义在一个名为my_agent的变量中并且你已经初始化了client。# dev_ui.py from google import genai from google.adk.agents import Agent from google.adk.engines import AgentEngine from google.adk.contrib.development.ui import run_ui client genai.Client() my_agent Agent(nameDebugAgent, modelgemini-2.0-flash-exp, instruction...) engine AgentEngine(agentmy_agent, clientclient) # 一行代码启动本地Web UI run_ui(engineengine, host0.0.0.0, port8080)运行这个脚本然后在浏览器中打开http://localhost:8080。你会看到一个界面左侧是聊天窗右侧是详细的“开发者面板”里面包含会话树 可视化展示整个对话和工具调用的流程。消息详情 点击任何一条消息可以查看其原始的Prompt结构、模型的完整响应。工具调用 查看每次工具调用的输入参数和返回结果。会话状态 实时查看和编辑session.state中的自定义数据。在开发过程中我习惯一边在UI里测试各种边界案例一边观察右侧的日志。当Agent行为不符合预期时可以立刻检查是Instruction描述不清还是工具返回的数据格式有问题亦或是模型的理解出现了偏差。这种即时反馈极大地提升了调试效率。4.2 编写评估集量化Agent性能Agent的“感觉”对了还不够我们需要可量化的评估。ADK内置了评估框架允许你定义“评估集”Eval Set来系统化地测试Agent。一个评估集是一个JSON文件包含了一系列的“测试用例”。// translation_eval_set.json [ { input: {messages: [{role: user, content: 将‘Hello, world!’翻译成法语。}]}, expected_output: {messages: [{role: model, content: Bonjour le monde !}]} }, { input: {messages: [{role: user, content: 将‘这座建筑非常古老。’翻译成英语。}]}, expected_output: {messages: [{role: model, content: This building is very old.}]} }, { input: {messages: [{role: user, content: 今天的天气真好对吧}]}, expected_output: {messages: [{role: model, content: 今天天气不错但这不是一个翻译请求我主要负责翻译工作。}]}, eval_instruction: 检查模型是否识别出这不是翻译请求并做出恰当回应。 } ]然后使用ADK的命令行工具adk eval来运行评估adk eval \ path/to/your/agent_project \ # 包含你agent定义和依赖的目录 path/to/translation_eval_set.json评估器会运行每个测试用例将Agent的实际输出与预期输出进行比较并生成一份报告。报告会包含通过率、失败案例的详细对比等信息。你还可以自定义评估逻辑例如使用另一个LLM来判断回答的相关性而不仅仅是字符串匹配这为评估复杂的、开放性的任务提供了可能。评估实践建议从核心场景开始 先为最核心、最常用的用户对话路径编写评估集。包含负面用例 不仅要测Agent“应该做什么”还要测它“不应该做什么”比如拒绝不合理请求、处理边界输入等。集成到CI/CD 将adk eval作为自动化测试流水线的一环确保每次代码更新都不会导致关键功能的回归。4.3 部署为生产级API服务ADK与FastAPI集成得非常好可以轻松地将你的Agent系统包装成REST API服务。# api_server.py from fastapi import FastAPI, HTTPException from pydantic import BaseModel from google import genai from google.adk.agents import Agent from google.adk.engines import AgentEngine from google.adk.sessions import Session import uvicorn import logging # 初始化 app FastAPI(titleMy Agent API) client genai.Client() agent Agent(nameAPIAgent, modelgemini-2.0-flash-exp, instruction..., tools[...]) engine AgentEngine(agentagent, clientclient) # 定义请求/响应模型 class ChatRequest(BaseModel): message: str session_id: str | None None # 支持多轮对话的会话ID class ChatResponse(BaseModel): reply: str session_id: str tool_calls: list | None None # 内存中的会话存储生产环境应使用数据库或Redis sessions_store {} app.post(/chat, response_modelChatResponse) async def chat_endpoint(request: ChatRequest): 主要的聊天端点 try: # 获取或创建会话 session sessions_store.get(request.session_id) if session is None: session Session() sessions_store[session.id] session # 添加用户消息并运行引擎 session.add_human_message(request.message) response await engine.run(sessionsession) # 获取最终回复 last_message response.messages[-1] # 更新存储 sessions_store[session.id] session return ChatResponse( replylast_message.text, session_idsession.id, tool_callsresponse.tool_calls # 可选返回本次交互中的工具调用详情 ) except Exception as e: logging.error(fError processing chat request: {e}) raise HTTPException(status_code500, detailInternal server error) app.get(/session/{session_id}) async def get_session_state(session_id: str): 获取会话状态用于调试 session sessions_store.get(session_id) if not session: raise HTTPException(status_code404, detailSession not found) # 注意生产环境可能需要过滤敏感信息 return {session_id: session.id, state: session.state.__dict__} app.delete(/session/{session_id}) async def delete_session(session_id: str): 清理会话 if session_id in sessions_store: del sessions_store[session_id] return {message: Session deleted} if __name__ __main__: uvicorn.run(app, host0.0.0.0, port8000)现在你的Agent就变成了一个标准的Web服务。前端应用可以通过POST /chat接口与之交互并通过session_id来维持多轮对话的上下文。部署注意事项会话存储 上述示例使用内存字典这只适用于单进程开发。生产环境必须使用外部存储如Redis或数据库并考虑会话的过期清理策略。认证与鉴权 使用FastAPI的Depends和安全中间件如OAuth2为你的API添加认证。限流与监控 使用像slowapi这样的中间件添加速率限制。集成Prometheus或OpenTelemetry来监控API的延迟、错误率和Token消耗。容器化 使用Docker将你的应用及其依赖打包。Dockerfile可以非常简单FROM python:3.11-slim WORKDIR /app COPY requirements.txt . RUN pip install --no-cache-dir -r requirements.txt COPY . . CMD [uvicorn, api_server:app, --host, 0.0.0.0, --port, 8000]部署到云平台 将容器镜像推送到Google Container Registry (GCR)就可以一键部署到Cloud Run享受自动扩缩容和免运维的优势。这也是ADK与Google Cloud生态无缝集成的优势所在。5. 避坑指南与性能优化在实际项目中踩过一些坑后我总结出以下经验希望能帮你少走弯路。5.1 工具设计与调用的常见陷阱工具描述模糊 这是导致模型“乱调用”或“不调用”工具的最常见原因。工具的description和参数的Field(description...)必须精确、无歧义。用动词开头如“计算...”、“查询...”、“发送...”并明确说明工具的用途、输入要求和输出格式。工具过多与冲突 给一个Agent注入太多功能相似的工具会让模型困惑。如果工具超过10个考虑是否应该拆分成多个专门的Agent。确保工具名称和描述有足够的区分度。复杂的输出格式 工具函数应返回简单的字符串或结构清晰的字典。避免返回过长的HTML、复杂的嵌套JSON或二进制数据。模型处理简洁文本的效果最好。如果需要复杂数据可以考虑让工具返回一个摘要或引用ID然后由前端或另一个服务去获取完整数据。异步工具处理 如果工具需要执行网络I/O如调用外部API务必将其定义为async函数并在内部使用await。这能避免阻塞整个引擎提升并发性能。5.2 会话管理与状态维护的实践状态爆炸 不要无限制地在session.state中存储数据。长时间对话后状态可能变得巨大不仅影响性能在作为上下文传给模型时也可能超出令牌限制。定期清理旧状态或只存储摘要和关键引用。状态序列化 如果你将会话存储到数据库session.state必须是可JSON序列化的。避免存储复杂的Python对象如数据库连接。只存储基本数据类型str, int, dict, list或Pydantic模型。利用“Rewind”功能 ADK新引入的rewind能力非常强大。它允许你将会话回滚到某个工具调用之前。这在调试时非常有用当工具调用出错后你可以回滚状态修改工具代码或参数然后重新执行而无需从头开始整个对话。5.3 性能与成本优化策略模型选择 不是所有任务都需要gemini-2.5-pro这样的顶级模型。对于简单的分类、路由、信息提取任务gemini-2.0-flash甚至gemini-2.0-flash-thinking-exp在成本和速度上更有优势。根据任务复杂度进行模型分级。上下文长度管理 这是控制成本的关键。ADK的Session会自动管理消息历史但你需要设置一个合理的截断策略。例如只保留最近10轮对话或者将更早的对话总结成一段摘要后再存入上下文。缓存 对于频繁出现的、结果确定的用户查询例如“公司的退货政策是什么”可以考虑引入缓存层。在调用引擎之前先检查缓存中是否有相同问题的答案。这能显著降低LLM调用次数和延迟。批量处理 如果有大量独立的文本需要处理如情感分析、关键词提取不要逐个调用Agent。可以设计一个批处理工具或者将多个请求打包成一个提示利用模型的并行处理能力但要注意提示长度限制。5.4 测试策略让Agent行为可预测单元测试工具 像测试普通函数一样测试你的工具。确保它们在各种边界输入下都能正确运行并返回预期格式。集成测试会话流 编写测试来模拟完整的用户对话流。使用adk的测试工具或在内存中运行Engine断言在给定一系列用户输入后会话的最终状态和输出符合预期。Mock外部依赖 在测试中务必Mock所有外部API调用如搜索、数据库、支付网关。使用unittest.mock来模拟这些服务返回预设的响应保证测试的稳定性和速度。评估集即测试 将你的评估集Eval Set视为功能测试套件。在每次重大更新后运行它们确保核心功能没有退化。ADK代表的是一种更工程化、更可控的Agent开发范式。它将AI Agent从“神奇的咒语”变成了可组装的软件组件。虽然初期学习曲线可能比一些“一行代码调用Agent”的框架要陡峭但它带来的可维护性、可测试性和架构上的清晰度对于构建严肃的、生产级的AI应用来说是至关重要的。从定义一个简单的工具开始逐步构建复杂的多Agent工作流在这个过程中你会发现自己对智能体系统的掌控力越来越强。