Agent/examples/tool_examples.py

"""
Tool Examples - 工具装饰器使用样例

本文件展示 @tool 装饰器的各种用法，参考 Resonote 项目的实际工具实现。

样例包括：
1. 基础工具（最简形式）
2. 带 i18n 展示信息的工具
3. 带可编辑参数的工具
4. 需要用户确认的危险操作
5. 带 context 参数的工具
6. 同步工具

注意：
- uid 参数会由框架自动注入，不需要用户传递
- context 参数用于传递额外上下文（如当前阅读位置等）
- 返回值会自动序列化为 JSON 字符串
"""

from typing import List, Dict, Any, Optional
from reson_agent import tool


# ============================================================
# 1. 基础工具（最简形式）
# ============================================================

@tool()
async def hello_world(name: str, uid: str = "") -> Dict[str, str]:
    """
    最简单的工具示例

    Args:
        name: 要问候的名字
        uid: 用户ID（自动注入）

    Returns:
        包含问候语的字典
    """
    return {"greeting": f"Hello, {name}!"}


# ============================================================
# 2. 带 i18n 展示信息的工具
# ============================================================

@tool(
    display={
        "zh": {
            "name": "搜索内容",
            "params": {
                "query": "搜索关键词",
                "limit": "返回数量"
            }
        },
        "en": {
            "name": "Search Content",
            "params": {
                "query": "Search query",
                "limit": "Number of results"
            }
        }
    }
)
async def search_content(
    query: str,
    limit: int = 10,
    uid: str = ""
) -> List[Dict[str, Any]]:
    """
    搜索用户的内容

    使用语义搜索查找相关内容。display 参数用于前端展示：
    - 工具名称会根据用户语言显示为"搜索内容"或"Search Content"
    - 参数名称也会相应翻译

    Args:
        query: 搜索查询文本
        limit: 返回结果数量（默认10）
        uid: 用户ID（自动注入）

    Returns:
        搜索结果列表，每个包含 id, title, content, score
    """
    # 实际实现中会调用向量搜索
    # 这里只是示例
    return [
        {
            "id": "doc_001",
            "title": f"关于 {query} 的文档",
            "content": f"这是与 {query} 相关的内容...",
            "score": 0.95
        }
    ]


# ============================================================
# 3. 带可编辑参数的工具
# ============================================================

@tool(
    editable_params=["query", "filters"],
    display={
        "zh": {
            "name": "高级搜索",
            "params": {
                "query": "搜索关键词",
                "filters": "过滤条件",
                "sort_by": "排序方式"
            }
        },
        "en": {
            "name": "Advanced Search",
            "params": {
                "query": "Search query",
                "filters": "Filters",
                "sort_by": "Sort by"
            }
        }
    }
)
async def advanced_search(
    query: str,
    filters: Optional[Dict[str, Any]] = None,
    sort_by: str = "relevance",
    limit: int = 20,
    uid: str = ""
) -> Dict[str, Any]:
    """
    高级搜索工具（允许用户编辑参数）

    editable_params 指定哪些参数允许用户在 LLM 生成后编辑：
    - LLM 会先生成 query 和 filters
    - 用户可以在确认前修改这些参数
    - 适用于搜索、创建等需要用户微调的场景

    Args:
        query: 搜索查询
        filters: 过滤条件（如 {"type": "note", "date_range": "7d"}）
        sort_by: 排序方式（relevance/date/title）
        limit: 返回数量
        uid: 用户ID（自动注入）

    Returns:
        搜索结果和元数据
    """
    return {
        "results": [
            {"id": "1", "title": "Result 1", "score": 0.9},
            {"id": "2", "title": "Result 2", "score": 0.8},
        ],
        "total": 42,
        "query": query,
        "filters_applied": filters or {},
        "sort_by": sort_by
    }


# ============================================================
# 4. 需要用户确认的危险操作
# ============================================================

@tool(
    requires_confirmation=True,
    display={
        "zh": {
            "name": "删除内容",
            "params": {
                "content_id": "内容ID",
                "permanent": "永久删除"
            }
        },
        "en": {
            "name": "Delete Content",
            "params": {
                "content_id": "Content ID",
                "permanent": "Permanent delete"
            }
        }
    }
)
async def delete_content(
    content_id: str,
    permanent: bool = False,
    uid: str = ""
) -> Dict[str, Any]:
    """
    删除内容（需要用户确认）

    requires_confirmation=True 表示这是一个危险操作：
    - LLM 调用此工具时，不会立即执行
    - 会先向用户展示操作详情，等待确认
    - 用户确认后才会真正执行

    适用场景：
    - 删除操作
    - 发送消息
    - 修改重要设置
    - 任何不可逆操作

    Args:
        content_id: 要删除的内容ID
        permanent: 是否永久删除（False=移到回收站）
        uid: 用户ID（自动注入）

    Returns:
        删除结果
    """
    # 实际实现中会执行删除
    return {
        "success": True,
        "content_id": content_id,
        "permanent": permanent,
        "message": f"内容 {content_id} 已{'永久删除' if permanent else '移到回收站'}"
    }


# ============================================================
# 5. 带 context 参数的工具
# ============================================================

