"""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