File size: 6,211 Bytes
5854ce9 8476091 5854ce9 8476091 5854ce9 8476091 5854ce9 8476091 5854ce9 8476091 5854ce9 8476091 5854ce9 8476091 5854ce9 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 |
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 |