""" 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