Optimize code reviewer with memory management and improved parsing

src/code_reviewer.py

@@ -1,8 +1,9 @@
 from typing import Dict, List, Optional
 import logging
-from datetime import datetime
+from datetime import datetime, timedelta
 from .model_manager import ModelManager
 from .config import Config
+import gc
 
 logger = logging.getLogger(__name__)
 
@@ -19,32 +20,38 @@ class CodeReviewer:
     def __init__(self, model_manager: ModelManager):
         self.model_manager = model_manager
         self.review_history: List[CodeReview] = []
+        self._last_cleanup = datetime.now()
         
     def _create_review_prompt(self, code: str, language: str) -> str:
         """Create a structured prompt for code review."""
-        return f"""As a code reviewer, analyze the following {language} code and provide specific suggestions in exactly these sections:
-- Issues: (list critical problems)
-- Improvements: (list suggested enhancements)
-- Best Practices: (list recommendations)
-- Security: (list security concerns)
+        # More concise prompt to reduce token usage
+        return f"""Review this {language} code. List specific points in these sections:
+Issues:
+Improvements:
+Best Practices:
+Security:
 
-Code to review:
+Code:
 ```{language}
 {code}
-```
-
-Provide your review in exactly these sections: Issues, Improvements, Best Practices, Security.
-Each section should contain a list of specific points.
-"""
+```"""
 
     def review_code(self, code: str, language: str, review_id: str) -> CodeReview:
         """Perform code review using the LLM."""
         try:
             start_time = datetime.now()
             
+            # Clean up old reviews periodically
+            self._cleanup_old_reviews()
+            
             # Create review instance
             review = CodeReview(code, language, review_id)
             
+            # Truncate code if too long
+            max_code_length = Config.MAX_INPUT_LENGTH - 200  # Reserve tokens for prompt
+            if len(code) > max_code_length:
+                code = code[:max_code_length] + "\n# ... (code truncated for length)"
+            
             # Generate review prompt
             prompt = self._create_review_prompt(code, language)
             
@@ -71,6 +78,9 @@ Each section should contain a list of specific points.
             # Store review in history
             self._add_to_history(review)
             
+            # Force garbage collection
+            gc.collect()
+            
             return review
             
         except Exception as e:
@@ -81,34 +91,38 @@ Each section should contain a list of specific points.
         """Parse the LLM response into structured sections."""
         sections = []
         current_section = None
+        required_sections = ['Issues', 'Improvements', 'Best Practices', 'Security']
         
-        # Split response into lines and process each line
-        lines = response.split('\n')
-        for line in lines:
-            line = line.strip()
-            if not line:
-                continue
-            
-            # Check for section headers
-            if line.startswith('- ') and ':' in line:
-                section_type = line[2:].split(':', 1)[0].strip()
-                current_section = {
-                    'type': section_type,
-                    'items': []
-                }
-                sections.append(current_section)
-                # Add any content after the colon as first item
-                content = line.split(':', 1)[1].strip()
-                if content:
-                    current_section['items'].append(content)
-            # Add items to current section
-            elif current_section and line.strip('-* '):
-                item = line.strip('-* ')
-                if item:  # Only add non-empty items
-                    current_section['items'].append(item)
+        try:
+            # Split response into lines and process each line
+            lines = response.split('\n')
+            for line in lines:
+                line = line.strip()
+                if not line:
+                    continue
+                
+                # Check for section headers
+                for section in required_sections:
+                    if line.lower().startswith(section.lower()):
+                        current_section = {
+                            'type': section,
+                            'items': []
+                        }
+                        sections.append(current_section)
+                        break
+                
+                # Add items to current section if not a section header
+                if current_section and line.strip('-* ') and not any(
+                    line.lower().startswith(s.lower()) for s in required_sections
+                ):
+                    item = line.strip('-* ')
+                    if item and not any(item == existing for existing in current_section['items']):
+                        current_section['items'].append(item)
+        
+        except Exception as e:
+            logger.error(f"Error parsing response: {str(e)}")
         
         # Ensure all required sections exist
-        required_sections = ['Issues', 'Improvements', 'Best Practices', 'Security']
         result = []
         for section_type in required_sections:
             found_section = next((s for s in sections if s['type'] == section_type), None)
@@ -125,9 +139,17 @@ Each section should contain a list of specific points.
     def _add_to_history(self, review: CodeReview):
         """Add review to history and maintain size limit."""
         self.review_history.append(review)
-        if len(self.review_history) > Config.MAX_HISTORY_ITEMS:
+        while len(self.review_history) > Config.MAX_HISTORY_ITEMS:
             self.review_history.pop(0)
 
+    def _cleanup_old_reviews(self):
+        """Clean up reviews older than retention period."""
+        if (datetime.now() - self._last_cleanup) > timedelta(hours=1):
+            cutoff_date = datetime.now() - timedelta(days=Config.HISTORY_RETENTION_DAYS)
+            self.review_history = [r for r in self.review_history if r.timestamp > cutoff_date]
+            self._last_cleanup = datetime.now()
+            gc.collect()
+
     def get_review_metrics(self) -> Dict:
         """Calculate aggregate metrics from review history."""
         if not self.review_history:
@@ -153,4 +175,4 @@ Each section should contain a list of specific points.
         """Get review history with optional limit."""
         if limit:
             return self.review_history[-limit:]
-        return self.review_history
+        return self.review_history.copy()  # Return copy to prevent external modifications