butler/tools/tool_service.py
Rama main · réplica de lectura del repositorio original.
"""Unified tool service — integrates experience, taxonomy, discovery, optimization, and metrics.
This is the main entry point for intelligent tool selection and management.
Usage:
from butler.tools.tool_service import ToolService
service = ToolService()
# Get tool recommendations
recommendations = service.recommend_tools("read a python file")
# Execute with optimization
result, meta = service.execute_tool("read_file", {"path": "test.py"})
"""
from __future__ import annotations
import logging
from typing import Any, Callable
from butler.tools.experience_selector import (
get_tool_selector,
)
from butler.tools.tool_call_optimizer import get_tool_optimizer
from butler.tools.tool_discovery import get_tool_discovery
from butler.tools.tool_metrics import get_tool_metrics, ToolCallRecorder
from butler.tools.tool_taxonomy import (
get_domains_for_tool,
get_recommended_tools,
get_tool_info,
)
logger = logging.getLogger(__name__)
class ToolService:
"""Unified tool service combining all tool intelligence features."""
def __init__(self):
self._selector = get_tool_selector()
self._discovery = get_tool_discovery()
self._optimizer = get_tool_optimizer()
self._metrics = get_tool_metrics()
# ─── Tool Recommendation ───
def recommend_tools(self, query: str, top_k: int = 5) -> list[dict[str, Any]]:
"""Get tool recommendations for a query.
Combines experience-based, taxonomy-based, and semantic discovery.
Args:
query: User query or task description
top_k: Maximum number of recommendations
Returns:
List of tool recommendation dicts
"""
recommendations: dict[str, dict[str, Any]] = {}
# 1. Experience-based selection
exp_recs = self._selector.select_tools(query, top_k=top_k * 2)
for rec in exp_recs:
recommendations[rec.tool_name] = {
"tool_name": rec.tool_name,
"score": rec.score,
"source": rec.source,
"success_rate": rec.success_rate,
"domains": get_domains_for_tool(rec.tool_name),
}
# 2. Semantic discovery
sem_recs = self._discovery.discover(query, top_k=top_k * 2, threshold=0.25)
for rec in sem_recs:
tool_name = rec["tool_name"]
if tool_name in recommendations:
# Boost existing recommendation
recommendations[tool_name]["score"] += rec["similarity"] * 0.3
else:
recommendations[tool_name] = {
"tool_name": tool_name,
"score": rec["similarity"],
"source": "semantic",
"success_rate": 0,
"domains": get_domains_for_tool(tool_name),
}
# 3. Taxonomy fallback
tax_tools = get_recommended_tools(query, top_k=top_k)
for tool_name in tax_tools:
if tool_name not in recommendations:
recommendations[tool_name] = {
"tool_name": tool_name,
"score": 0.2,
"source": "taxonomy",
"success_rate": 0,
"domains": get_domains_for_tool(tool_name),
}
# Sort and return
sorted_recs = sorted(
recommendations.values(),
key=lambda r: r["score"],
reverse=True
)
return sorted_recs[:top_k]
def recommend_for_domain(self, domain_id: str, top_k: int = 10) -> list[dict[str, Any]]:
"""Get tool recommendations for a specific domain."""
from butler.tools.tool_taxonomy import get_tools_by_domain
tools = get_tools_by_domain(domain_id)
return [
{
"tool_name": tool,
"score": 0.5,
"source": "domain_taxonomy",
"success_rate": 0,
"domains": [domain_id],
**get_tool_info(tool),
}
for tool in tools[:top_k]
]
# ─── Tool Execution ───
def execute_tool(
self,
tool_name: str,
args: dict[str, Any],
handler: Callable[[], str] | None = None,
) -> tuple[str, dict[str, Any]]:
"""Execute a tool call with optimization and metrics.
Args:
tool_name: Name of the tool
args: Tool arguments
handler: Function to call (if None, looks up in registry)
Returns:
Tuple of (result, metadata)
"""
recorder = ToolCallRecorder(self._metrics)
recorder.start(tool_name)
try:
if self._optimizer.should_cache(tool_name):
cached = self._optimizer.check_cache(tool_name, args)
if cached is not None:
recorder.end(success=True, cache_hit=True)
return cached, {"cached": True, "duration_ms": 0}
if self._optimizer.should_dedup(tool_name):
if self._optimizer.check_duplicate(tool_name, args):
self._metrics.record_duplicate_block(tool_name)
recorder.end(success=False, error_type="DUPLICATE_BLOCKED")
return '{"ok": false, "code": "DUPLICATE_BLOCKED"}', {"blocked": True}
if handler is None:
from butler.tools.registry import dispatch_tool
result = dispatch_tool(tool_name, args)
else:
result = handler()
if self._optimizer.should_cache(tool_name):
self._optimizer.set_cache(tool_name, args, result)
recorder.end(success=True, cache_hit=False)
return result, {"cached": False}
except Exception as e:
logger.error("Tool %s failed: %s", tool_name, e)
recorder.end(success=False, error_type=type(e).__name__)
raise
# ─── Statistics & Monitoring ───
def get_stats(self) -> dict[str, Any]:
"""Get comprehensive tool statistics."""
return {
"metrics": self._metrics.get_all_stats(),
"optimizer": self._optimizer.get_stats(),
}
def get_tool_report(self, tool_name: str) -> dict[str, Any]:
"""Get a detailed report for a specific tool."""
stats = self._metrics.get_tool_stats(tool_name)
info = get_tool_info(tool_name)
similar = self._discovery.get_similar_tools(tool_name, top_k=5)
return {
**stats,
**info,
"similar_tools": similar,
}
def get_performance_report(self) -> dict[str, Any]:
"""Get a performance summary report."""
stats = self._metrics.get_all_stats()
return {
"session_summary": {
"duration_seconds": stats["session_duration_seconds"],
"total_calls": stats["total_calls"],
"success_rate": stats["overall_success_rate"],
"cache_hit_rate": stats["overall_cache_hit_rate"],
},
"top_by_calls": self._metrics.get_top_tools(by="call_count"),
"slowest": self._metrics.get_slowest_tools(),
"most_errors": self._metrics.get_error_prone_tools(),
}
# ─── Learning ───
def record_usage(self, tool_name: str, query: str, success: bool) -> None:
"""Record a tool usage for learning."""
self._selector.record_tool_usage(tool_name, query, success)
def refresh_discovery(self) -> None:
"""Refresh semantic discovery (after tool changes)."""
self._discovery.refresh()
def clear_cache(self) -> None:
"""Clear all cached tool results."""
self._optimizer.clear_cache()
# Singleton
_service: ToolService | None = None
def get_tool_service() -> ToolService:
"""Get the singleton tool service instance."""
global _service
if _service is None:
_service = ToolService()
return _service
def recommend_tools(query: str, top_k: int = 5) -> list[dict[str, Any]]:
"""Convenience function to get tool recommendations."""
return get_tool_service().recommend_tools(query, top_k)