@tool(
    display={
        "zh": {
            "name": "获取相关推荐",
            "params": {
                "top_k": "推荐数量"
            }
        },
        "en": {
            "name": "Get Recommendations",
            "params": {
                "top_k": "Number of recommendations"
            }
        }
    }
)
async def get_recommendations(
    top_k: int = 5,
    uid: str = "",
    context: Optional[Dict[str, Any]] = None
) -> List[Dict[str, Any]]:
    """
    获取相关推荐（使用 context 获取额外信息）

    context 参数用于传递执行上下文，由框架自动注入：
    - 当前阅读位置 (current_location)
    - 当前会话 ID (session_id)
    - 排除的内容 ID (exclude_ids)
    - 等等...

    框架会检查函数签名，如果有 context 参数就自动传入。

    Args:
        top_k: 返回推荐数量
        uid: 用户ID（自动注入）
        context: 执行上下文（自动注入）

    Returns:
        推荐列表
    """
    # 从 context 中提取信息
    current_location = None
    exclude_ids = []

    if context:
        current_location = context.get("current_location")
        exclude_ids = context.get("exclude_ids", [])

    # 实际实现中会根据 context 生成推荐
    return [
        {
            "id": "rec_001",
            "title": "推荐内容 1",
            "reason": f"基于当前位置 {current_location}" if current_location else "基于您的兴趣"
        },
        {
            "id": "rec_002",
            "title": "推荐内容 2",
            "reason": "热门内容"
        }
    ]


# ============================================================
# 6. 同步工具（非 async）
# ============================================================

@tool(
    display={
        "zh": {
            "name": "格式化文本",
            "params": {
                "text": "原始文本",
                "format_type": "格式类型"
            }
        },
        "en": {
            "name": "Format Text",
            "params": {
                "text": "Raw text",
                "format_type": "Format type"
            }
        }
    }
)
def format_text(
    text: str,
    format_type: str = "markdown",
    uid: str = ""
) -> str:
    """
    格式化文本（同步工具）

    不需要 async 的工具可以定义为普通函数。
    框架会自动检测并正确调用。

    适用于：
    - 纯计算操作
    - 文本处理
    - 不需要 I/O 的操作

    Args:
        text: 要格式化的文本
        format_type: 格式类型（markdown/plain/html）
        uid: 用户ID（自动注入）

    Returns:
        格式化后的文本
    """
    if format_type == "markdown":
        return f"**{text}**"
    elif format_type == "html":
        return f"<p>{text}</p>"
    else:
        return text


# ============================================================
# 7. 复杂返回类型的工具
# ============================================================

@tool(
    display={
        "zh": {
            "name": "分析内容",
            "params": {
                "content_id": "内容ID",
                "analysis_types": "分析类型"
            }
        },
        "en": {
            "name": "Analyze Content",
            "params": {
                "content_id": "Content ID",
                "analysis_types": "Analysis types"
            }
        }
    }
)
async def analyze_content(
    content_id: str,
    analysis_types: Optional[List[str]] = None,
    uid: str = ""
) -> Dict[str, Any]:
    """
    分析内容（复杂返回类型）

    展示如何返回复杂的嵌套结构。
    返回值会自动序列化为 JSON。

    Args:
        content_id: 要分析的内容ID
        analysis_types: 分析类型列表（sentiment/keywords/summary）
        uid: 用户ID（自动注入）

    Returns:
        分析结果，包含多种分析数据
    """
    types = analysis_types or ["sentiment", "keywords"]

    result = {
        "content_id": content_id,
        "analyses": {}
    }

    if "sentiment" in types:
        result["analyses"]["sentiment"] = {
            "score": 0.8,
            "label": "positive",
            "confidence": 0.92
        }

    if "keywords" in types:
        result["analyses"]["keywords"] = [
            {"word": "AI", "weight": 0.9},
            {"word": "学习", "weight": 0.7},
            {"word": "创新", "weight": 0.6}
        ]

    if "summary" in types:
        result["analyses"]["summary"] = {
            "short": "这是一篇关于AI学习的文章",
            "long": "本文详细介绍了AI在学习领域的应用..."
        }

    return result


# ============================================================
# 使用示例
# ============================================================

if __name__ == "__main__":
    import asyncio
    from reson_agent import get_tool_registry

    async def main():
        # 获取全局注册表
        registry = get_tool_registry()

        # 查看已注册的工具
        print("已注册的工具:")
        for name in registry.get_tool_names():
            print(f"  - {name}")

        # 获取工具 Schema
        schemas = registry.get_schemas(["search_content"])
        print("\nsearch_content Schema:")
        import json
        print(json.dumps(schemas[0], indent=2, ensure_ascii=False))

        # 执行工具
        result = await registry.execute(
            "search_content",
            {"query": "人工智能", "limit": 5},
            uid="user123"
        )
        print("\n执行结果:")
        print(result)

        # 获取 UI 元数据
        ui_meta = registry.get_ui_metadata(locale="zh", tool_names=["advanced_search"])
        print("\nUI 元数据 (中文):")
        print(json.dumps(ui_meta, indent=2, ensure_ascii=False))

    asyncio.run(main())