# 工具系统文档 > 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. **可观测性**:内建统计和监控能力