decision_maker.py

from typing import Dict, List, Any, Optional
from dataclasses import dataclass, asdict
from enum import Enum
import json
import time

class ToolType(Enum):
    SEARCH = "search"
    CODE_ANALYSIS = "code_analysis"
    FILE_OPERATION = "file_operation"
    UNKNOWN = "unknown"

@dataclass
class Tool:
    name: str
    type: ToolType
    description: str
    required_params: List[str]
    optional_params: List[str]
    confidence_threshold: float = 0.7

    def to_dict(self) -> Dict[str, Any]:
        """Convert Tool object to a dictionary for JSON serialization."""
        return {
            "name": self.name,
            "type": self.type.value,
            "description": self.description,
            "required_params": self.required_params,
            "optional_params": self.optional_params,
            "confidence_threshold": self.confidence_threshold
        }

class DecisionMaker:
    def __init__(self):
        self.tools = self._initialize_tools()
        self.decision_history = []
        
    def _initialize_tools(self) -> Dict[str, Tool]:
        """Initialize available tools with their metadata."""
        return {
            "simple_search": Tool(
                name="simple_search",
                type=ToolType.SEARCH,
                description="Perform web search using DuckDuckGo",
                required_params=["query"],
                optional_params=["max_results"],
                confidence_threshold=0.6
            ),
            "code_analysis": Tool(
                name="code_analysis",
                type=ToolType.CODE_ANALYSIS,
                description="Analyze Python code structure and provide insights",
                required_params=["code"],
                optional_params=[],
                confidence_threshold=0.8
            ),
            "file_operation": Tool(
                name="file_operation",
                type=ToolType.FILE_OPERATION,
                description="Perform file operations like read/write",
                required_params=["path"],
                optional_params=["mode"],
                confidence_threshold=0.9
            )
        }
    
    def analyze_request(self, request: str) -> Dict[str, Any]:
        """
        Analyze the user request to determine the best course of action.
        """
        analysis = {
            "intent": self._detect_intent(request),
            "required_tools": [],
            "confidence": 0.0,
            "suggested_actions": []
        }
        
        # Determine required tools based on intent
        if "search" in analysis["intent"]:
            analysis["required_tools"].append(self.tools["simple_search"].to_dict())
        if "code" in analysis["intent"]:
            analysis["required_tools"].append(self.tools["code_analysis"].to_dict())
        if "file" in analysis["intent"]:
            analysis["required_tools"].append(self.tools["file_operation"].to_dict())
            
        # Calculate confidence based on tool requirements
        if analysis["required_tools"]:
            analysis["confidence"] = min(tool["confidence_threshold"] for tool in analysis["required_tools"])
            
        # Generate suggested actions
        analysis["suggested_actions"] = self._generate_actions(analysis)
        
        return analysis
    
    def _detect_intent(self, request: str) -> List[str]:
        """Detect the intent(s) from the user request."""
        intents = []
        
        # Python-specific keyword-based intent detection
        keywords = {
            "search": ["search", "find", "look up", "query"],
            "code": ["python", "code", "function", "class", "analyze", "def", "import", "from"],
            "file": ["file", "read", "write", "save", "load", ".py"]
        }
        
        request_lower = request.lower()
        for intent, words in keywords.items():
            if any(word in request_lower for word in words):
                intents.append(intent)
                
        return intents if intents else ["unknown"]
    
    def _generate_actions(self, analysis: Dict[str, Any]) -> List[Dict[str, Any]]:
        """Generate suggested actions based on the analysis."""
        actions = []
        
        for tool in analysis["required_tools"]:
            action = {
                "tool": tool["name"],
                "type": tool["type"],
                "confidence": tool["confidence_threshold"],
                "required_params": tool["required_params"],
                "optional_params": tool["optional_params"]
            }
            actions.append(action)
            
        return actions
    
    def validate_tool_usage(self, tool_name: str, params: Dict[str, Any]) -> Dict[str, Any]:
        """
        Validate if a tool can be used with the given parameters.
        """
        if tool_name not in self.tools:
            return {
                "valid": False,
                "error": f"Unknown tool: {tool_name}"
            }
            
        tool = self.tools[tool_name]
        validation = {
            "valid": True,
            "missing_params": [],
            "extra_params": []
        }
        
        # Check required parameters
        for param in tool.required_params:
            if param not in params:
                validation["valid"] = False
                validation["missing_params"].append(param)
                
        # Check for extra parameters
        for param in params:
            if param not in tool.required_params and param not in tool.optional_params:
                validation["extra_params"].append(param)
                
        return validation
    
    def log_decision(self, request: str, analysis: Dict[str, Any], outcome: Dict[str, Any]):
        """
        Log a decision made by the system for future reference.
        """
        decision = {
            "timestamp": time.time(),
            "request": request,
            "analysis": analysis,
            "outcome": outcome
        }
        self.decision_history.append(decision)
        
    def get_decision_history(self) -> List[Dict[str, Any]]:
        """Get the history of decisions made."""
        return self.decision_history