Google Agent Development Kit (ADK) 指南 第三章：核心概念与架构

Google Agent Development Kit (ADK) 指南 第三章：核心概念与架构

系列教程：这是《Google ADK 指南》系列的第三章。

前置知识：已完成第二章，完成环境搭建。

目录

1. ADK 架构概览
2. 核心组件详解
3. 数据流与执行模型
4. 扩展点与插件
5. 最佳实践

1. ADK 架构概览

1.1 整体架构

┌─────────────────────────────────────────────────────────────────┐ │ 应用层 │ │ (Chatbot、API Service、Batch Processing、Web UI) │ ├─────────────────────────────────────────────────────────────────┤ │ ADK 框架层 │ │ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ ┌───────────┐ │ │ │ Agent │ │ Tool │ │ Memory │ │ Evaluator│ │ │ │ Engine │ │ Manager │ │ Manager │ │ │ │ │ └─────────────┘ └─────────────┘ └─────────────┘ └───────────┘ │ │ ┌─────────────────────────────────────────────────────────────┐│ │ │ Orchestrator ││ │ │ (Single Agent | Multi-Agent | Workflow | Parallel) ││ │ └─────────────────────────────────────────────────────────────┘│ ├─────────────────────────────────────────────────────────────────┤ │ 模型层 │ │ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ ┌───────────┐ │ │ │ Gemini │ │ PaLM │ │ Codey │ │ Third- │ │ │ │ 2.0 │ │ 2 │ │ │ │ party │ │ │ └─────────────┘ └─────────────┘ └─────────────┘ └───────────┘ │ ├─────────────────────────────────────────────────────────────────┤ │ GCP 基础设施层 │ │ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ ┌───────────┐ │ │ │ Vertex │ │ GCS │ │ BigQuery │ │ IAM │ │ │ │ AI │ │ │ │ │ │ │ │ │ └─────────────┘ └─────────────┘ └─────────────┘ └───────────┘ │ │ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ ┌───────────┐ │ │ │ Cloud │ │ Pub/Sub │ │ Cloud │ │ Secret │ │ │ │ Logging │ │ │ │ Functions │ │ Manager │ │ │ └─────────────┘ └─────────────┘ └─────────────┘ └───────────┘ │ └─────────────────────────────────────────────────────────────────┘

1.2 设计原则

2. 核心组件详解

2.1 Agent（智能体）

定义：Agent 是 ADK 的核心，负责接收输入、决策、调用工具、生成输出。

核心属性：

