565 lines
19 KiB
Markdown
565 lines
19 KiB
Markdown
# 第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 协作系统。
|