butler/tools/experience_selector.py
Rama main · réplica de lectura del repositorio original.
"""Experience-based tool selector — recommends tools from historical success.
Integrates with the Experience Tree to:
1. Retrieve relevant experiences for a query
2. Extract tool recommendations from successful experiences
3. Score tools by success rate and relevance
"""
from __future__ import annotations
import logging
from dataclasses import dataclass
from typing import Any
logger = logging.getLogger(__name__)
try:
from butler.memory.experience.tree import get_experience_tree
from butler.memory.experience.store import ExperienceHit
_EXPERIENCE_AVAILABLE = True
except ImportError:
_EXPERIENCE_AVAILABLE = False
def get_experience_tree():
return None
@dataclass
class ToolRecommendation:
"""A tool recommendation with score and source."""
tool_name: str
score: float
source: str # "experience" | "taxonomy" | "semantic"
experience_id: str | None = None
success_rate: float = 0.0
metadata: dict[str, Any] | None = None
class ExperienceBasedToolSelector:
"""Select tools based on historical experience from the Experience Tree."""
def __init__(self):
self._experience_tree = get_experience_tree() if _EXPERIENCE_AVAILABLE else None
self._tool_taxonomy = self._load_tool_taxonomy()
def _load_tool_taxonomy(self) -> dict[str, list[str]]:
"""Load default tool taxonomy for fallback."""
return {
"agent_dev": ["delegate_task", "skill_view", "skills_list", "run_workflow"],
"database": ["execute_code", "terminal", "read_file"],
"llm_usage": ["call_llm", "embed_text", "web_search"],
"code_engineering": ["read_file", "write_file", "patch", "search_files", "execute_code"],
"dev_ops": ["terminal", "run_workflow", "delegate_task"],
"network_info": ["web_search", "web_fetch"],
"daily_life": ["reminder", "memo", "habits"],
"project_mgmt": ["project_todos", "session_todos"],
}
def select_tools(self, query: str, top_k: int = 5) -> list[ToolRecommendation]:
"""Select tools based on experience retrieval.
Args:
query: The user query or task description
top_k: Maximum number of tools to return
Returns:
List of ToolRecommendation sorted by score
"""
recommendations: dict[str, ToolRecommendation] = {}
# 1. Try experience-based selection
if self._experience_tree:
exp_recommendations = self._select_from_experience(query, top_k)
for rec in exp_recommendations:
recommendations[rec.tool_name] = rec
# 2. Fallback to taxonomy-based selection
if len(recommendations) < top_k:
taxonomy_recommendations = self._select_from_taxonomy(query, top_k - len(recommendations))
for rec in taxonomy_recommendations:
if rec.tool_name not in recommendations:
recommendations[rec.tool_name] = rec
# 3. Sort and return
sorted_recs = sorted(recommendations.values(), key=lambda r: r.score, reverse=True)
return sorted_recs[:top_k]
def _select_from_experience(self, query: str, top_k: int) -> list[ToolRecommendation]:
"""Select tools from experience hits."""
if not self._experience_tree:
return []
try:
hits = self._experience_tree.retrieve(query, top_k=top_k * 3)
recommendations: dict[str, ToolRecommendation] = {}
for hit in hits:
tool_name = self._extract_tool_name(hit)
if not tool_name:
continue
# Calculate score based on experience success
base_score = hit.score
success_rate = getattr(hit.node, 'success_rate', 0.0)
# Boost score by success rate
adjusted_score = base_score * (1.0 + success_rate * 0.5)
# Update or create recommendation
if tool_name in recommendations:
# Merge: boost score if already recommended
existing = recommendations[tool_name]
existing.score = max(existing.score, adjusted_score)
else:
recommendations[tool_name] = ToolRecommendation(
tool_name=tool_name,
score=adjusted_score,
source="experience",
experience_id=hit.node.node_id,
success_rate=success_rate,
metadata={
"domain": hit.node.domain,
"category": hit.node.category,
},
)
return sorted(recommendations.values(), key=lambda r: r.score, reverse=True)[:top_k]
except Exception as e:
logger.debug("Experience-based tool selection failed: %s", e)
return []
def _extract_tool_name(self, hit) -> str | None:
"""Extract tool name from an experience hit."""
# Check metadata for explicit tool_name
if hasattr(hit.node, 'metadata') and hit.node.metadata:
tool_name = hit.node.metadata.get("tool_name")
if tool_name:
return tool_name
# Check category
if hit.node.category == "tools":
# Extract from name
name = hit.node.name
# Common patterns: "tool_name: description" or just "tool_name"
if ":" in name:
return name.split(":")[0].strip()
return name.lower().replace(" ", "_")
# Check content for tool mentions
content = hit.node.content or ""
import re
tool_patterns = [
r"tool[:\s]+`?([a-z_]+)`?",
r"调用[:\s]+`?([a-z_]+)`?",
r"used[:\s]+`?([a-z_]+)`?",
]
for pattern in tool_patterns:
match = re.search(pattern, content, re.IGNORECASE)
if match:
return match.group(1).lower()
return None
def _select_from_taxonomy(self, query: str, top_k: int) -> list[ToolRecommendation]:
"""Select tools from taxonomy based on query domain."""
q = query.lower()
# Simple keyword-based domain matching
domain_scores: dict[str, float] = {}
for domain, tools in self._tool_taxonomy.items():
if domain in q:
domain_scores[domain] = 0.3
# Check for specific tool mentions
all_tools = set()
for tools in self._tool_taxonomy.values():
all_tools.update(tools)
matched_tools: list[ToolRecommendation] = []
for tool in all_tools:
if tool.replace("_", " ") in q or tool in q:
matched_tools.append(ToolRecommendation(
tool_name=tool,
score=0.2,
source="taxonomy",
))
# Add domain-based tools
for domain, score in sorted(domain_scores.items(), key=lambda x: x[1], reverse=True):
for tool in self._tool_taxonomy.get(domain, []):
if not any(r.tool_name == tool for r in matched_tools):
matched_tools.append(ToolRecommendation(
tool_name=tool,
score=score,
source="taxonomy",
))
return sorted(matched_tools, key=lambda r: r.score, reverse=True)[:top_k]
def get_tools_for_domain(self, domain_id: str) -> list[str]:
"""Get recommended tools for a specific domain."""
return self._tool_taxonomy.get(domain_id, [])
def record_tool_usage(self, tool_name: str, query: str, success: bool) -> None:
"""Record a tool usage to the experience tree for future learning."""
if not self._experience_tree:
return
try:
tree = self._experience_tree
# Determine domain from query
from butler.memory.experience.domain_router import DomainRouter
router = DomainRouter()
domain_id, _ = router.route(query)
# Write experience
tree.write(
query=query,
result=f"Tool {tool_name} {'succeeded' if success else 'failed'}",
metadata={
"tool_name": tool_name,
"success": success,
"type": "tool_usage",
},
)
logger.debug("Recorded tool usage: %s (success=%s)", tool_name, success)
except Exception as e:
logger.debug("Failed to record tool usage: %s", e)
# Singleton
_selector: ExperienceBasedToolSelector | None = None
def get_tool_selector() -> ExperienceBasedToolSelector:
"""Get the singleton tool selector instance."""
global _selector
if _selector is None:
_selector = ExperienceBasedToolSelector()
return _selector
def select_tools(query: str, top_k: int = 5) -> list[ToolRecommendation]:
"""Convenience function to select tools."""
return get_tool_selector().select_tools(query, top_k)