Upload folder using huggingface_hub

reasoning/specialized.py

@@ -11,6 +11,119 @@ from collections import defaultdict
 
 from .base import ReasoningStrategy
 
+class SpecializedReasoning(ReasoningStrategy):
+    """
+    A composite reasoning strategy that combines multiple specialized strategies
+    for different domains and tasks.
+    """
+    
+    def __init__(self, config: Optional[Dict[str, Any]] = None):
+        """Initialize specialized reasoning with component strategies."""
+        super().__init__()
+        self.config = config or {}
+        
+        # Initialize component strategies
+        self.strategies = {
+            'code_rewrite': CodeRewriteStrategy(),
+            'security_audit': SecurityAuditStrategy(),
+            'performance': PerformanceOptimizationStrategy(),
+            'testing': TestGenerationStrategy(),
+            'documentation': DocumentationStrategy(),
+            'api_design': APIDesignStrategy(),
+            'dependencies': DependencyManagementStrategy(),
+            'code_review': CodeReviewStrategy()
+        }
+    
+    async def reason(self, query: str, context: Dict[str, Any]) -> Dict[str, Any]:
+        """
+        Apply specialized reasoning by selecting and combining appropriate
+        strategies based on the query and context.
+        
+        Args:
+            query: The input query to reason about
+            context: Additional context and parameters
+            
+        Returns:
+            Dict containing reasoning results and confidence scores
+        """
+        try:
+            # Determine which strategies to use based on context
+            selected_strategies = await self._select_strategies(query, context)
+            
+            # Get results from each selected strategy
+            results = {}
+            for strategy_name in selected_strategies:
+                strategy = self.strategies[strategy_name]
+                results[strategy_name] = await strategy.reason(query, context)
+            
+            # Combine results
+            combined_result = await self._combine_results(results, context)
+            
+            return {
+                'answer': combined_result.get('answer', ''),
+                'confidence': combined_result.get('confidence', 0.0),
+                'reasoning_path': {
+                    'selected_strategies': selected_strategies,
+                    'individual_results': results,
+                    'combination_method': combined_result.get('method', '')
+                }
+            }
+            
+        except Exception as e:
+            logging.error(f"Specialized reasoning failed: {str(e)}")
+            return {
+                'error': f"Specialized reasoning failed: {str(e)}",
+                'confidence': 0.0
+            }
+    
+    async def _select_strategies(self, query: str, context: Dict[str, Any]) -> List[str]:
+        """Select appropriate strategies based on query and context."""
+        selected = []
+        
+        # Simple keyword-based selection for now
+        keywords = {
+            'code_rewrite': ['rewrite', 'refactor', 'improve'],
+            'security_audit': ['security', 'vulnerability', 'audit'],
+            'performance': ['performance', 'optimize', 'speed'],
+            'testing': ['test', 'coverage', 'verify'],
+            'documentation': ['document', 'explain', 'describe'],
+            'api_design': ['api', 'interface', 'endpoint'],
+            'dependencies': ['dependency', 'package', 'version'],
+            'code_review': ['review', 'quality', 'check']
+        }
+        
+        query_lower = query.lower()
+        for strategy, terms in keywords.items():
+            if any(term in query_lower for term in terms):
+                selected.append(strategy)
+        
+        # If no specific strategies selected, use code review as default
+        if not selected:
+            selected = ['code_review']
+        
+        return selected
+    
+    async def _combine_results(
+        self, 
+        results: Dict[str, Dict[str, Any]], 
+        context: Dict[str, Any]
+    ) -> Dict[str, Any]:
+        """Combine results from multiple strategies."""
+        if not results:
+            return {'answer': '', 'confidence': 0.0, 'method': 'none'}
+        
+        # For now, use the highest confidence result
+        best_result = max(
+            results.items(),
+            key=lambda x: x[1].get('confidence', 0)
+        )
+        
+        return {
+            'answer': best_result[1].get('answer', ''),
+            'confidence': best_result[1].get('confidence', 0.0),
+            'method': 'highest_confidence'
+        }
+
 class CodeRewriteStrategy(ReasoningStrategy):
     """
     Advanced code rewriting strategy that: