125 lines
4.3 KiB
Python
125 lines
4.3 KiB
Python
"""
|
|
Tool Decision Engine - decides which tools to invoke autonomously.
|
|
"""
|
|
|
|
import logging
|
|
from typing import Dict, List, Any
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class ToolDecisionEngine:
|
|
"""Decides which tools to invoke based on context analysis."""
|
|
|
|
async def analyze_tool_needs(
|
|
self,
|
|
user_prompt: str,
|
|
monologue: Dict[str, Any],
|
|
context_state: Dict[str, Any],
|
|
available_tools: List[str]
|
|
) -> Dict[str, Any]:
|
|
"""
|
|
Analyze if tools should be invoked and which ones.
|
|
|
|
Args:
|
|
user_prompt: User's message
|
|
monologue: Inner monologue analysis
|
|
context_state: Full context
|
|
available_tools: List of available tools
|
|
|
|
Returns:
|
|
{
|
|
"should_invoke_tools": bool,
|
|
"tools_to_invoke": [
|
|
{
|
|
"tool": "RAG | WEB | WEATHER | etc",
|
|
"query": "search query",
|
|
"reason": "why this tool",
|
|
"priority": 0.0-1.0
|
|
},
|
|
...
|
|
],
|
|
"confidence": 0.0-1.0
|
|
}
|
|
"""
|
|
|
|
tools_to_invoke = []
|
|
|
|
# Check for memory/context needs
|
|
if any(word in user_prompt.lower() for word in [
|
|
"remember", "you said", "we discussed", "earlier", "before",
|
|
"last time", "previously", "what did"
|
|
]):
|
|
tools_to_invoke.append({
|
|
"tool": "RAG",
|
|
"query": user_prompt,
|
|
"reason": "User references past conversation",
|
|
"priority": 0.9
|
|
})
|
|
|
|
# Check for web search needs
|
|
if any(word in user_prompt.lower() for word in [
|
|
"current", "latest", "news", "today", "what's happening",
|
|
"look up", "search for", "find information", "recent"
|
|
]):
|
|
tools_to_invoke.append({
|
|
"tool": "WEB",
|
|
"query": user_prompt,
|
|
"reason": "Requires current information",
|
|
"priority": 0.8
|
|
})
|
|
|
|
# Check for weather needs
|
|
if any(word in user_prompt.lower() for word in [
|
|
"weather", "temperature", "forecast", "rain", "sunny", "climate"
|
|
]):
|
|
tools_to_invoke.append({
|
|
"tool": "WEATHER",
|
|
"query": user_prompt,
|
|
"reason": "Weather information requested",
|
|
"priority": 0.95
|
|
})
|
|
|
|
# Check for code-related needs
|
|
if any(word in user_prompt.lower() for word in [
|
|
"code", "function", "debug", "implement", "algorithm",
|
|
"programming", "script", "syntax"
|
|
]):
|
|
if "CODEBRAIN" in available_tools:
|
|
tools_to_invoke.append({
|
|
"tool": "CODEBRAIN",
|
|
"query": user_prompt,
|
|
"reason": "Code-related task",
|
|
"priority": 0.85
|
|
})
|
|
|
|
# Proactive RAG for complex queries (based on monologue)
|
|
intent = monologue.get("intent", "") if monologue else ""
|
|
if monologue and monologue.get("consult_executive"):
|
|
# Complex query - might benefit from context
|
|
if not any(t["tool"] == "RAG" for t in tools_to_invoke):
|
|
tools_to_invoke.append({
|
|
"tool": "RAG",
|
|
"query": user_prompt,
|
|
"reason": "Complex query benefits from context",
|
|
"priority": 0.6
|
|
})
|
|
|
|
# Sort by priority
|
|
tools_to_invoke.sort(key=lambda x: x["priority"], reverse=True)
|
|
|
|
max_priority = max([t["priority"] for t in tools_to_invoke]) if tools_to_invoke else 0.0
|
|
|
|
result = {
|
|
"should_invoke_tools": len(tools_to_invoke) > 0,
|
|
"tools_to_invoke": tools_to_invoke,
|
|
"confidence": max_priority
|
|
}
|
|
|
|
if tools_to_invoke:
|
|
logger.info(f"[TOOL_DECISION] Autonomous tool invocation recommended: {len(tools_to_invoke)} tools")
|
|
for tool in tools_to_invoke:
|
|
logger.info(f" - {tool['tool']} (priority: {tool['priority']:.2f}): {tool['reason']}")
|
|
|
|
return result
|