Update agents/assembler_agent.py

agents/assembler_agent.py

@@ -1,204 +1,204 @@
-from typing import Dict, List
-from models.model_config import ModelConfig
-from models.data_models import AssemblerInput
-from .base_agent import BaseAgent
-import datetime 
-
-class AssemblerAgent(BaseAgent):
-    def __init__(self, name: str = "AssemblerAgent"):
-        super().__init__(name)
-        self.final_report: Dict = {}
-
-
-    def reason(self, input_data: AssemblerInput) -> List[str]:
-        """
-        Plan how to assemble the final report from all agent results
-        """
-        thoughts = []
-        
-        try:
-            # Analyze available inputs
-            thoughts.append("Analyzing inputs from all agents:")
-            thoughts.append(f"- User input processing results available: {bool(input_data.user_input_results)}")
-            thoughts.append(f"- Context learning results available: {bool(input_data.context_results)}")
-            thoughts.append(f"- Image analysis results available: {bool(input_data.image_results)}")
-            
-            # Plan report structure
-            thoughts.append("\nPlanning report structure:")
-            thoughts.append("1. User Query Summary")
-            thoughts.append("2. Context Analysis")
-            thoughts.append("3. Image Analysis Results")
-            thoughts.append("4. Model Performance Metrics")
-            thoughts.append("5. Final Recommendations")
-            
-            # Consider report format
-            thoughts.append(f"\nReport Format: {input_data.report_format}")
-            if input_data.report_format == "detailed":
-                thoughts.append("- Will include full model decisions")
-                thoughts.append("- Will include confidence scores")
-                thoughts.append("- Will include processing statistics")
-            else:
-                thoughts.append("- Will provide condensed summary")
-                thoughts.append("- Will focus on key findings")
-            
-            # Store thoughts in state
-            self.state.thoughts.extend(thoughts)
-            self.logger.info("Reasoning complete for report assembly")
-            
-            return thoughts
-            
-        except Exception as e:
-            error_msg = f"Error during assembly reasoning: {str(e)}"
-            self.state.errors.append(error_msg)
-            self.logger.error(error_msg)
-            return thoughts
-        
-    def execute(self, input_data: AssemblerInput) -> Dict:
-        """
-        Assemble final report from all agent results
-        """
-        try:
-            if not self.validate(input_data):
-                return {
-                    'status': 'error',
-                    'error': self.state.errors[-1]
-                }
-
-            report = {
-                'summary': {
-                    'user_query': {},
-                    'context_analysis': {},
-                    'image_analysis': {},
-                    'recommendations': []
-                },
-                'details': {
-                    'model_decisions': {},
-                    'processing_stats': {},
-                    'confidence_scores': {}
-                },
-                'metadata': {
-                    'report_format': input_data.report_format,
-                    'timestamp': datetime.now().isoformat()
-                },
-                'status': 'processing'
-            }
-
-            # Process user input results
-            if input_data.user_input_results:
-                report['summary']['user_query'] = {
-                    'original_query': input_data.user_input_results.get('query', ''),
-                    'constraints': input_data.user_input_results.get('constraints', []),
-                    'intent': input_data.user_input_results.get('intent', '')
-                }
-
-            # Process context results
-            if input_data.context_results:
-                report['summary']['context_analysis'] = {
-                    'key_findings': input_data.context_results.get('summaries', {}),
-                    'relevant_keywords': list(input_data.context_results.get('keywords', set())),
-                    'sources': list(input_data.context_results.get('gathered_context', {}).keys())
-                }
-
-            # Process image analysis results
-            if input_data.image_results:
-                report['summary']['image_analysis'] = {
-                    'selected_images': input_data.image_results.get('selected_images', []),
-                    'analysis_summary': {
-                        path: results['caption']['text'] 
-                        for path, results in input_data.image_results.get('analyzed_images', {}).items()
-                    }
-                }
-
-            # Add detailed information if requested
-            if input_data.report_format == "detailed":
-                report['details']['model_decisions'] = {
-                    'context_models': input_data.context_results.get('model_decisions', {}),
-                    'image_models': input_data.image_results.get('model_decisions', {})
-                }
-                report['details']['processing_stats'] = {
-                    'context_processing': input_data.context_results.get('model_decisions', {}).get('processing_stats', {}),
-                    'image_processing': input_data.image_results.get('model_decisions', {}).get('processing_stats', {})
-                }
-
-            # Generate recommendations
-            report['summary']['recommendations'] = self._generate_recommendations(
-                report['summary']['context_analysis'],
-                report['summary']['image_analysis']
-            )
-
-            report['status'] = 'success'
-            self.final_report = report
-            
-            # Log decision
-            self.log_decision({
-                'action': 'report_assembly',
-                'format': input_data.report_format,
-                'sections_completed': list(report['summary'].keys())
-            })
-
-            return report
-
-        except Exception as e:
-            error_msg = f"Error executing report assembly: {str(e)}"
-            self.state.errors.append(error_msg)
-            self.logger.error(error_msg)
-            return {'status': 'error', 'error': error_msg}
-
-
-    def _generate_recommendations(self, context_analysis: Dict, image_analysis: Dict) -> List[str]:
-        """
-        Generate recommendations based on context and image analysis
-        
-        Args:
-            context_analysis: Results from context learning
-            image_analysis: Results from image analysis
-            
-        Returns:
-            List[str]: List of recommendations
-        """
-        try:
-            recommendations = []
-            
-            # Check if we have sufficient data
-            if not context_analysis or not image_analysis:
-                return ["Insufficient data to generate recommendations"]
-            
-            # Analyze context findings
-            if context_analysis.get('key_findings'):
-                recommendations.append("Based on context analysis:")
-                for source, finding in context_analysis['key_findings'].items():
-                    if finding:  # Check if finding exists
-                        recommendations.append(f"- {finding}")
-            
-            # Analyze image findings
-            if image_analysis.get('selected_images'):
-                recommendations.append("\nBased on image analysis:")
-                recommendations.append(f"- Found {len(image_analysis['selected_images'])} relevant images")
-                
-                # Add specific image recommendations
-                if image_analysis.get('analysis_summary'):
-                    for img_path, caption in image_analysis['analysis_summary'].items():
-                        if caption:  # Check if caption exists
-                            recommendations.append(f"- {caption}")
-            
-            # Combine findings for final recommendations
-            recommendations.append("\nKey Recommendations:")
-            if context_analysis.get('relevant_keywords'):
-                keywords = context_analysis['relevant_keywords'][:5]  # Top 5 keywords
-                recommendations.append(f"- Focus areas identified: {', '.join(keywords)}")
-            
-            # Add source credibility note
-            if context_analysis.get('sources'):
-                recommendations.append(f"- Analysis based on {len(context_analysis['sources'])} credible sources")
-            
-            # Add confidence note
-            recommendations.append("- Regular monitoring and updates recommended")
-            
-            self.logger.info("Generated recommendations successfully")
-            return recommendations
-            
-        except Exception as e:
-            error_msg = f"Error generating recommendations: {str(e)}"
-            self.logger.error(error_msg)
-            return ["Error generating recommendations. Please check the detailed report."]
-
+from typing import Dict, List
+from models.model_config import ModelConfig
+from models.data_models import AssemblerInput
+from .base_agent import BaseAgent
+from datetime import datetime
+
+class AssemblerAgent(BaseAgent):
+    def __init__(self, name: str = "AssemblerAgent"):
+        super().__init__(name)
+        self.final_report: Dict = {}
+
+
+    def reason(self, input_data: AssemblerInput) -> List[str]:
+        """
+        Plan how to assemble the final report from all agent results
+        """
+        thoughts = []
+        
+        try:
+            # Analyze available inputs
+            thoughts.append("Analyzing inputs from all agents:")
+            thoughts.append(f"- User input processing results available: {bool(input_data.user_input_results)}")
+            thoughts.append(f"- Context learning results available: {bool(input_data.context_results)}")
+            thoughts.append(f"- Image analysis results available: {bool(input_data.image_results)}")
+            
+            # Plan report structure
+            thoughts.append("\nPlanning report structure:")
+            thoughts.append("1. User Query Summary")
+            thoughts.append("2. Context Analysis")
+            thoughts.append("3. Image Analysis Results")
+            thoughts.append("4. Model Performance Metrics")
+            thoughts.append("5. Final Recommendations")
+            
+            # Consider report format
+            thoughts.append(f"\nReport Format: {input_data.report_format}")
+            if input_data.report_format == "detailed":
+                thoughts.append("- Will include full model decisions")
+                thoughts.append("- Will include confidence scores")
+                thoughts.append("- Will include processing statistics")
+            else:
+                thoughts.append("- Will provide condensed summary")
+                thoughts.append("- Will focus on key findings")
+            
+            # Store thoughts in state
+            self.state.thoughts.extend(thoughts)
+            self.logger.info("Reasoning complete for report assembly")
+            
+            return thoughts
+            
+        except Exception as e:
+            error_msg = f"Error during assembly reasoning: {str(e)}"
+            self.state.errors.append(error_msg)
+            self.logger.error(error_msg)
+            return thoughts
+        
+    def execute(self, input_data: AssemblerInput) -> Dict:
+        """
+        Assemble final report from all agent results
+        """
+        try:
+            if not self.validate(input_data):
+                return {
+                    'status': 'error',
+                    'error': self.state.errors[-1]
+                }
+
+            report = {
+                'summary': {
+                    'user_query': {},
+                    'context_analysis': {},
+                    'image_analysis': {},
+                    'recommendations': []
+                },
+                'details': {
+                    'model_decisions': {},
+                    'processing_stats': {},
+                    'confidence_scores': {}
+                },
+                'metadata': {
+                    'report_format': input_data.report_format,
+                    'timestamp': datetime.now().isoformat()
+                },
+                'status': 'processing'
+            }
+
+            # Process user input results
+            if input_data.user_input_results:
+                report['summary']['user_query'] = {
+                    'original_query': input_data.user_input_results.get('query', ''),
+                    'constraints': input_data.user_input_results.get('constraints', []),
+                    'intent': input_data.user_input_results.get('intent', '')
+                }
+
+            # Process context results
+            if input_data.context_results:
+                report['summary']['context_analysis'] = {
+                    'key_findings': input_data.context_results.get('summaries', {}),
+                    'relevant_keywords': list(input_data.context_results.get('keywords', set())),
+                    'sources': list(input_data.context_results.get('gathered_context', {}).keys())
+                }
+
+            # Process image analysis results
+            if input_data.image_results:
+                report['summary']['image_analysis'] = {
+                    'selected_images': input_data.image_results.get('selected_images', []),
+                    'analysis_summary': {
+                        path: results['caption']['text'] 
+                        for path, results in input_data.image_results.get('analyzed_images', {}).items()
+                    }
+                }
+
+            # Add detailed information if requested
+            if input_data.report_format == "detailed":
+                report['details']['model_decisions'] = {
+                    'context_models': input_data.context_results.get('model_decisions', {}),
+                    'image_models': input_data.image_results.get('model_decisions', {})
+                }
+                report['details']['processing_stats'] = {
+                    'context_processing': input_data.context_results.get('model_decisions', {}).get('processing_stats', {}),
+                    'image_processing': input_data.image_results.get('model_decisions', {}).get('processing_stats', {})
+                }
+
+            # Generate recommendations
+            report['summary']['recommendations'] = self._generate_recommendations(
+                report['summary']['context_analysis'],
+                report['summary']['image_analysis']
+            )
+
+            report['status'] = 'success'
+            self.final_report = report
+            
+            # Log decision
+            self.log_decision({
+                'action': 'report_assembly',
+                'format': input_data.report_format,
+                'sections_completed': list(report['summary'].keys())
+            })
+
+            return report
+
+        except Exception as e:
+            error_msg = f"Error executing report assembly: {str(e)}"
+            self.state.errors.append(error_msg)
+            self.logger.error(error_msg)
+            return {'status': 'error', 'error': error_msg}
+
+
+    def _generate_recommendations(self, context_analysis: Dict, image_analysis: Dict) -> List[str]:
+        """
+        Generate recommendations based on context and image analysis
+        
+        Args:
+            context_analysis: Results from context learning
+            image_analysis: Results from image analysis
+            
+        Returns:
+            List[str]: List of recommendations
+        """
+        try:
+            recommendations = []
+            
+            # Check if we have sufficient data
+            if not context_analysis or not image_analysis:
+                return ["Insufficient data to generate recommendations"]
+            
+            # Analyze context findings
+            if context_analysis.get('key_findings'):
+                recommendations.append("Based on context analysis:")
+                for source, finding in context_analysis['key_findings'].items():
+                    if finding:  # Check if finding exists
+                        recommendations.append(f"- {finding}")
+            
+            # Analyze image findings
+            if image_analysis.get('selected_images'):
+                recommendations.append("\nBased on image analysis:")
+                recommendations.append(f"- Found {len(image_analysis['selected_images'])} relevant images")
+                
+                # Add specific image recommendations
+                if image_analysis.get('analysis_summary'):
+                    for img_path, caption in image_analysis['analysis_summary'].items():
+                        if caption:  # Check if caption exists
+                            recommendations.append(f"- {caption}")
+            
+            # Combine findings for final recommendations
+            recommendations.append("\nKey Recommendations:")
+            if context_analysis.get('relevant_keywords'):
+                keywords = context_analysis['relevant_keywords'][:5]  # Top 5 keywords
+                recommendations.append(f"- Focus areas identified: {', '.join(keywords)}")
+            
+            # Add source credibility note
+            if context_analysis.get('sources'):
+                recommendations.append(f"- Analysis based on {len(context_analysis['sources'])} credible sources")
+            
+            # Add confidence note
+            recommendations.append("- Regular monitoring and updates recommended")
+            
+            self.logger.info("Generated recommendations successfully")
+            return recommendations
+            
+        except Exception as e:
+            error_msg = f"Error generating recommendations: {str(e)}"
+            self.logger.error(error_msg)
+            return ["Error generating recommendations. Please check the detailed report."]
+