fromgoogle.adkimportAgentfromgoogle.adk.modelsimportGeminifromgoogle.adk.memoryimportConversationBufferMemory agent=Agent(# 基础配置name="customer_service_agent",description="客户服务智能代理",# 模型配置model=Gemini("gemini-2.0-pro"),# 指令（System Prompt）instruction="""你是一个专业的客户服务代表。 - 用友好、专业的语气回答 - 只根据知识库内容回答 - 不知道的事情要承认""",# 工具列表tools=[search_tool,order_tool,refund_tool],# 记忆配置memory=ConversationBufferMemory(max_turns=20),# 生成配置generation_config={"temperature":0.7,"max_output_tokens":2048,"top_p":0.95},# 安全配置safety_settings={"HARM_CATEGORY_HARASSMENT":"BLOCK_MEDIUM_AND_ABOVE"})

Agent 生命周期：

初始化 → 接收输入 → 思考规划 → 调用工具 → 生成响应 → 更新记忆 → 输出 ↑ │ └────────────────────────── 循环 ──────────────────────────────────┘

2.2 Tool（工具）

定义：Tool 是 Agent 可调用的外部函数或 API。

工具类型：

创建工具：

fromgoogle.adkimportToolfrompydanticimportBaseModel,Field# 方式 1：装饰器（推荐）@Tool(name="calculate_shipping",description="计算运费",parameters={"weight":{"type":"number","description":"包裹重量 (kg)"},"destination":{"type":"string","description":"目的地城市"},"express":{"type":"boolean","description":"是否快递"}})defcalculate_shipping(weight:float,destination:str,express:bool=False)->str:"""计算运费并返回结果"""base_rate=10.0express_rate=2.0ifexpresselse0.0total=base_rate*weight+express_ratereturnf"运费：${total:.2f}"# 方式 2：类定义（复杂工具）fromgoogle.adk.toolsimportBaseToolclassOrderLookupTool(BaseTool):name="lookup_order"description="查询订单状态"classParameters(BaseModel):order_id:str=Field(...,description="订单 ID")email:str=Field(...,description="用户邮箱")defexecute(self,order_id:str,email:str)->dict:# 查询数据库order=db.query("SELECT * FROM orders WHERE id=? AND email=?",order_id,email)return{"status":order.status,"items":order.items,"tracking":order.tracking_number}

2.3 Memory（记忆）

定义：Memory 管理 Agent 的状态和历史信息。

记忆类型：

# 1. 对话缓冲记忆（短期）fromgoogle.adk.memoryimportConversationBufferMemory memory=ConversationBufferMemory(max_turns=10)# 2. 向量记忆（长期）fromgoogle.adk.memoryimportVectorStoreMemoryfromgoogle.adk.vectorstoresimportVertexAISearch memory=VectorStoreMemory(store=VertexAISearch(project_id="your-project",location_id="global"))# 3. 实体记忆（结构化）fromgoogle.adk.memoryimportEntityMemory memory=EntityMemory(entities=["user_profile","preferences","history"])# 4. 组合记忆fromgoogle.adk.memoryimportCompositeMemory memory=CompositeMemory(short_term=ConversationBufferMemory(max_turns=10),long_term=VectorStoreMemory(...))

记忆操作：

# 添加记忆memory.add_message(role="user",content="你好")memory.add_message(role="assistant",content="你好！有什么可以帮助你的？")# 获取历史history=memory.get_history(last_n=5)# 搜索记忆results=memory.search("用户偏好",top_k=3)# 清除记忆memory.clear()

2.4 Evaluator（评估器）

定义：Evaluator 评估 Agent 输出质量。

内置评估器：

fromgoogle.adk.evaluatorsimport(RelevanceEvaluator,FaithfulnessEvaluator,SafetyEvaluator,CustomEvaluator)# 相关性评估relevance_eval=RelevanceEvaluator(model=Gemini("gemini-2.0-flash"),threshold=0.7)# 忠实度评估（是否基于事实）faithfulness_eval=FaithfulnessEvaluator(context_required=True)# 安全评估safety_eval=SafetyEvaluator(blocked_categories=["HARASSMENT","HATE_SPEECH"])# 自定义评估classToneEvaluator(CustomEvaluator):defevaluate(self,response:str,context:dict)->float:# 评估语气是否友好ifany(wordinresponseforwordin["抱歉","感谢","请"]):return1.0return0.5# 组合评估fromgoogle.adk.evaluatorsimportEnsembleEvaluator evaluator=EnsembleEvaluator(evaluators=[relevance_eval,faithfulness_eval,safety_eval],weights=[0.4,0.4,0.2])

2.5 Orchestrator（编排器）

定义：Orchestrator 协调多个 Agent 或复杂工作流。

编排模式：

# 1. 单 Agent（默认）fromgoogle.adkimportSingleAgentOrchestrator orchestrator=SingleAgentOrchestrator(agent=single_agent)# 2. 多 Agent 路由fromgoogle.adkimportMultiAgentOrchestrator orchestrator=MultiAgentOrchestrator(agents={"sales":sales_agent,"support":support_agent,"billing":billing_agent},router=SemanticRouter(# 语义路由model=Gemini("gemini-2.0-flash")))# 3. 顺序工作流fromgoogle.adkimportSequentialWorkflow workflow=SequentialWorkflow(steps=[intent_agent,# 识别意图retrieval_agent,# 检索信息response_agent# 生成回复])# 4. 并行处理fromgoogle.adkimportParallelWorkflow workflow=ParallelWorkflow(agents=[agent1,agent2,agent3],merge_strategy="concatenate"# 合并策略)# 5. 条件分支fromgoogle.adkimportConditionalWorkflow workflow=ConditionalWorkflow(conditions=[{"condition":lambdactx:ctx["intent"]=="complaint","agent":complaint_agent},{"condition":lambdactx:ctx["intent"]=="inquiry","agent":inquiry_agent}],default=default_agent)

3. 数据流与执行模型

3.1 请求处理流程

用户输入 ↓ ┌─────────────────┐ │ 输入预处理 │ (清洗、验证、标准化) └────────┬────────┘ ↓ ┌─────────────────┐ │ 意图识别 │ (分类、实体抽取) └────────┬────────┘ ↓ ┌─────────────────┐ │ 检索记忆 │ (历史对话、用户信息) └────────┬────────┘ ↓ ┌─────────────────┐ │ 规划与决策 │ (选择工具、制定步骤) └────────┬────────┘ ↓ ┌─────────────────┐ │ 工具执行 │ (并行/串行调用) └────────┬────────┘ ↓ ┌─────────────────┐ │ 响应生成 │ (LLM 生成) └────────┬────────┘ ↓ ┌─────────────────┐ │ 质量评估 │ (安全性、相关性) └────────┬────────┘ ↓ ┌─────────────────┐ │ 输出后处理 │ (格式化、过滤) └────────┬────────┘ ↓ 用户输出

3.2 执行模式

同步执行：

response=agent.run("你好")print(response.text)

异步执行：

importasyncioasyncdefmain():response=awaitagent.run_async("你好")print(response.text)asyncio.run(main())

流式执行：

forchunkinagent.run_stream("写一篇长文章"):print(chunk.text,end="",flush=True)

批处理：

inputs=["问题 1","问题 2","问题 3"]responses=agent.batch_run(inputs,max_concurrent=5)

3.3 错误处理

fromgoogle.adk.exceptionsimport(AgentError,ToolExecutionError,ModelAPIError,TimeoutError)try:response=agent.run(user_input,timeout=30)exceptToolExecutionErrorase:# 工具执行失败logger.error(f"工具失败：{e.tool_name}, 错误：{e.message}")response=agent.run("抱歉，暂时无法完成该操作。")exceptModelAPIErrorase:# 模型 API 错误logger.error(f"模型错误：{e.status_code}")response=agent.run("服务暂时不可用，请稍后重试。")exceptTimeoutError:# 超时logger.warning("请求超时")response=agent.run("处理超时，请简化问题。")exceptAgentErrorase:# 通用错误logger.error(f"Agent 错误：{e}")raise

4. 扩展点与插件

4.1 自定义模型 provider

fromgoogle.adk.modelsimportBaseModel,register_modelclassCustomModel(BaseModel):def__init__(self,model_name:str,api_key:str):self.model_name=model_name self.api_key=api_keydefgenerate(self,prompt:str,**kwargs)->str:# 调用自定义模型 APIresponse=requests.post("https://api.custom-llm.com/generate",headers={"Authorization":f"Bearer{self.api_key}"},json={"prompt":prompt,"model":self.model_name})returnresponse.json()["text"]# 注册register_model("custom",CustomModel)# 使用agent=Agent(model=CustomModel("custom-v1","your-api-key"))

4.2 自定义工具

fromgoogle.adk.toolsimportBaseToolfrompydanticimportBaseModelclassDatabaseQueryTool(BaseTool):name="query_database"description="执行 SQL 查询"classParameters(BaseModel):query:strread_only:bool=Truedefexecute(self,query:str,read_only:bool=True)->list:ifread_onlyandnotquery.strip().upper().startswith("SELECT"):raiseValueError("只读模式只允许 SELECT 查询")conn=get_db_connection()cursor=conn.cursor()cursor.execute(query)returncursor.fetchall()

4.3 自定义记忆存储

fromgoogle.adk.memoryimportBaseMemoryclassRedisMemory(BaseMemory):def__init__(self,redis_url:str,session_id:str):importredis self.redis=redis.from_url(redis_url)self.session_key=f"adk:memory:{session_id}"defadd_message(self,role:str,content:str):self.redis.lpush(self.session_key,f"{role}:{content}")self.redis.ltrim(self.session_key,0,99)# 保留 100 条defget_history(self,last_n:int=10):messages=self.redis.lrange(self.session_key,0,last_n-1)return[m.decode().split(":",1)forminreversed(messages)]defclear(self):self.redis.delete(self.session_key)

5. 最佳实践

5.1 Agent 设计

推荐：

• 单一职责原则（一个 Agent 做好一件事）
• 清晰的指令（System Prompt）
• 合理的工具数量（3-7 个）
• 适当的记忆长度（10-20 轮）

避免：

• 万能 Agent（什么都做）
• 模糊的指令
• 过多工具（>15 个）
• 无限记忆

5.2 工具开发

推荐：

• 详细的描述和参数说明
• 完善的错误处理
• 超时控制
• 结果格式化

@Tool(name="search_products",description="搜索产品，支持关键词、价格范围、分类筛选",parameters={"keyword":{"type":"string","description":"搜索关键词"},"min_price":{"type":"number","description":"最低价格"},"max_price":{"type":"number","description":"最高价格"},"category":{"type":"string","description":"产品分类"}},timeout=10,# 超时设置cache_ttl=300# 缓存 5 分钟)defsearch_products(keyword:str,min_price:float=None,max_price:float=None,category:str=None)->list:try:# 实现...passexceptExceptionase:logger.error(f"搜索失败：{e}")return[]

5.3 性能优化

# 1. 启用缓存fromgoogle.adk.cachingimportResponseCache agent.cache=ResponseCache(ttl=3600)# 2. 批量工具调用@Tool(batch_enabled=True)defbatch_lookup_ids(ids:list[str])->list:# 一次调用查询多个 IDpass# 3. 异步并发importasyncio results=awaitasyncio.gather(tool1.run_async(arg1),tool2.run_async(arg2),tool3.run_async(arg3))# 4. 流式响应forchunkinagent.run_stream(long_query):process(chunk)# 边生成边处理

5.4 安全实践

# 1. 输入验证fromgoogle.adk.securityimportInputValidator validator=InputValidator(max_length=1000,blocked_patterns=[r"<script>",r"DROP TABLE"],require_auth=True)# 2. 输出过滤fromgoogle.adk.securityimportOutputFilterfilter=OutputFilter(remove_pii=True,# 移除个人信息block_sensitive=True# 阻止敏感信息)# 3. 速率限制fromgoogle.adk.securityimportRateLimiter limiter=RateLimiter(requests_per_minute=60,tokens_per_minute=10000)# 4. 审计日志fromgoogle.adk.loggingimportAuditLogger audit=AuditLogger(log_input=True,log_output=True,log_tool_calls=True,retention_days=90)

系列教程导航：

• 第一章：ADK 简介与对比
• 第二章：环境搭建与快速开始
• 第三章：核心概念与架构 ← 本章
• 第四章：Agent 开发与编排
• 第五章：工具集成与自定义
• 第六章：记忆与状态管理
• 第七章：企业级功能与安全
• 第八章：实战案例与最佳实践