Coverage for /home/admin/Documents/AI/applications/lexigram-dev/lexigram/experimental/ai/lexigram-ai-rag/src/lexigram/ai/rag/synthesis/synthesizers/direct.py: 22%

1"""Direct synthesizer implementation. 

2 

3This module implements a simple direct concatenation synthesizer that combines 

4context chunks with minimal processing. 

5""" 

6 

7from __future__ import annotations 

8 

9from lexigram.ai.rag.synthesis.synthesizers.base import AbstractSynthesizer 

10from lexigram.ai.rag.synthesis.types import ( 

11 ContextChunk, 

12 SynthesisResult, 

13 SynthesisStrategy, 

14) 

15 

16 

17class DirectSynthesizer(AbstractSynthesizer): 

18 """Direct concatenation synthesizer. 

19 

20 This synthesizer simply concatenates context chunks with minimal processing. 

21 It's the fastest approach but may produce less coherent responses. 

22 

23 Attributes: 

24 separator: Separator between chunks (default: double newline) 

25 max_chunks: Maximum number of chunks to include (None = all) 

26 include_sources: Whether to include source citations 

27 """ 

28 

29 def __init__( 

30 self, 

31 separator: str = "\n\n", 

32 max_chunks: int | None = None, 

33 include_sources: bool = True, 

34 ): 

35 """Initialize the direct synthesizer. 

36 

37 Args: 

38 separator: Separator between chunks 

39 max_chunks: Maximum number of chunks to include 

40 include_sources: Whether to include source citations 

41 """ 

42 self.separator = separator 

43 self.max_chunks = max_chunks 

44 self.include_sources = include_sources 

45 

46 async def _synthesize_internal( 

47 self, 

48 query: str, 

49 context_chunks: list[ContextChunk], 

50 **kwargs, 

51 ) -> SynthesisResult: 

52 """Synthesize response by concatenating context chunks. 

53 

54 Args: 

55 query: The user query 

56 context_chunks: Retrieved context chunks 

57 **kwargs: Additional parameters (ignored) 

58 

59 Returns: 

60 SynthesisResult with concatenated response 

61 

62 Raises: 

63 ValueError: If query is empty or no context chunks provided 

64 """ 

65 if not query: 

66 msg = "Query cannot be empty" 

67 raise ValueError(msg) 

68 if not context_chunks: 

69 msg = "No context chunks provided" 

70 raise ValueError(msg) 

71 

72 # Sort by rank (or score if rank not set) 

73 sorted_chunks = sorted( 

74 context_chunks, 

75 key=lambda c: c.rank if c.rank is not None else -c.score, 

76 ) 

77 

78 # Limit number of chunks if specified 

79 chunks_to_use = ( 

80 sorted_chunks[: self.max_chunks] if self.max_chunks else sorted_chunks 

81 ) 

82 

83 # Build response text 

84 response_parts: list[str] = [] 

85 

86 for i, chunk in enumerate(chunks_to_use, 1): 

87 if self.include_sources: 

88 response_parts.append(f"[{i}] {chunk.text}") 

89 else: 

90 response_parts.append(chunk.text) 

91 

92 response = self.separator.join(response_parts) 

93 

94 # Build citations if sources included 

95 citations = [] 

96 if self.include_sources: 

97 citations = [ 

98 { 

99 "number": i, 

100 "source": chunk.source, 

101 "score": chunk.score, 

102 "metadata": chunk.metadata, 

103 } 

104 for i, chunk in enumerate(chunks_to_use, 1) 

105 ] 

106 

107 return SynthesisResult( 

108 query=query, 

109 response=response, 

110 strategy=SynthesisStrategy.DIRECT, 

111 context_chunks=chunks_to_use, 

112 citations=citations, 

113 metadata={ 

114 "num_chunks": len(chunks_to_use), 

115 "separator": self.separator, 

116 "include_sources": self.include_sources, 

117 }, 

118 )