Update definitions.py

definitions.py

@@ -1,4 +1,3 @@
-from __future__ import annotations
 import platform
 import psutil
 from typing import List, Dict, Optional, Any, Tuple
@@ -27,12 +26,14 @@ from typing import Dict, Any
 from datetime import datetime
 from pathlib import Path
 
+# Set logging level from environment variable
+logging.basicConfig(level=os.getenv('LOG_LEVEL', 'INFO'))
 
 class AutonomousAgentApp:
     """Main application class for the Autonomous Agent System"""
 
     def __init__(self):
-        self.workspace_manager = self.WorkspaceManager(workspace_dir='workspace')  # Use self.WorkspaceManager
+        self.workspace_manager = self.WorkspaceManager(workspace_dir=os.getenv('WORKSPACE_DIR', 'workspace'))  # Use self.WorkspaceManager
         self.pipeline = self._initialize_pipeline()
         self.refinement_loop = self.RefinementLoop(pipeline=self.pipeline)  # Use self.RefinementLoop
         self.interface = self.StreamlitInterface(self)  # Use self.StreamlitInterface
@@ -137,8 +138,11 @@ class AutonomousAgentApp:
         async def default_task(self):
             """Perform the default task of analyzing and generating tools/agents."""
             logging.info("Running default task...")
-            # Simulate task processing
-            await asyncio.sleep(2)  # Simulate time taken for the task
+            try:
+                # Simulate task processing
+                await asyncio.sleep(2)  # Simulate time taken for the task
+            except Exception as e:
+                logging.error(f"Error during default task: {str(e)}")
 
         async def pause(self):
             """Pause the current operation to accept user input."""
@@ -151,37 +155,34 @@ class AutonomousAgentApp:
             commands = self.extract_commands(user_input)
 
             for command in commands:
-                if command.startswith("generate tool"):
-                    await self.generate_tool(command)
-                elif command.startswith("generate agent"):
-                    await self.generate_agent(command)
-                # Add more command handling as needed
+                try:
+                    if command.startswith("generate tool"):
+                        await self.generate_tool(command)
+                    elif command.startswith("generate agent"):
+                        await self.generate_agent(command)
+                except Exception as e:
+                    logging.error(f"Error processing command '{command}': {str(e)}")
 
         def extract_commands(self, user_input: str) -> List[str]:
             """Extract commands from user input."""
-            # Simple command extraction logic (can be improved with NLP)
             return user_input.split(';')  # Assume commands are separated by semicolons
 
         async def run_refinement_cycle(self, task: str) -> Dict[str, Any]:
             """Run a refinement cycle for the given task."""
-            # Step 1: Analyze the task
-            task_analysis = await self._analyze_task(task)
-
-            # Step 2: Search for relevant approaches/methods
-            search_results = await self._web_search(task)
-
-            # Step 3: Build tools/agents based on the task
-            tools_built = await self._build_tools(task_analysis, search_results)
-
-            # Step 4: Execute the tools/agents
-            execution_results = await self._execute_tools(tools_built)
+            try:
+                task_analysis = await self._analyze_task(task)
+                search_results = await self._web_search(task)
+                tools_built = await self._build_tools(task_analysis, search_results)
+                execution_results = await self._execute_tools(tools_built)
 
-            return {
-                "task_analysis": task_analysis,
-                "search_results": search_results,
-                "tools_built": tools_built,
-                "execution_results": execution_results,
-            }
+                return {
+                    "task_analysis": task_analysis,
+                    "search_results": search_results,
+                    "tools_built": tools_built,
+                    "execution_results": execution_results,
+                }
+            except Exception as e:
+                logging.error(f"Error during refinement cycle: {str(e)}")
 
         async def _analyze_task(self, task: str) -> Dict[str, Any]:
             """Analyze the task to determine requirements."""
@@ -220,7 +221,7 @@ class AutonomousAgentApp:
             """Perform a web search for relevant approaches/methods."""
             try:
                 response = requests.get(
-                    "https://api.example.com/search",
+                    os.getenv('API_URL', 'https://api.example.com/search'),
                     params={"q": query, "limit": 5}
                 )
                 response.raise_for_status()
@@ -549,8 +550,6 @@ if __name__ == "__main__":
                 return {
                     "test_cases": test_generation["result"],
                     "test_results": test_results["result"],
-                    "coverage_report": coverage_report["result"]
-                }
             except Exception as e:
                 raise Exception(f"Testing stage failed: {str(e)}")
 
@@ -847,10 +846,10 @@ if __name__ == "__main__":
         test_coverage: float = 0.0
         security_score: str = "unknown"
         performance_score: float = 0.0
-        metrics_analyzer: CodeMetricsAnalyzer = None  # Forward reference
-    
+        metrics_analyzer: CodeMetricsAnalyzer = None
+
         def __post_init__(self):
-            self.metrics_analyzer = CodeMetricsAnalyzer()  # Instantiate here
+            self.metrics_analyzer = CodeMetricsAnalyzer()
             self.history = []
             self.thresholds = {
                 "code_quality": 0.85,
@@ -1260,4 +1259,4 @@ if __name__ == "__main__":
 
 if __name__ == "__main__":
     import streamlit as st
-    main()
+    main()