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

1from __future__ import annotations 

2 

3from lexigram.ai.rag.query.base import AbstractQueryTransformer, TransformedQuery 

4 

5 

6class TransformationPipeline: 

7 """Pipeline for applying multiple query transformations.""" 

8 

9 def __init__( 

10 self, 

11 transformers: list[AbstractQueryTransformer], 

12 combine_results: bool = True, 

13 max_total_queries: int = 10, 

14 ): 

15 self.transformers = transformers 

16 self.combine_results = combine_results 

17 self.max_total_queries = max_total_queries 

18 

19 async def transform(self, query: str) -> list[TransformedQuery]: 

20 """Apply all transformers to query.""" 

21 results = [] 

22 for transformer in self.transformers: 

23 result = await transformer.transform(query) 

24 results.append(result) 

25 return results 

26 

27 async def transform_combined(self, query: str) -> list[str]: 

28 """Apply transformers and combine results.""" 

29 all_results = await self.transform(query) 

30 combined = [] 

31 seen = set() 

32 

33 for result in all_results: 

34 for q in result.transformed: 

35 if q not in seen: 

36 combined.append(q) 

37 seen.add(q) 

38 if len(combined) >= self.max_total_queries: 

39 return combined 

40 return combined