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