reasoning/base.py

"""Base classes and types for reasoning strategies."""

from abc import ABC, abstractmethod
from typing import Dict, Any, List, Optional, AsyncGenerator, Generator
from dataclasses import dataclass, field
from datetime import datetime

@dataclass
class StrategyResult:
    """Result from a reasoning strategy."""
    
    strategy_type: str
    success: bool
    answer: Optional[str]
    confidence: float
    reasoning_trace: List[Dict[str, Any]]
    metadata: Dict[str, Any]
    performance_metrics: Dict[str, Any]
    timestamp: str = field(default_factory=lambda: datetime.now().isoformat())

class ReasoningStrategy(ABC):
    """Base class for all reasoning strategies."""
    
    @abstractmethod
    async def reason(
        self,
        query: str,
        context: Dict[str, Any],
        **kwargs
    ) -> StrategyResult:
        """
        Reason about a query using this strategy.
        
        Args:
            query: The query to reason about
            context: Additional context for reasoning
            **kwargs: Additional strategy-specific parameters
            
        Returns:
            StrategyResult containing the reasoning output and metadata
        """
        pass
    
    def _prepare_messages(
        self,
        query: str,
        context: Dict[str, Any]
    ) -> List[Dict[str, str]]:
        """
        Prepare messages for the model.
        
        Args:
            query: The query to reason about
            context: Additional context
            
        Returns:
            List of message dictionaries
        """
        messages = []
        
        # Add system message if provided
        if "system_message" in context:
            messages.append({
                "role": "system",
                "content": context["system_message"]
            })
        
        # Add chat history if provided
        if "chat_history" in context:
            messages.extend(context["chat_history"])
        
        # Add the current query
        messages.append({
            "role": "user",
            "content": query
        })
        
        return messages
    
    def _calculate_confidence(self, response: Any) -> float:
        """
        Calculate confidence score for a response.
        
        Args:
            response: Response from the model
            
        Returns:
            Confidence score between 0 and 1
        """
        return 0.8  # Default confidence