工具系统文档

Agent 框架的工具系统：定义、注册、执行工具调用。

核心概念

三个核心类型

from reson_agent import tool, ToolResult, ToolContext

@tool()
async def my_tool(arg: str, ctx: ToolContext) -> 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
3. LLM 选择工具并生成参数
   ↓ registry.execute(name, args)
4. 注入 uid 和 context
   ↓ 调用工具函数
5. 返回 ToolResult
   ↓ 转换为 LLM 消息
6. 添加到对话历史

定义工具

最简形式

from reson_agent import tool

@tool()
async def hello(name: str, uid: str = "") -> str:
    """向用户问好"""
    return f"Hello, {name}!"

要点：

uid 参数由框架自动注入（用户不传递）
可以是同步或异步函数
返回值自动序列化为 JSON

带完整注释

@tool()
async def search_notes(
    query: str,
    limit: int = 10,
    uid: str = ""
) -> str:
    """
    搜索用户的笔记

    Args:
        query: 搜索关键词
        limit: 返回结果数量

    Returns:
        JSON 格式的搜索结果
    """
    # 自动从 docstring 提取 function description 和 parameter descriptions
    ...

带 UI 元数据

@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 和记忆管理

基础用法

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 支持双层记忆：

@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

错误处理

@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)
        )

附件和图片

@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 参数，框架自动注入。

from reson_agent import ToolContext

@tool()
async def get_current_state(ctx: ToolContext) -> ToolResult:
    return ToolResult(
        title="Current state",
        output=f"Trace ID: {ctx.trace_id}\nStep ID: {ctx.step_id}"
    )

ToolContext 字段

class ToolContext(Protocol):
    # 基础字段（所有工具）
    trace_id: str               # 当前 Trace ID
    step_id: str                # 当前 Step ID
    uid: Optional[str]          # 用户 ID

    # 浏览器相关（Browser-Use 集成）
    browser_session: Optional[Any]      # 浏览器会话
    page_url: Optional[str]             # 当前页面 URL
    file_system: Optional[Any]          # 文件系统访问
    sensitive_data: Optional[Dict]      # 敏感数据

    # 扩展字段
    context: Optional[Dict[str, Any]]   # 额外上下文

使用示例

@tool()
async def analyze_current_page(ctx: ToolContext) -> ToolResult:
    """分析当前浏览器页面"""

    if not ctx.browser_session:
        return ToolResult(
            title="Error",
            error="Browser session not available"
        )

    # 使用浏览器会话
    page_content = await ctx.browser_session.get_content()

    return ToolResult(
        title=f"Analyzed {ctx.page_url}",
        output=page_content,
        long_term_memory=f"Analyzed page at {ctx.page_url}"
    )

创建 ToolContext

from reson_agent import ToolContextImpl

ctx = ToolContextImpl(
    trace_id="trace_123",
    step_id="step_456",
    uid="user_789",
    page_url="https://example.com"
)

# 执行工具
result = await registry.execute(
    "analyze_current_page",
    arguments={},
    context=ctx
)

高级特性

1. 需要用户确认

@tool(requires_confirmation=True)
async def delete_all_notes(uid: str = "") -> ToolResult:
    """删除所有笔记（危险操作）"""
    # 执行前会等待用户确认
    ...

适用场景：

删除操作
发送消息
修改重要设置
任何不可逆操作

2. 可编辑参数

@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 占用。

@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:
    """截图（所有页面都可用）"""
    ...

支持的模式：

# 通配符域名
"*.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"]

使用过滤后的工具：

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，但不想在对话历史中显示明文。

设置敏感数据：

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 输出占位符：

# LLM 决定需要输入密码
{
    "tool": "browser_input",
    "arguments": {
        "index": 5,
        "text": "<secret>github_password</secret>"  # 占位符
    }
}

自动替换：

# 执行工具前，框架自动替换
registry.execute(
    "browser_input",
    arguments={"index": 5, "text": "<secret>github_password</secret>"},
    context={"page_url": "https://github.com/login"},
    sensitive_data=sensitive_data
)

# 实际执行：
# arguments = {"index": 5, "text": "my_secret_password"}

TOTP 2FA 支持：

