Coverage for /home/admin/Documents/AI/applications/lexigram-dev/lexigram/experimental/ai/lexigram-ai-rag/src/lexigram/ai/rag/query/base.py: 100%

1from __future__ import annotations 

2 

3from abc import ABC, abstractmethod 

4from dataclasses import dataclass, field 

5from enum import StrEnum 

6from typing import Any 

7 

8from lexigram.domain import DomainModel 

9from lexigram.validation import Field 

10 

11 

12class TransformationStrategy(StrEnum): 

13 """Available query transformation strategies.""" 

14 

15 EXPANSION = "expansion" 

16 MULTI_QUERY = "multi_query" 

17 HYDE = "hyde" 

18 REWRITE = "rewrite" 

19 CUSTOM = "custom" 

20 

21 

22@dataclass 

23class TransformedQuery: 

24 """Result of query transformation.""" 

25 

26 original: str 

27 transformed: list[str] 

28 strategy: TransformationStrategy 

29 metadata: dict = field(default_factory=dict) 

30 

31 def __len__(self) -> int: 

32 return len(self.transformed) 

33 

34 def __iter__(self) -> Any: 

35 return iter(self.transformed) 

36 

37 

38class AbstractQueryTransformer(ABC): 

39 """Abstract base class for query transformers.""" 

40 

41 @abstractmethod 

42 async def transform(self, query: str) -> TransformedQuery: 

43 """Transform a query.""" 

44 

45 @property 

46 @abstractmethod 

47 def strategy(self) -> TransformationStrategy: 

48 """Get the transformation strategy.""" 

49 

50 

51@dataclass(init=False) 

52class TransformationConfig(DomainModel): 

53 """Configuration for query transformation pipeline.""" 

54 

55 strategies: list[TransformationStrategy] = Field( 

56 default_factory=lambda: [TransformationStrategy.EXPANSION], 

57 description="Transformation strategies to apply", 

58 ) 

59 combine_results: bool = Field( 

60 default=True, 

61 description="Combine results from all strategies", 

62 ) 

63 max_total_queries: int = Field( 

64 default=10, 

65 ge=1, 

66 description="Maximum total queries to generate", 

67 )