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

1"""Single HyDE generator implementation.""" 

2 

3from __future__ import annotations 

4 

5from typing import Any 

6 

7from lexigram.ai.rag.hyde.base import AbstractHyDEGenerator 

8from lexigram.ai.rag.hyde.protocols import EmbeddingClientProtocol 

9from lexigram.ai.rag.hyde.types import HyDEResult, HyDEStrategy, HypotheticalDocument 

10from lexigram.contracts import ( 

11 LLMClientProtocol, 

12) 

13 

14 

15class SingleHyDEGenerator(AbstractHyDEGenerator): 

16 """Generator for single hypothetical document.""" 

17 

18 def __init__( 

19 self, 

20 llm_client: LLMClientProtocol, 

21 embedding_client: EmbeddingClientProtocol | None = None, 

22 temperature: float = 0.7, 

23 max_tokens: int = 200, 

24 ): 

25 """Initialize single HyDE generator. 

26 

27 Args: 

28 llm_client: Client for generating hypothetical documents 

29 embedding_client: Optional client for generating embeddings 

30 temperature: Sampling temperature for generation 

31 max_tokens: Maximum tokens per document 

32 """ 

33 super().__init__(llm_client, embedding_client) 

34 self.temperature = temperature 

35 self.max_tokens = max_tokens 

36 

37 async def generate( 

38 self, 

39 query: str, 

40 num_documents: int = 1, 

41 **kwargs: Any, 

42 ) -> HyDEResult: 

43 """Generate single hypothetical document. 

44 

45 Args: 

46 query: User query 

47 num_documents: Ignored (always generates 1) 

48 **kwargs: Additional parameters (context, domain) 

49 

50 Returns: 

51 HyDE result with single hypothetical document 

52 """ 

53 content = await self._generate_single_document( 

54 query, 

55 temperature=self.temperature, 

56 max_tokens=self.max_tokens, 

57 **kwargs, 

58 ) 

59 

60 doc = HypotheticalDocument( 

61 content=content, 

62 query=query, 

63 confidence=1.0, 

64 metadata={ 

65 "temperature": self.temperature, 

66 "max_tokens": self.max_tokens, 

67 }, 

68 ) 

69 

70 # Generate embedding if client available 

71 aggregated_embedding = None 

72 if self.embedding_client: 

73 embeddings = await self._embed_documents([doc]) 

74 aggregated_embedding = embeddings[0] if embeddings else None 

75 

76 return HyDEResult( 

77 query=query, 

78 hypothetical_docs=[doc], 

79 strategy=HyDEStrategy.SINGLE, 

80 aggregated_embedding=aggregated_embedding, 

81 metadata={ 

82 "temperature": self.temperature, 

83 "max_tokens": self.max_tokens, 

84 }, 

85 )