# 密钥以 _bu_2fa_code 结尾，自动生成 TOTP 代码
sensitive_data = {
    "*.google.com": {
        "google_2fa_bu_2fa_code": "JBSWY3DPEHPK3PXP"
    }
}

# LLM 输出
{
    "text": "<secret>google_2fa_bu_2fa_code</secret>"
}

# 自动替换为当前的 6 位数字验证码
{
    "text": "123456"  # 当前时间的 TOTP 代码
}

完整示例：

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": "<secret>github_token</secret>",
        "totp": "<secret>github_2fa_bu_2fa_code</secret>"
    },
    context={"page_url": "https://github.com"},
    sensitive_data=sensitive_data
)

# 实际调用时参数已被替换：
# {
#     "endpoint": "/user",
#     "token": "ghp_xxxxxxxxxxxxx",
#     "totp": "123456"
# }

安全性：

✅ 对话历史中只有 <secret>key</secret> 占位符
✅ 实际密码仅在执行时注入
✅ 域名匹配防止密钥泄露到错误的网站
✅ TOTP 验证码实时生成，无需手动输入

5. 工具使用统计

自动记录：

每个工具调用自动记录：

调用次数
成功/失败次数
平均执行时间
最后调用时间

查询统计：

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")

优化工具排序：

# 根据使用频率优化工具顺序
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)

监控和告警：

# 监控工具失败率
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. 组合使用

完整示例：浏览器自动化工具

@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 中的 <secret>api_key</secret> 已被替换

    # 执行搜索
    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
    )

使用：

# 设置环境
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 <secret>search_api_key</secret>"
    }
}

# 执行（自动替换敏感数据）
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%}")

集成 Browser-Use

适配器模式

将 Browser-Use 的 25 个工具适配为你的工具系统：

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))

使用示例

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. 工具命名

# 好：清晰的动词 + 名词
@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. 返回结构化数据

# 好：返回 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. 错误处理

# 好：捕获异常并返回 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. 记忆管理

# 好：大量数据用 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. 工具粒度

# 好：单一职责，细粒度
@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. 文档和示例

@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 匹配器
敏感数据	✅ 已实现	`<secret>` 占位符 + TOTP 支持
工具统计	✅ 已实现	自动记录调用次数、成功率、执行时间

核心设计原则：

简单优先：最简工具只需要一个装饰器
按需扩展：高级特性可选
类型安全：充分利用 Python 类型注解
灵活集成：支持各种工具库（Browser-Use, MCP 等）
可观测性：内建统计和监控能力

tools.md 23 KB

Trvalý odkaz Historie Surový

工具系统文档

目录

核心概念

三个核心类型

工具的生命周期

定义工具

最简形式

带完整注释

带 UI 元数据

ToolResult 和记忆管理

基础用法

双层记忆管理

错误处理

附件和图片

ToolContext 和依赖注入

基本概念

ToolContext 字段

使用示例

创建 ToolContext

高级特性

1. 需要用户确认

2. 可编辑参数

3. 域名过滤（URL Patterns）

4. 敏感数据处理

5. 工具使用统计

6. 组合使用

集成 Browser-Use

适配器模式

使用示例

Context 占用分析

最佳实践

1. 工具命名

2. 返回结构化数据

3. 错误处理

4. 记忆管理

5. 工具粒度

6. 文档和示例

总结

tools.md 23 KB Trvalý odkaz Historie Surový

工具系统文档

目录

核心概念

三个核心类型

工具的生命周期

定义工具

最简形式

带完整注释

带 UI 元数据

ToolResult 和记忆管理

基础用法

双层记忆管理

错误处理

附件和图片

ToolContext 和依赖注入

基本概念

ToolContext 字段

使用示例

创建 ToolContext

高级特性

1. 需要用户确认

2. 可编辑参数

3. 域名过滤（URL Patterns）

4. 敏感数据处理

5. 工具使用统计

6. 组合使用

集成 Browser-Use

适配器模式

使用示例

Context 占用分析

最佳实践

1. 工具命名

2. 返回结构化数据

3. 错误处理

4. 记忆管理

5. 工具粒度

6. 文档和示例

总结

tools.md 23 KB

Trvalý odkaz Historie Surový