# 工具系统文档
> Agent 框架的工具系统:定义、注册、执行工具调用。
---
## 目录
1. [核心概念](#核心概念)
2. [定义工具](#定义工具)
3. [ToolResult 和记忆管理](#toolresult-和记忆管理)
4. [ToolContext 和依赖注入](#toolcontext-和依赖注入)
5. [高级特性](#高级特性)
6. [内置基础工具](#内置基础工具)
7. [集成 Browser-Use](#集成-browser-use)
8. [最佳实践](#最佳实践)
---
## 核心概念
### 三个核心类型
```python
from reson_agent import tool, ToolResult, ToolContext
@tool()
async def my_tool(arg: str, context: Optional[ToolContext] = None) -> ToolResult:
return ToolResult(
title="Success",
output="Result content"
)
```
| 类型 | 作用 | 定义位置 |
|------|------|---------|
| **`@tool`** | 装饰器,自动注册工具并生成 Schema | `tools/registry.py` |
| **`ToolResult`** | 工具执行结果(支持记忆管理) | `tools/models.py` |
| **`ToolContext`** | 工具执行上下文(依赖注入) | `tools/models.py` |
### 工具的生命周期
```
1. 定义工具
↓ @tool() 装饰器
2. 自动注册到 ToolRegistry
↓ 生成 OpenAI Tool Schema(跳过 hidden_params)
3. LLM 选择工具并生成参数
↓ registry.execute(name, args)
4. 注入框架参数(hidden_params + inject_params)
↓ 调用工具函数
5. 返回 ToolResult
↓ 转换为 LLM 消息
6. 添加到对话历史
```
### 参数注入机制
工具参数分为三类:
1. **业务参数**:LLM 可见,由 LLM 填写(如 `query`, `limit`)
2. **隐藏参数**:LLM 不可见,框架自动注入(如 `context`, `uid`)
3. **注入参数**:LLM 可见但有默认值,框架自动注入默认值(如 `owner`, `tags`)
```python
@tool(
hidden_params=["context", "uid"], # 不生成 schema,LLM 看不到
inject_params={ # 自动注入默认值
"owner": lambda ctx: ctx.config.knowledge.get_owner(),
"tags": lambda ctx, args: {**ctx.config.default_tags, **args.get("tags", {})},
}
)
async def knowledge_save(
task: str, # 业务参数:LLM 填写
content: str, # 业务参数:LLM 填写
types: List[str], # 业务参数:LLM 填写
tags: Optional[Dict] = None, # 注入参数:LLM 可填,框架提供默认值
owner: Optional[str] = None, # 注入参数:LLM 可填,框架提供默认值
context: Optional[ToolContext] = None, # 隐藏参数:LLM 看不到
uid: str = "", # 隐藏参数:LLM 看不到
) -> ToolResult:
"""保存知识到知识库"""
...
```
**注入时机**:
- Schema 生成时:跳过 `hidden_params`,不暴露给 LLM
- 工具执行前:注入 `hidden_params` 和 `inject_params` 的默认值
**实现位置**:
- Schema 生成:`agent/tools/schema.py:SchemaGenerator.generate()`
- 参数注入:`agent/tools/registry.py:ToolRegistry.execute()`
---
## 定义工具
### 最简形式
```python
from reson_agent import tool
@tool()
async def hello(name: str) -> str:
"""向用户问好"""
return f"Hello, {name}!"
```
**要点**:
- 可以是同步或异步函数
- 返回值自动序列化为 JSON
- 所有参数默认对 LLM 可见
### 带框架参数
```python
@tool(hidden_params=["context", "uid"])
async def search_notes(
query: str,
limit: int = 10,
context: Optional[ToolContext] = None,
uid: str = ""
) -> str:
"""
搜索笔记
Args:
query: 搜索关键词
limit: 返回结果数量
"""
# context 和 uid 由框架注入,LLM 看不到这两个参数
...
```
### 带参数注入
```python
@tool(
hidden_params=["context"],
inject_params={
"owner": lambda ctx: ctx.config.knowledge.get_owner(),
"tags": lambda ctx, args: {**ctx.config.default_tags, **args.get("tags", {})},
}
)
async def knowledge_save(
task: str,
content: str,
types: List[str],
tags: Optional[Dict] = None, # LLM 可填,框架提供默认值
owner: Optional[str] = None, # LLM 可填,框架提供默认值
context: Optional[ToolContext] = None
) -> ToolResult:
"""
保存知识
Args:
task: 任务描述
content: 知识内容
types: 知识类型
tags: 业务标签(可选,有默认值)
owner: 所有者(可选,有默认值)
"""
# owner 和 tags 如果 LLM 未提供,框架会注入默认值
...
```
**注入规则**:
- `inject_params` 的 value 可以是:
- `lambda ctx: ...` - 从 context 计算
- `lambda ctx, args: ...` - 从 context 和已有参数计算
- 字符串 - 直接使用该值
- 注入时机:工具执行前,使用 `setdefault` 注入(不覆盖 LLM 提供的值)
### 带 UI 元数据
```python
@tool(
display={
"zh": {
"name": "搜索笔记",
"params": {
"query": "搜索关键词",
"limit": "结果数量"
}
},
"en": {
"name": "Search Notes",
"params": {
"query": "Search query",
"limit": "Result limit"
}
}
}
)
async def search_notes(query: str, limit: int = 10, uid: str = "") -> str:
"""搜索用户的笔记"""
...
```
---
## ToolResult 和记忆管理
### 基础用法
```python
from reson_agent import ToolResult
@tool()
async def read_file(path: str) -> ToolResult:
content = Path(path).read_text()
return ToolResult(
title=f"Read {path}",
output=content
)
```
### 双层记忆管理
**问题**:某些工具返回大量内容(如 Browser-Use 的 `extract`),如果每次都放入对话历史,会快速耗尽 context。
**解决**:`ToolResult` 支持双层记忆:
```python
@tool()
async def extract_page_data(url: str) -> ToolResult:
# 假设提取了 10K tokens 的内容
full_content = extract_all_data(url)
return ToolResult(
title="Extracted page data",
output=full_content, # 完整内容(可能很长)
long_term_memory=f"Extracted {len(full_content)} chars from {url}", # 简短摘要
include_output_only_once=True # output 只给 LLM 看一次
)
```
**效果**:
- **第一次**:LLM 看到 `output`(完整内容)+ `long_term_memory`(摘要)
- **后续**:LLM 只看到 `long_term_memory`(摘要)
**对话历史示例**:
```
[User] 提取 amazon.com 的商品价格
[Assistant] 调用 extract_page_data(url="amazon.com")
[Tool]
# Extracted page data
<完整的 10K tokens 数据...>
Summary: Extracted 10000 chars from amazon.com
[User] 现在保存到文件
[Assistant] 调用 write_file(content="...")
[Tool] (此时不再包含 10K tokens,只有摘要)
Summary: Extracted 10000 chars from amazon.com
```
### 错误处理
```python
@tool()
async def risky_operation() -> ToolResult:
try:
result = perform_operation()
return ToolResult(
title="Success",
output=result
)
except Exception as e:
return ToolResult(
title="Failed",
output="",
error=str(e)
)
```
### 附件和图片
```python
@tool()
async def generate_report() -> ToolResult:
report_path = create_pdf_report()
screenshot_data = take_screenshot()
return ToolResult(
title="Report generated",
output="Report created successfully",
attachments=[report_path], # 文件路径列表
images=[{
"name": "screenshot.png",
"data": screenshot_data # Base64 或路径
}]
)
```
---
## ToolContext 和依赖注入
### 基本概念
工具函数可以声明需要 `ToolContext` 参数,框架自动注入。需要在 `@tool()` 装饰器中声明 `hidden_params=["context"]`,使其对 LLM 不可见。
```python
from reson_agent import ToolContext
@tool(hidden_params=["context"])
async def get_current_state(context: Optional[ToolContext] = None) -> ToolResult:
return ToolResult(
title="Current state",
output=f"Trace ID: {context.trace_id}\nStep ID: {context.step_id}"
)
```
### ToolContext 字段
```python
class ToolContext(Protocol):
# 基础字段(所有工具)
trace_id: str # 当前 Trace ID
step_id: str # 当前 Step ID
uid: Optional[str] # 用户 ID
# 扩展字段(由 runner 注入)
store: Optional[TraceStore] # Trace 存储
runner: Optional[AgentRunner] # Runner 实例
goal_tree: Optional[GoalTree] # 目标树
goal_id: Optional[str] # 当前 Goal ID
config: Optional[RunConfig] # 运行配置
# 浏览器相关(Browser-Use 集成)
browser_session: Optional[Any] # 浏览器会话
page_url: Optional[str] # 当前页面 URL
file_system: Optional[Any] # 文件系统访问
sensitive_data: Optional[Dict] # 敏感数据
# 额外上下文
context: Optional[Dict[str, Any]] # 额外上下文数据
```
### 使用示例
```python
@tool(hidden_params=["context"])
async def analyze_current_page(context: Optional[ToolContext] = None) -> ToolResult:
"""分析当前浏览器页面"""
if not context or not context.browser_session:
return ToolResult(
title="Error",
error="Browser session not available"
)
# 使用浏览器会话
page_content = await context.browser_session.get_content()
return ToolResult(
title=f"Analyzed {context.page_url}",
output=page_content,
long_term_memory=f"Analyzed page at {context.page_url}"
)
```
### 创建 ToolContext
Runner 在执行工具时自动创建并注入 context:
```python
# 在 AgentRunner._agent_loop 中
context = {
"store": self.trace_store,
"trace_id": trace_id,
"goal_id": current_goal_id,
"runner": self,
"goal_tree": goal_tree,
"config": config,
}
result = await self.tools.execute(
tool_name,
tool_args,
uid=config.uid or "",
context=context
)
```
---
## 高级特性
### 1. 需要用户确认
```python
@tool(requires_confirmation=True)
async def delete_all_notes(uid: str = "") -> ToolResult:
"""删除所有笔记(危险操作)"""
# 执行前会等待用户确认
...
```
**适用场景**:
- 删除操作
- 发送消息
- 修改重要设置
- 任何不可逆操作
### 2. 可编辑参数
```python
@tool(editable_params=["query", "filters"])
async def advanced_search(
query: str,
filters: Optional[Dict] = None,
uid: str = ""
) -> ToolResult:
"""高级搜索"""
# LLM 生成参数后,用户可以编辑 query 和 filters
...
```
**适用场景**:
- 搜索查询
- 内容创建
- 需要用户微调的参数
### 3. 域名过滤(URL Patterns)
**场景**:某些工具只在特定网站可用,减少无关工具的 context 占用。
```python
@tool(url_patterns=["*.google.com", "www.google.*"])
async def google_advanced_search(
query: str,
date_range: Optional[str] = None,
uid: str = ""
) -> ToolResult:
"""Google 高级搜索技巧(仅在 Google 页面可用)"""
...
@tool(url_patterns=["*.github.com"])
async def github_pr_create(
title: str,
body: str,
uid: str = ""
) -> ToolResult:
"""创建 GitHub PR(仅在 GitHub 页面可用)"""
...
@tool() # 无 url_patterns,所有页面都可用
async def take_screenshot() -> ToolResult:
"""截图(所有页面都可用)"""
...
```
**支持的模式**:
```python
# 通配符域名
"*.google.com" # 匹配 www.google.com, mail.google.com
"www.google.*" # 匹配 www.google.com, www.google.co.uk
# 路径匹配
"https://github.com/**/issues" # 匹配所有 issues 页面
# 多个模式
url_patterns=["*.github.com", "*.gitlab.com"]
```
**使用过滤后的工具**:
```python
from reson_agent import get_tool_registry
registry = get_tool_registry()
# 根据 URL 获取可用工具
current_url = "https://www.google.com/search?q=test"
tool_names = registry.get_tool_names(current_url)
# 返回:["google_advanced_search", "take_screenshot"](不包含 github_pr_create)
# 获取过滤后的 Schema
schemas = registry.get_schemas_for_url(current_url)
# 传递给 LLM,只包含相关工具
```
**效果**:
| 场景 | 无过滤 | 有过滤 | 节省 |
|------|--------|--------|------|
| 在 Google 页面 | 35 工具 (~5K tokens) | 20 工具 (~3K tokens) | 40% |
| 在 GitHub 页面 | 35 工具 (~5K tokens) | 18 工具 (~2.5K tokens) | 50% |
### 4. 敏感数据处理
**场景**:浏览器自动化需要输入密码、Token,但不想在对话历史中显示明文。
**设置敏感数据**:
```python
sensitive_data = {
# 格式 1:全局密钥(适用于所有域名)
"api_key": "sk-xxxxx",
# 格式 2:域名特定密钥(推荐)
"*.github.com": {
"github_token": "ghp_xxxxx",
"github_password": "my_secret_password"
},
"*.google.com": {
"google_email": "user@example.com",
"google_password": "another_secret",
"google_2fa_bu_2fa_code": "JBSWY3DPEHPK3PXP" # TOTP secret
}
}
```
**LLM 输出占位符**:
```python
# LLM 决定需要输入密码
{
"tool": "browser_input",
"arguments": {
"index": 5,
"text": "github_password" # 占位符
}
}
```
**自动替换**:
```python
# 执行工具前,框架自动替换
registry.execute(
"browser_input",
arguments={"index": 5, "text": "github_password"},
context={"page_url": "https://github.com/login"},
sensitive_data=sensitive_data
)
# 实际执行:
# arguments = {"index": 5, "text": "my_secret_password"}
```
**TOTP 2FA 支持**:
```python
# 密钥以 _bu_2fa_code 结尾,自动生成 TOTP 代码
sensitive_data = {
"*.google.com": {
"google_2fa_bu_2fa_code": "JBSWY3DPEHPK3PXP"
}
}
# LLM 输出
{
"text": "google_2fa_bu_2fa_code"
}
# 自动替换为当前的 6 位数字验证码
{
"text": "123456" # 当前时间的 TOTP 代码
}
```
**完整示例**:
```python
from reson_agent import get_tool_registry, ToolContext
# 设置敏感数据
sensitive_data = {
"*.github.com": {
"github_token": "ghp_xxxxxxxxxxxxx",
"github_2fa_bu_2fa_code": "JBSWY3DPEHPK3PXP"
}
}
# 执行工具(LLM 输出的参数包含占位符)
result = await registry.execute(
"github_api_call",
arguments={
"endpoint": "/user",
"token": "github_token",
"totp": "github_2fa_bu_2fa_code"
},
context={"page_url": "https://github.com"},
sensitive_data=sensitive_data
)
# 实际调用时参数已被替换:
# {
# "endpoint": "/user",
# "token": "ghp_xxxxxxxxxxxxx",
# "totp": "123456"
# }
```
**安全性**:
- ✅ 对话历史中只有 `key` 占位符
- ✅ 实际密码仅在执行时注入
- ✅ 域名匹配防止密钥泄露到错误的网站
- ✅ TOTP 验证码实时生成,无需手动输入
### 5. 工具使用统计
**自动记录**:
每个工具调用自动记录:
- 调用次数
- 成功/失败次数
- 平均执行时间
- 最后调用时间
**查询统计**:
```python
from reson_agent import get_tool_registry
registry = get_tool_registry()
# 获取所有工具统计
all_stats = registry.get_stats()
print(all_stats)
# {
# "search_notes": {
# "call_count": 145,
# "success_count": 142,
# "failure_count": 3,
# "average_duration": 0.32,
# "success_rate": 0.979,
# "last_called": 1704123456.78
# },
# ...
# }
# 获取单个工具统计
search_stats = registry.get_stats("search_notes")
# 获取 Top 工具
top_tools = registry.get_top_tools(limit=5, by="call_count")
# ['search_notes', 'read_file', 'browser_click', ...]
top_by_success = registry.get_top_tools(limit=5, by="success_rate")
fastest_tools = registry.get_top_tools(limit=5, by="average_duration")
```
**优化工具排序**:
```python
# 根据使用频率优化工具顺序
def get_optimized_schemas(registry, current_url):
# 获取可用工具
tool_names = registry.get_tool_names(current_url)
# 按调用次数排序(高频工具排前面)
all_stats = registry.get_stats()
sorted_tools = sorted(
tool_names,
key=lambda name: all_stats.get(name, {}).get("call_count", 0),
reverse=True
)
# 返回排序后的 Schema
return registry.get_schemas(sorted_tools)
```
**监控和告警**:
```python
# 监控工具失败率
stats = registry.get_stats()
for tool_name, tool_stats in stats.items():
if tool_stats["call_count"] > 10 and tool_stats["success_rate"] < 0.8:
logger.warning(
f"Tool {tool_name} has low success rate: "
f"{tool_stats['success_rate']:.1%} "
f"({tool_stats['failure_count']}/{tool_stats['call_count']} failures)"
)
# 监控执行时间
for tool_name, tool_stats in stats.items():
if tool_stats["average_duration"] > 5.0:
logger.warning(
f"Tool {tool_name} is slow: "
f"average {tool_stats['average_duration']:.2f}s"
)
```
### 6. 组合使用
**完整示例:浏览器自动化工具**
```python
@tool(
requires_confirmation=False,
editable_params=["query"],
url_patterns=["*.google.com"],
display={
"zh": {"name": "Google 搜索", "params": {"query": "搜索关键词"}},
"en": {"name": "Google Search", "params": {"query": "Query"}}
}
)
async def google_search(
query: str,
ctx: ToolContext,
uid: str = ""
) -> ToolResult:
"""
在 Google 执行搜索
仅在 Google 页面可用,支持敏感数据注入
"""
# 使用浏览器会话
if not ctx.browser_session:
return ToolResult(title="Error", error="Browser session not available")
# 敏感数据已在 registry.execute() 中自动处理
# 例如 query 中的 api_key 已被替换
# 执行搜索
await ctx.browser_session.navigate(f"https://google.com/search?q={query}")
# 提取结果
results = await ctx.browser_session.extract_results()
return ToolResult(
title=f"Search results for {query}",
output=json.dumps(results),
long_term_memory=f"Searched Google for '{query}', found {len(results)} results",
include_output_only_once=True
)
```
**使用**:
```python
# 设置环境
registry = get_tool_registry()
sensitive_data = {"*.google.com": {"search_api_key": "sk-xxxxx"}}
# Agent 在 Google 页面时
current_url = "https://www.google.com"
# 获取工具(自动过滤)
tool_names = registry.get_tool_names(current_url)
# 包含 google_search(匹配 *.google.com)
# LLM 决定使用工具(可能包含敏感占位符)
tool_call = {
"name": "google_search",
"arguments": {
"query": "site:github.com search_api_key"
}
}
# 执行(自动替换敏感数据)
result = await registry.execute(
tool_call["name"],
tool_call["arguments"],
context={"page_url": current_url, "browser_session": browser},
sensitive_data=sensitive_data
)
# 查看统计
stats = registry.get_stats("google_search")
print(f"Success rate: {stats['success_rate']:.1%}")
```
---
## 内置基础工具
> 参考 opencode 实现的文件操作和命令执行工具
框架提供一组内置的基础工具,用于文件读取、编辑、搜索和命令执行等常见任务。这些工具参考了 [opencode](https://github.com/anomalyco/opencode) 的成熟设计,在 Python 中重新实现。
**实现位置**:
- 工具实现:`agent/tools/builtin/`
- 适配器层:`agent/tools/adapters/`
- OpenCode 参考:`vendor/opencode/` (git submodule)
**详细文档**:参考 [`docs/tools-adapters.md`](./tools-adapters.md)
### 可用工具
| 工具 | 功能 | 参考 |
|------|------|------|
| `read_file` | 读取文件内容(支持图片、PDF) | opencode read.ts |
| `edit_file` | 智能文件编辑(多种匹配策略) | opencode edit.ts |
| `write_file` | 写入文件(创建或覆盖) | opencode write.ts |
| `bash_command` | 执行 shell 命令 | opencode bash.ts |
| `glob_files` | 文件模式匹配 | opencode glob.ts |
| `grep_content` | 内容搜索(正则表达式) | opencode grep.ts |
| `agent` | 创建 Agent 执行任务(单任务 delegate / 多任务并行 explore) | 自研 |
| `evaluate` | 评估目标执行结果是否满足要求 | 自研 |
### Agent 工具
创建子 Agent 执行任务。通过 `task` 参数的类型自动区分模式:
| task 类型 | 模式 | 并行执行 | 工具权限 |
|-----------|------|---------|---------|
| `str`(单任务) | delegate | ❌ | 完整(除 agent/evaluate 外) |
| `List[str]`(多任务) | explore | ✅ | 只读(read_file, grep_content, glob_files, goal) |
```python
@tool(description="创建 Agent 执行任务")
async def agent(
task: Union[str, List[str]],
messages: Optional[Union[Messages, List[Messages]]] = None,
continue_from: Optional[str] = None,
context: Optional[dict] = None,
) -> Dict[str, Any]:
```
**messages 参数**:
- `None`:无预置消息
- `Messages`(1D 列表):所有 agent 共享
- `List[Messages]`(2D 列表):per-agent 独立消息
运行时判断:`messages[0]` 是 dict → 1D 共享;是 list → 2D per-agent。
**单任务(delegate)**:
- 适合委托专门任务(如代码分析、文档生成)
- 完整工具权限,可执行复杂操作
- 支持 `continue_from` 参数续跑已有 Sub-Trace
**多任务(explore)**:
- 适合对比多个方案(如技术选型、架构设计)
- 使用 `asyncio.gather()` 并行执行,显著提升效率
- 每个任务创建独立的 Sub-Trace,互不干扰
- 只读权限(文件系统层面),可使用 goal 工具管理计划
- 不支持 `continue_from`
### Evaluate 工具
评估指定 Goal 的执行结果,提供质量评估和改进建议。
```python
@tool(description="评估目标执行结果是否满足要求")
async def evaluate(
messages: Optional[Messages] = None,
target_goal_id: Optional[str] = None,
continue_from: Optional[str] = None,
context: Optional[dict] = None,
) -> Dict[str, Any]:
```
- 无 `criteria` 参数——代码自动从 GoalTree 注入目标描述
- 模型把执行结果和上下文放在 `messages` 中
- `target_goal_id` 默认为当前 `goal_id`
- 只读工具权限
- 返回评估结论和改进建议
**Sub-Trace 结构**:
- 每个 `agent`/`evaluate` 调用创建独立的 Sub-Trace
- Sub-Trace ID 格式:`{parent_id}@{mode}-{序号}-{timestamp}-001`
- 通过 `parent_trace_id` 和 `parent_goal_id` 建立父子关系
- Sub-Trace 信息存储在独立的 trace 目录中
**Goal 集成**:
- `agent`/`evaluate` 调用会将 Goal 标记为 `type: "agent_call"`
- `agent_call_mode` 记录使用的模式
- `sub_trace_ids` 记录所有创建的 Sub-Trace
- Goal 完成后,`summary` 包含格式化的汇总结果
**实现位置**:`agent/tools/builtin/subagent.py`
### 快速使用
```python
from agent.tools.builtin import read_file, edit_file, bash_command
# 读取文件
result = await read_file(file_path="config.py", limit=100)
print(result.output)
# 编辑文件(智能匹配)
result = await edit_file(
file_path="config.py",
old_string="DEBUG = True",
new_string="DEBUG = False"
)
# 执行命令
result = await bash_command(
command="git status",
timeout=30,
description="Check git status"
)
```
### 核心特性
**Read Tool 特性**:
- 二进制文件检测
- 分页读取(offset/limit)
- 行长度和字节限制
- 图片/PDF 支持
**Edit Tool 特性**:
- 多种智能匹配策略:
- SimpleReplacer - 精确匹配
- LineTrimmedReplacer - 忽略行首尾空白
- WhitespaceNormalizedReplacer - 空白归一化
- 自动生成 unified diff
- 唯一性检查(防止错误替换)
**Bash Tool 特性**:
- 异步执行
- 超时控制(默认 120 秒)
- 工作目录设置
- 输出截断(防止过长)
### 更新 OpenCode 参考
内置工具参考 `vendor/opencode/` 中的实现,通过 git submodule 管理:
```bash
# 更新 opencode 参考
cd vendor/opencode
git pull origin main
cd ../..
git add vendor/opencode
git commit -m "chore: update opencode reference"
# 查看最近变更
cd vendor/opencode
git log --oneline --since="1 month ago" -- packages/opencode/src/tool/
```
更新后,检查是否需要同步改进到 Python 实现。
---
## 集成 Browser-Use
### 适配器模式
将 Browser-Use 的 25 个工具适配为你的工具系统:
```python
from browser_use import BrowserSession, Tools as BrowserUseTools
from reson_agent import tool, ToolResult, ToolContext
class BrowserToolsAdapter:
"""Browser-Use 工具适配器"""
def __init__(self):
self.session = BrowserSession(headless=False)
self.browser_tools = BrowserUseTools()
async def __aenter__(self):
await self.session.__aenter__()
return self
async def __aexit__(self, *args):
await self.session.__aexit__(*args)
def register_all(self, registry):
"""批量注册所有 Browser-Use 工具"""
for action_name, registered_action in self.browser_tools.registry.actions.items():
self._adapt_action(registry, action_name, registered_action)
def _adapt_action(self, registry, action_name, registered_action):
"""适配单个 Browser-Use action"""
@tool()
async def adapted_tool(args: dict, ctx: ToolContext) -> ToolResult:
# 构建 Browser-Use 需要的 special context
special_context = {
'browser_session': self.session,
'page_url': ctx.page_url,
'file_system': ctx.file_system,
}
# 执行 Browser-Use action
result = await registered_action.function(
params=registered_action.param_model(**args),
**special_context
)
# 转换 ActionResult -> ToolResult
return ToolResult(
title=action_name,
output=result.extracted_content or '',
long_term_memory=result.long_term_memory,
include_output_only_once=result.include_extracted_content_only_once,
error=result.error,
attachments=result.attachments or [],
images=result.images or [],
metadata=result.metadata or {}
)
# 注册到你的 registry
registry.register(adapted_tool, schema=generate_schema(registered_action))
```
### 使用示例
```python
from reson_agent import AgentRunner
async def main():
async with BrowserToolsAdapter() as browser:
# 创建 Agent
agent = AgentRunner(
task="在 Amazon 找最便宜的 iPhone 15",
tools=[], # 空列表
)
# 批量注册浏览器工具
browser.register_all(agent.tool_registry)
# 现在 Agent 有 25 个浏览器工具 + 其他工具
result = await agent.run()
```
### Context 占用分析
| 工具类型 | 数量 | Token 占用 | 占比(200K) |
|---------|------|-----------|-------------|
| Browser-Use 工具 | 25 | ~4,000 | 2% |
| 你的自定义工具 | 10 | ~1,000 | 0.5% |
| **总计** | **35** | **~5,000** | **2.5%** |
**结论**:完全可接受,且 Prompt Caching 会优化后续调用。
---
## 最佳实践
### 1. 工具命名
```python
# 好:清晰的动词 + 名词
@tool()
async def search_notes(...): ...
@tool()
async def create_document(...): ...
# 不好:模糊或过长
@tool()
async def do_something(...): ...
@tool()
async def search_and_filter_notes_with_advanced_options(...): ...
```
### 2. 返回结构化数据
```python
# 好:返回 ToolResult 或结构化字典
@tool()
async def get_weather(city: str) -> ToolResult:
data = fetch_weather(city)
return ToolResult(
title=f"Weather in {city}",
output=json.dumps(data, indent=2)
)
# 不好:返回纯文本
@tool()
async def get_weather(city: str) -> str:
return "The weather is sunny, 25°C, humidity 60%" # 难以解析
```
### 3. 错误处理
```python
# 好:捕获异常并返回 ToolResult
@tool()
async def risky_operation() -> ToolResult:
try:
result = dangerous_call()
return ToolResult(title="Success", output=result)
except Exception as e:
logger.error(f"Operation failed: {e}")
return ToolResult(title="Failed", error=str(e))
# 不好:让异常传播(会中断 Agent 循环)
@tool()
async def risky_operation() -> str:
return dangerous_call() # 可能抛出异常
```
### 4. 记忆管理
```python
# 好:大量数据用 include_output_only_once
@tool()
async def fetch_all_logs() -> ToolResult:
logs = get_last_10000_logs() # 很大
return ToolResult(
title="Fetched logs",
output=logs,
long_term_memory=f"Fetched {len(logs)} log entries",
include_output_only_once=True # 只给 LLM 看一次
)
# 不好:大量数据每次都传给 LLM
@tool()
async def fetch_all_logs() -> str:
return get_last_10000_logs() # 每次都占用 context
```
### 5. 工具粒度
```python
# 好:单一职责,细粒度
@tool()
async def search_notes(query: str) -> ToolResult: ...
@tool()
async def get_note_detail(note_id: str) -> ToolResult: ...
@tool()
async def update_note(note_id: str, content: str) -> ToolResult: ...
# 不好:功能过多,难以使用
@tool()
async def manage_notes(
action: Literal["search", "get", "update", "delete"],
query: Optional[str] = None,
note_id: Optional[str] = None,
content: Optional[str] = None
) -> ToolResult:
# 太复杂,LLM 容易用错
...
```
### 6. 文档和示例
```python
@tool()
async def search_notes(
query: str,
filters: Optional[Dict[str, Any]] = None,
sort_by: str = "relevance",
limit: int = 10,
uid: str = ""
) -> ToolResult:
"""
搜索用户的笔记
使用语义搜索查找相关笔记,支持过滤和排序。
Args:
query: 搜索关键词(必需)
filters: 过滤条件,例如 {"type": "markdown", "tags": ["work"]}
sort_by: 排序方式,可选 "relevance" | "date" | "title"
limit: 返回结果数量,默认 10,最大 100
Returns:
ToolResult 包含搜索结果列表
Example:
搜索包含 "项目计划" 的工作笔记:
{
"query": "项目计划",
"filters": {"tags": ["work"]},
"limit": 5
}
"""
...
```
---
## 总结
| 特性 | 状态 | 说明 |
|------|------|------|
| **基础注册** | ✅ 已实现 | `@tool()` 装饰器 |
| **Schema 生成** | ✅ 已实现 | 自动从函数签名生成 |
| **双层记忆** | ✅ 已实现 | `ToolResult` 支持 long_term_memory |
| **依赖注入** | ✅ 已实现 | `ToolContext` 提供上下文 |
| **UI 元数据** | ✅ 已实现 | `display`, `requires_confirmation`, `editable_params` |
| **域名过滤** | ✅ **已实现** | `url_patterns` 参数 + URL 匹配器 |
| **敏感数据** | ✅ **已实现** | `` 占位符 + TOTP 支持 |
| **工具统计** | ✅ **已实现** | 自动记录调用次数、成功率、执行时间 |
**核心设计原则**:
1. **简单优先**:最简工具只需要一个装饰器
2. **按需扩展**:高级特性可选
3. **类型安全**:充分利用 Python 类型注解
4. **灵活集成**:支持各种工具库(Browser-Use, MCP 等)
5. **可观测性**:内建统计和监控能力