butler/tools/tool_taxonomy.py
Rama main · réplica de lectura del repositorio original.
"""Tool taxonomy — organizes tools by domain for intelligent discovery.
Integrates with the Experience Tree taxonomy (taxonomy.py) to provide
consistent domain-based tool categorization.
"""
from __future__ import annotations
from typing import Any
# Tool taxonomy aligned with experience tree domains
# Each domain maps to tools that are commonly used in that context
TOOL_TAXONOMY: dict[str, dict[str, Any]] = {
"agent_dev": {
"name": "Agent 开发",
"description": "LLM 编排、prompt 工程、工具集成",
"tools": [
"delegate_task",
"skill_view",
"skills_list",
"run_workflow",
"execute_code",
"call_llm",
],
"keywords": ["agent", "llm", "prompt", "tool", "skill", "编排", "推理"],
},
"database": {
"name": "数据库使用",
"description": "SQL/NoSQL 操作、迁移、性能调优",
"tools": [
"execute_code",
"terminal",
"read_file",
"patch",
],
"keywords": ["sql", "postgres", "mysql", "redis", "mongodb", "数据库", "索引"],
},
"llm_usage": {
"name": "大模型使用",
"description": "模型选择、参数调优、prompt 优化",
"tools": [
"call_llm",
"embed_text",
"web_search",
"delegate_task",
],
"keywords": ["model", "llm", "gpt", "claude", "deepseek", "模型", "prompt"],
},
"code_engineering": {
"name": "代码工程",
"description": "架构设计、重构、测试、代码审查",
"tools": [
"read_file",
"write_file",
"patch",
"search_files",
"execute_code",
"terminal",
"list_directory",
],
"keywords": ["code", "refactor", "test", "lint", "review", "代码", "架构"],
},
"dev_ops": {
"name": "开发运维",
"description": "CI/CD、容器化、监控、部署",
"tools": [
"terminal",
"run_workflow",
"delegate_task",
"execute_code",
"patch",
],
"keywords": ["docker", "kubernetes", "ci", "cd", "deploy", "容器", "部署"],
},
"network_info": {
"name": "网络信息查找",
"description": "搜索策略、信息抽取、事实核查",
"tools": [
"web_search",
"web_fetch",
"read_file",
"execute_code",
],
"keywords": ["search", "web", "fetch", "url", "搜索", "网页", "信息"],
},
"daily_life": {
"name": "日常生活",
"description": "用户偏好、习惯、提醒、日程",
"tools": [
"reminder",
"memo",
"habits",
"delegate_task",
],
"keywords": ["日程", "提醒", "偏好", "习惯", "天气", "生活"],
},
"project_mgmt": {
"name": "项目管理",
"description": "需求拆解、进度追踪、优先级排序",
"tools": [
"project_todos",
"session_todos",
"delegate_task",
"run_workflow",
],
"keywords": ["task", "progress", "backlog", "priority", "任务", "进度", "待办"],
},
"security": {
"name": "安全防护",
"description": "安全漏洞、加密、认证、授权",
"tools": [
"execute_code",
"terminal",
"patch",
"search_files",
],
"keywords": ["security", "encrypt", "jwt", "oauth", "安全", "加密", "认证"],
},
"data_science": {
"name": "数据科学",
"description": "数据分析、机器学习、数据可视化",
"tools": [
"execute_code",
"read_file",
"web_search",
],
"keywords": ["data", "pandas", "numpy", "ml", "数据", "分析", "机器学习"],
},
"math_reasoning": {
"name": "数学推导",
"description": "数学公式、逻辑推导、算法分析",
"tools": [
"execute_code",
"delegate_task",
],
"keywords": ["数学", "公式", "推导", "算法", "复杂度", "概率"],
},
"troubleshooting": {
"name": "故障定位",
"description": "问题诊断、错误分析、调试技巧",
"tools": [
"terminal",
"read_file",
"search_files",
"execute_code",
],
"keywords": ["debug", "错误", "异常", "traceback", "调试", "排查", "bug"],
},
"system_admin": {
"name": "系统管理",
"description": "操作系统、网络、服务器管理",
"tools": [
"terminal",
"execute_code",
"read_file",
"patch",
],
"keywords": ["linux", "bash", "shell", "network", "server", "系统", "进程"],
},
}
# Reverse mapping: tool_name -> list of domains
_TOOL_TO_DOMAINS: dict[str, list[str]] = {}
def _build_reverse_mapping() -> None:
"""Build reverse mapping from tool name to domains."""
global _TOOL_TO_DOMAINS
_TOOL_TO_DOMAINS = {}
for domain_id, domain_info in TOOL_TAXONOMY.items():
for tool_name in domain_info.get("tools", []):
if tool_name not in _TOOL_TO_DOMAINS:
_TOOL_TO_DOMAINS[tool_name] = []
_TOOL_TO_DOMAINS[tool_name].append(domain_id)
_build_reverse_mapping()
def get_tools_by_domain(domain_id: str) -> list[str]:
"""Get all tools for a specific domain."""
domain_info = TOOL_TAXONOMY.get(domain_id, {})
return domain_info.get("tools", [])
def get_domains_for_tool(tool_name: str) -> list[str]:
"""Get all domains a tool belongs to."""
return _TOOL_TO_DOMAINS.get(tool_name, [])
def get_all_tools() -> list[str]:
"""Get all unique tool names from taxonomy."""
return list(_TOOL_TO_DOMAINS.keys())
def get_domain_for_query(query: str) -> str:
"""Determine the most likely domain for a query."""
from butler.memory.experience.domain_router import DomainRouter
router = DomainRouter()
domain_id, _ = router.route(query)
return domain_id
def get_recommended_tools(query: str, top_k: int = 5) -> list[str]:
"""Get recommended tools for a query based on domain taxonomy."""
domain_id = get_domain_for_query(query)
domain_tools = get_tools_by_domain(domain_id)
# Also check query for direct tool mentions
q_lower = query.lower()
all_tools = get_all_tools()
direct_mentions = [t for t in all_tools if t.replace("_", " ") in q_lower or t in q_lower]
# Combine and deduplicate
seen = set()
result = []
for tool in direct_mentions + domain_tools:
if tool not in seen:
seen.add(tool)
result.append(tool)
return result[:top_k]
def get_tool_description(tool_name: str) -> str | None:
"""Get the description for a tool from registry."""
from butler.tools.registry import _REGISTRY, _ensure_builtins
_ensure_builtins()
entry = _REGISTRY.get(tool_name)
if entry:
return entry.description
return None
def get_tool_info(tool_name: str) -> dict[str, Any]:
"""Get comprehensive info about a tool."""
domains = get_domains_for_tool(tool_name)
description = get_tool_description(tool_name)
return {
"name": tool_name,
"domains": domains,
"primary_domain": domains[0] if domains else None,
"description": description,
"is_registered": description is not None,
}