# 第04章：自定义工具开发 ## 📌 本章目标 - 理解 ADK 工具系统的工作原理 - 掌握函数工具（Function Tool）的开发方法 - 学习 MCP 工具的集成方式 - 了解 OpenAPI 工具的生成方法 - 掌握工具间数据传递的技巧 --- ## 4.1 工具系统概述 ### 4.1.1 工具是什么？ 在 ADK 中，**工具（Tool）** 是 Agent 与外部世界交互的桥梁。Agent 通过调用工具来执行特定任务，如查询数据库、调用 API、执行计算等。 ``` ┌─────────────────────────────────────────────┐ │ Agent │ │ │ │ 用户: "帮我查一下北京的天气" │ │ ↓ │ │ LLM 分析: 需要调用天气查询工具 │ │ ↓ │ │ 调用工具: get_weather(city="北京") │ │ ↓ │ │ 工具返回: {"status": "success", │ │ "report": "晴天，25°C"} │ │ ↓ │ │ LLM 生成回复: "北京今天天气晴朗，气温25°C" │ └─────────────────────────────────────────────┘ ``` ### 4.1.2 工具类型 | 类型 | 说明 | 适用场景 | |------|------|----------| | **Function Tool** | Python 函数自动转换 | 自定义逻辑、API 调用 | | **MCP Tool** | 连接 MCP 服务器 | 标准化工具集成 | | **OpenAPI Tool** | 从 OpenAPI 规范生成 | REST API 集成 | | **Agent Tool** | 将 Agent 作为工具使用 | Agent 复用 | | **内置工具** | Google 提供的预置工具 | 搜索、代码执行等 | --- ## 4.2 函数工具（Function Tool） ### 4.2.1 基本原理 ADK 会自动检查 Python 函数的以下信息来生成工具 Schema： - **函数名**：工具的名称 - **docstring**：工具的描述（发送给 LLM） - **参数类型**：参数的类型提示 - **默认值**：区分必选和可选参数 - **返回值**：工具的输出 ### 4.2.2 基础示例 ```python """ 函数工具基础示例 演示如何创建简单的自定义工具 """ from google.adk.agents import Agent # 导入 Agent 类 # ======================================== # 示例一：简单的天气查询工具 # ======================================== def get_weather(city: str) -> dict: """ 获取指定城市的天气信息 Args: city (str): 城市名称，例如"北京"、"上海" Returns: dict: 包含天气状态的字典 """ # 模拟天气数据（实际应用中应调用真实天气 API） weather_data = { # 模拟天气数据库 "北京": {"temp": "25°C", "condition": "晴天"}, # 北京天气 "上海": {"temp": "28°C", "condition": "多云"}, # 上海天气 "广州": {"temp": "32°C", "condition": "雷阵雨"}, # 广州天气 } # 查找城市天气数据 data = weather_data.get(city) # 从字典中获取天气数据 # 如果找不到该城市的天气 if not data: # 检查数据是否存在 return { # 返回错误信息 "status": "error", # 状态：错误 "error_message": f"未找到'{city}'的天气信息" # 错误描述 } # 返回成功结果 return { # 返回天气信息 "status": "success", # 状态：成功 "city": city, # 城市名称 "temperature": data["temp"], # 温度 "condition": data["condition"], # 天气状况 } # ======================================== # 示例二：带可选参数的工具 # ======================================== def search_restaurant( city: str, # 必选参数：城市 cuisine: str = "中餐", # 可选参数：菜系，默认中餐 price_range: str = "中等", # 可选参数：价格范围，默认中等 ) -> dict: """ 搜索指定城市的餐厅 Args: city (str): 城市名称 cuisine (str, optional): 菜系类型，默认为"中餐" price_range (str, optional): 价格范围，默认为"中等" Returns: dict: 包含搜索结果的字典 """ # 模拟餐厅搜索结果 return { # 返回搜索结果 "status": "success", # 状态：成功 "city": city, # 城市 "cuisine": cuisine, # 菜系 "price_range": price_range, # 价格范围 "results": [ # 餐厅列表 {"name": "美味餐厅", "rating": 4.5}, # 餐厅1 {"name": "佳肴轩", "rating": 4.2}, # 餐厅2 ], } # ======================================== # 创建使用工具的 Agent # ======================================== agent = Agent( name="tool_agent", # Agent 名称 model="gemini-2.0-flash", # 使用 Gemini 模型 instruction=( # 系统指令 "你是一个生活助手。\n" "你可以查询天气和搜索餐厅。\n" "根据用户需求使用合适的工具。" ), tools=[ # 注册工具列表 get_weather, # 天气查询工具 search_restaurant, # 餐厅搜索工具 ], ) ``` ### 4.2.3 参数类型详解 ```python """ 工具参数类型详解 展示不同类型的参数定义方式 """ from typing import Optional, List # 导入类型提示 from google.adk.agents import Agent # 导入 Agent 类 # ======================================== # 必选参数 vs 可选参数 # ======================================== def calculate_loan( principal: float, # 必选：贷款本金 annual_rate: float, # 必选：年利率 years: int, # 必选：贷款年限 extra_payment: float = 0.0, # 可选：额外月供，默认0 ) -> dict: """ 计算贷款月供和总利息 Args: principal (float): 贷款本金（元） annual_rate (float): 年利率（百分比，如 4.5 表示 4.5%） years (int): 贷款年限 extra_payment (float, optional): 每月额外还款金额，默认为 0 Returns: dict: 包含月供、总利息等信息的字典 """ monthly_rate = annual_rate / 100 / 12 # 计算月利率 total_months = years * 12 # 计算总月数 # 计算等额本息月供 monthly_payment = principal * monthly_rate * (1 + monthly_rate) ** total_months / ((1 + monthly_rate) ** total_months - 1) total_amount = monthly_payment * total_months + extra_payment * total_months # 总还款额 total_interest = total_amount - principal # 总利息 return { # 返回计算结果 "status": "success", # 状态 "monthly_payment": round(monthly_payment, 2), # 月供 "total_interest": round(total_interest, 2), # 总利息 "total_amount": round(total_amount, 2), # 总还款额 } # ======================================== # 使用 Optional 标记可选参数 # ======================================== def create_user_profile( username: str, # 必选：用户名 bio: Optional[str] = None, # 可选：个人简介 avatar_url: Optional[str] = None, # 可选：头像 URL ) -> dict: """ 创建用户档案 Args: username (str): 用户名 bio (str, optional): 个人简介 avatar_url (str, optional): 头像图片 URL Returns: dict: 创建结果 """ profile = { # 创建用户档案字典 "username": username, # 用户名 "bio": bio or "这个人很懒，什么都没写。", # 简介（默认值） "avatar_url": avatar_url or "", # 头像 URL } return { # 返回结果 "status": "success", # 状态 "profile": profile, # 用户档案 } # ======================================== # 使用 List 类型参数 # ======================================== def compare_cities(cities: List[str]) -> dict: """ 比较多个城市的信息 Args: cities (List[str]): 城市名称列表 Returns: dict: 城市比较结果 """ city_info = { # 城市信息数据库 "北京": {"population": "2189万", "area": "16410km²"}, # 北京 "上海": {"population": "2487万", "area": "6341km²"}, # 上海 "广州": {"population": "1881万", "area": "7434km²"}, # 广州 } results = [] # 初始化结果列表 for city in cities: # 遍历城市列表 info = city_info.get(city) # 获取城市信息 if info: # 如果信息存在 results.append({ # 添加到结果列表 "city": city, # 城市名 **info, # 展开城市信息 }) return { # 返回比较结果 "status": "success" if results else "error", # 状态 "results": results, # 比较结果 } # 创建 Agent agent = Agent( name="param_agent", # Agent 名称 model="gemini-2.0-flash", # 模型 instruction="你是一个多功能助手。", # 指令 tools=[ # 工具列表 calculate_loan, # 贷款计算器 create_user_profile, # 用户档案创建 compare_cities, # 城市比较 ], ) ``` ### 4.2.4 工具返回值最佳实践 ```python """ 工具返回值最佳实践 返回值会被 LLM 读取，因此需要清晰、结构化 """ # ❌ 不好的返回值：信息不清晰 def bad_tool(query: str) -> str: """不好的返回值示例""" return "0" # LLM 不知道 0 代表什么 # ✅ 好的返回值：结构化、有状态码 def good_tool(query: str) -> dict: """ 好的返回值示例 包含状态码和详细描述 """ return { "status": "success", # 状态码：success / error / pending "query": query, # 原始查询 "result": "查询结果详情", # 具体结果 "metadata": { # 可选的元数据 "source": "database", # 数据来源 "timestamp": "2025-04-05T10:00:00Z", # 时间戳 }, } # ✅ 错误处理：返回有意义的错误信息 def robust_tool(query: str) -> dict: """ 健壮的工具：包含完善的错误处理 """ if not query or not query.strip(): # 检查空输入 return { # 返回参数错误 "status": "error", # 状态：错误 "error_code": "INVALID_INPUT", # 错误码 "error_message": "查询内容不能为空，请提供有效的查询参数。" # 人类可读的错误信息 } try: # 尝试执行操作 result = do_something(query) # 执行实际操作 return { # 返回成功结果 "status": "success", # 状态：成功 "result": result, # 操作结果 } except Exception as e: # 捕获异常 return { # 返回异常信息 "status": "error", # 状态：错误 "error_code": "INTERNAL_ERROR", # 错误码 "error_message": f"处理请求时发生错误：{str(e)}" # 错误描述 } def do_something(query: str) -> str: """模拟操作函数""" return f"处理结果: {query}" # 返回处理结果 ``` --- ## 4.3 工具间数据传递 ```python """ 工具间数据传递 使用 session.state 的 temp: 前缀在工具间传递数据 """ from google.adk.agents import Agent # 导入 Agent 类 from google.adk.tools import FunctionTool # 导入 FunctionTool from google.adk.tools import ToolContext # 导入工具上下文 # ======================================== # 使用 ToolContext 访问会话状态 # ======================================== def step1_fetch_data( query: str, # 搜索关键词 ctx: ToolContext, # 工具上下文（自动注入） ) -> dict: """ 第一步：获取原始数据 Args: query (str): 搜索关键词 ctx (ToolContext): 工具上下文，可用于访问会话状态 Returns: dict: 原始数据 """ # 模拟获取数据 raw_data = f"关于'{query}'的原始数据..." # 模拟数据 # 将数据保存到临时状态（temp: 前缀） # temp: 状态只在当前调用中有效，调用结束后自动清除 ctx.state["temp:raw_data"] = raw_data # 保存到临时状态 return { # 返回结果 "status": "success", # 状态 "message": f"已获取关于'{query}'的数据", # 消息 "data_length": len(raw_data), # 数据长度 } def step2_analyze_data(ctx: ToolContext) -> dict: """ 第二步：分析第一步获取的数据 Args: ctx (ToolContext): 工具上下文 Returns: dict: 分析结果 """ # 从临时状态中读取第一步保存的数据 raw_data = ctx.state.get("temp:raw_data") # 读取临时状态 if not raw_data: # 如果没有数据 return { # 返回错误 "status": "error", "error_message": "没有找到原始数据，请先执行数据获取步骤。" } # 模拟数据分析 analysis = f"对数据进行了深度分析：{raw_data[:50]}..." # 分析结果 return { # 返回分析结果 "status": "success", # 状态 "analysis": analysis, # 分析结果 } # 创建 Agent agent = Agent( name="pipeline_agent", # Agent 名称 model="gemini-2.0-flash", # 模型 instruction=( # 指令 "你是一个数据分析助手。\n" "当用户需要分析数据时，先使用 step1_fetch_data 获取数据，\n" "然后使用 step2_analyze_data 进行分析。" ), tools=[ # 工具列表 step1_fetch_data, # 数据获取工具 step2_analyze_data, # 数据分析工具 ], ) ``` --- ## 4.4 Agent-as-a-Tool（智能体作为工具） ```python """ Agent-as-a-Tool 将一个 Agent 包装为工具，供其他 Agent 调用 """ from google.adk.agents import Agent # 导入 Agent 类 from google.adk.tools import AgentTool # 导入 AgentTool 类 # ======================================== # 定义一个专门的翻译 Agent # ======================================== translator_agent = Agent( name="translator", # 翻译 Agent 名称 model="gemini-2.0-flash", # 使用 Gemini 模型 instruction="你是一个专业翻译，将用户输入的文本翻译成目标语言。只输出翻译结果。", # 指令 ) # ======================================== # 将翻译 Agent 包装为工具 # ======================================== translation_tool = AgentTool( # 创建 AgentTool agent=translator_agent, # 传入要包装的 Agent ) # ======================================== # 在主 Agent 中使用翻译工具 # ======================================== main_agent = Agent( name="main_agent", # 主 Agent 名称 model="gemini-2.0-flash", # 使用 Gemini 模型 instruction=( # 系统指令 "你是一个多语言助手。\n" "当用户需要翻译时，使用 translation_tool 工具。\n" "其他情况直接回答用户问题。" ), tools=[translation_tool], # 注册翻译工具 ) # AgentTool 与 sub_agents 的区别： # - sub_agents：通过 LLM Transfer 机制委派任务，Agent 切换执行 # - AgentTool：作为工具被调用，在当前 Agent 上下文中执行 ``` --- ## 4.5 MCP 工具集成 ```python """ MCP（Model Context Protocol）工具集成 连接外部 MCP 服务器作为 Agent 的工具 """ from google.adk.agents import Agent # 导入 Agent 类 from google.adk.tools.mcp_tool.mcp_toolset import McpToolset # 导入 MCP 工具集 # ======================================== # 连接本地 MCP 服务器 # ======================================== async def create_mcp_agent(): """创建使用 MCP 工具的 Agent""" # 创建 MCP 工具集 # 需要先启动 MCP 服务器（如 filesystem、database 等） mcp_tools = McpToolset( connection_params={ # MCP 服务器连接参数 "url": "http://localhost:3000", # MCP 服务器地址 }, tool_names=[ # 要导入的工具名称列表 "read_file", # 读取文件工具 "write_file", # 写入文件工具 "list_directory", # 列出目录工具 ], ) # 获取 MCP 工具列表 tools = await mcp_tools.get_tools() # 异步获取工具 # 创建使用 MCP 工具的 Agent agent = Agent( name="mcp_agent", # Agent 名称 model="gemini-2.0-flash", # 使用 Gemini 模型 instruction="你是一个文件管理助手，可以读写文件和浏览目录。", # 指令 tools=tools, # 注册 MCP 工具 ) return agent # 返回 Agent ``` --- ## 4.6 完整代码示例 详见 `code/custom_tools_demo.py` — 包含完整的自定义工具开发示例。 --- ## 📌 本章小结 - ADK 自动将 Python 函数转换为工具，通过函数签名生成 Schema - 好的 docstring 是 LLM 理解工具的关键 - 返回值推荐使用字典，包含 `status` 字段 - `ToolContext` 允许工具访问会话状态，实现工具间数据传递 - `AgentTool` 可以将 Agent 包装为工具供其他 Agent 使用 - MCP 工具集支持连接外部标准化工具服务器 **下一章**：[第05章 - 多智能体系统](./05-multi-agent-systems.md) → 学习如何构建多 Agent 协作系统。