File size: 2,457 Bytes
bb8cb0c
1d75522
bb8cb0c
3feffc7
bb8cb0c
 
1d75522
bb8cb0c
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1d75522
 
 
 
bb8cb0c
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1d75522
 
bb8cb0c
1d75522
bb8cb0c
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
"""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