autonomous_agent.py

"""
Autonomous Agent for problem solving and self-improvement.

This module implements an autonomous agent that can understand problems,
decompose them into sub-tasks, generate and test solutions, and learn from results.
"""
import json
import logging
from typing import Dict, List, Optional, Any, Tuple
from datetime import datetime
from pathlib import Path
import hashlib

# Configure logging
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)

class AutonomousAgent:
    """
    An autonomous agent that can solve problems through decomposition, 
    solution generation, and learning.
    """
    
    def __init__(self, expert_system, code_agent, knowledge_base_path: str = "knowledge_base"):
        """
        Initialize the autonomous agent.
        
        Args:
            expert_system: Instance of ExpertSystem for problem solving
            code_agent: Instance of CodeAgent for code execution
            knowledge_base_path: Path to store learned knowledge
        """
        self.expert_system = expert_system
        self.code_agent = code_agent
        self.knowledge_base_path = Path(knowledge_base_path)
        self.knowledge_base_path.mkdir(exist_ok=True)
        self.memory = self._load_knowledge()
    
    def _load_knowledge(self) -> Dict[str, Any]:
        """Load existing knowledge from the knowledge base."""
        knowledge = {
            'solutions': {},
            'patterns': {},
            'lessons_learned': []
        }
        
        # Load solutions
        solutions_file = self.knowledge_base_path / 'solutions.json'
        if solutions_file.exists():
            with open(solutions_file, 'r', encoding='utf-8') as f:
                knowledge['solutions'] = json.load(f)
        
        # Load patterns
        patterns_file = self.knowledge_base_path / 'patterns.json'
        if patterns_file.exists():
            with open(patterns_file, 'r', encoding='utf-8') as f:
                knowledge['patterns'] = json.load(f)
        
        # Load lessons learned
        lessons_file = self.knowledge_base_path / 'lessons.json'
        if lessons_file.exists():
            with open(lessons_file, 'r', encoding='utf-8') as f:
                knowledge['lessons_learned'] = json.load(f)
        
        return knowledge
    
    def _save_knowledge(self) -> None:
        """Save current knowledge to the knowledge base."""
        # Save solutions
        with open(self.knowledge_base_path / 'solutions.json', 'w', encoding='utf-8') as f:
            json.dump(self.memory['solutions'], f, indent=2)
        
        # Save patterns
        with open(self.knowledge_base_path / 'patterns.json', 'w', encoding='utf-8') as f:
            json.dump(self.memory['patterns'], f, indent=2)
        
        # Save lessons learned
        with open(self.knowledge_base_path / 'lessons.json', 'w', encoding='utf-8') as f:
            json.dump(self.memory['lessons_learned'], f, indent=2)
    
    def _generate_solution_id(self, problem: str) -> str:
        """Generate a unique ID for a problem/solution."""
        return hashlib.md5(problem.encode('utf-8')).hexdigest()
    
    def solve_problem(self, problem: str) -> Dict[str, Any]:
        """
        Solve a problem autonomously.
        
        Args:
            problem: The problem to solve
            
        Returns:
            Dict containing the solution and metadata
        """
        logger.info(f"Solving problem: {problem[:100]}...")
        
        # Check if we've solved this problem before
        problem_id = self._generate_solution_id(problem)
        if problem_id in self.memory['solutions']:
            logger.info("Found existing solution in knowledge base")
            return self.memory['solutions'][problem_id]
        
        # Decompose the problem
        sub_tasks = self._decompose_problem(problem)
        
        # Solve each sub-task
        solutions = []
        for i, task in enumerate(sub_tasks, 1):
            logger.info(f"Solving sub-task {i}/{len(sub_tasks)}: {task}")
            solution = self._solve_sub_task(task)
            solutions.append({
                'task': task,
                'solution': solution
            })
        
        # Combine solutions
        final_solution = self._combine_solutions(problem, solutions)
        
        # Store the solution
        self.memory['solutions'][problem_id] = {
            'problem': problem,
            'solution': final_solution,
            'timestamp': datetime.now().isoformat(),
            'sub_tasks': solutions
        }
        
        # Save updated knowledge
        self._save_knowledge()
        
        return self.memory['solutions'][problem_id]
    
    def _decompose_problem(self, problem: str) -> List[str]:
        """Break down a problem into sub-tasks."""
        try:
            # Use the expert system to break down the problem
            prompt = f"""
            Break down the following problem into clear, independent sub-tasks.
            Return the sub-tasks as a JSON array of strings.
            
            Problem: {problem}
            
            Sub-tasks:
            """
            
            # Use the expert system to generate the breakdown
            response = self.expert_system.solve_problem(
                problem=prompt,
                subject='general',
                temperature=0.3,
                max_tokens=1000
            )
            
            # Extract the content from the response
            content = response.get('solution', '').strip()
            
            # Clean up the response to extract JSON
            if content.startswith('```'):
                content = '\n'.join(content.split('\n')[1:-1])
            
            # Parse the JSON array
            sub_tasks = json.loads(content)
            return [str(task) for task in sub_tasks]
            
        except Exception as e:
            logger.error(f"Error decomposing problem: {e}")
            # Fallback to treating the entire problem as a single task
            return [problem]
    
    def _solve_sub_task(self, task: str) -> Dict[str, Any]:
        """Solve a single sub-task."""
        # First try to find a matching pattern
        for pattern, solution in self.memory['patterns'].items():
            if self._pattern_matches(pattern, task):
                logger.info(f"Matched pattern: {pattern}")
                return {
                    'type': 'pattern_match',
                    'pattern': pattern,
                    'solution': solution,
                    'source': 'memory'
                }
        
        # If no pattern matches, generate a new solution
        return self._generate_solution(task)
    
    def _pattern_matches(self, pattern: str, task: str) -> bool:
        """Check if a task matches a known pattern."""
        # Simple implementation - could be enhanced with more sophisticated matching
        pattern_terms = set(pattern.lower().split())
        task_terms = set(task.lower().split())
        return len(pattern_terms.intersection(task_terms)) / len(pattern_terms) > 0.6
    
    def _generate_solution(self, task: str) -> Dict[str, Any]:
        """Generate a solution for a task."""
        try:
            # Determine if this is a coding task
            is_code_task = any(term in task.lower() for term in ['code', 'function', 'script', 'program'])
            
            # Use the expert system to generate a solution
            prompt = f"""
            Provide a solution to the following task.
            """
            
            if is_code_task:
                prompt += """
                This appears to be a coding task. Please provide a complete, working Python function or script.
                Include all necessary imports and example usage if applicable.
                """
            
            prompt += f"""
            
            Task: {task}
            
            Solution:
            """
            
            # Use the expert system to generate the solution
            response = self.expert_system.solve_problem(
                problem=prompt,
                subject='coding' if is_code_task else 'general',
                temperature=0.2,
                max_tokens=2000
            )
            
            solution = response.get('solution', '').strip()
            
            # If this is a coding task, try to execute it
            if is_code_task:
                try:
                    result, output = self.code_agent.execute_code(solution)
                    return {
                        'type': 'code',
                        'solution': solution,
                        'result': str(result) if result is not None else None,
                        'output': output,
                        'success': result is not None,
                        'source': 'generated'
                    }
                except Exception as e:
                    logger.error(f"Error executing generated code: {e}")
                    return {
                        'type': 'code_error',
                        'solution': solution,
                        'error': str(e),
                        'source': 'generated'
                    }       
        except Exception as e:
            logger.error(f"Error generating solution: {e}")
            return {
                'type': 'error',
                'error': str(e),
                'source': 'generated'
            }
    
    def _combine_solutions(self, problem: str, solutions: List[Dict[str, Any]]) -> str:
        """Combine solutions to sub-tasks into a final solution."""
        # Simple implementation - could be enhanced with more sophisticated combination
        combined = f"# Solution to: {problem}\n\n"
        
        for i, sol in enumerate(solutions, 1):
            combined += f"## Sub-task {i}: {sol['task']}\n\n"
            if sol['solution'].get('type') == 'code_solution':
                combined += f"```python\n{sol['solution']['solution']}\n```\n\n"
                if 'execution_result' in sol['solution']:
                    exec_result = sol['solution']['execution_result']
                    if exec_result['output']:
                        combined += f"### Output\n```\n{exec_result['output']}\n```\n\n"
                    if exec_result['result']:
                        combined += f"### Result\n```\n{exec_result['result']}\n```\n\n"
            else:
                combined += f"{sol['solution'].get('solution', 'No solution generated')}\n\n"
        
        return combined
    
    def learn_from_experience(self) -> None:
        """Analyze past solutions to learn patterns and improve future performance."""
        logger.info("Analyzing past solutions to learn patterns...")
        
        # This is a simplified implementation - in practice, you'd want more sophisticated analysis
        for problem_id, solution_data in self.memory['solutions'].items():
            problem = solution_data['problem']
            solution = solution_data['solution']
            
            # Extract key terms from the problem
            terms = set(problem.lower().split())
            
            # Store the pattern for future matching
            for term in terms:
                if len(term) > 3:  # Only consider terms longer than 3 characters
                    if term not in self.memory['patterns']:
                        self.memory['patterns'][term] = []
                    self.memory['patterns'][term].append({
                        'problem': problem,
                        'solution': solution,
                        'timestamp': solution_data.get('timestamp', datetime.now().isoformat())
                    })
        
        # Save updated knowledge
        self._save_knowledge()
        logger.info(f"Learned {len(self.memory['patterns'])} patterns from past solutions")