Coverage for src / lexigram / contracts / ai / retrievers.py: 0%

1"""Retriever protocols and value objects for Lexigram. 

2 

3Defines contracts for document retrieval and node postprocessing, 

4analogous to LangChain's retriever interfaces. 

5""" 

6 

7from __future__ import annotations 

8 

9from dataclasses import dataclass, field 

10from typing import TYPE_CHECKING, Any, Protocol, runtime_checkable 

11 

12if TYPE_CHECKING: 

13 from lexigram.contracts.core.result import Result 

14 

15from lexigram.contracts.ai.exceptions import RetrieverError 

16 

17__all__ = [ 

18 "NodePostprocessorProtocol", 

19 "RetrievalQuery", 

20 "RetrievedNode", 

21 "RetrieverError", 

22 "RetrieverProtocol", 

23] 

24 

25 

26@dataclass(frozen=True) 

27class RetrievalQuery: 

28 """Query parameters for document retrieval. 

29 

30 Attributes: 

31 query: The search query text. 

32 top_k: Number of results to return. 

33 """ 

34 

35 query: str 

36 top_k: int = 10 

37 

38 

39@dataclass(frozen=True) 

40class RetrievedNode: 

41 """A single retrieved document node. 

42 

43 Attributes: 

44 id: Unique identifier for the node. 

45 content: Text content of the node. 

46 score: Relevance score (typically 0-1). 

47 metadata: Optional metadata dictionary. 

48 """ 

49 

50 id: str 

51 content: str 

52 score: float 

53 metadata: dict[str, Any] = field(default_factory=dict) 

54 

55 

56@runtime_checkable 

57class RetrieverProtocol(Protocol): 

58 """Protocol for document retrieval. 

59 

60 Implementations provide async retrieval of relevant documents 

61 based on a query string. 

62 """ 

63 

64 async def retrieve( 

65 self, query: str, top_k: int = 10 

66 ) -> Result[list[RetrievedNode], RetrieverError]: 

67 """Retrieve relevant documents for a query. 

68 

69 Args: 

70 query: The search query text. 

71 top_k: Number of results to return. 

72 

73 Returns: 

74 Ok(list of RetrievedNode) on success. 

75 Err(RetrieverError) on failure. 

76 """ 

77 ... 

78 

79 

80@runtime_checkable 

81class NodePostprocessorProtocol(Protocol): 

82 """Protocol for postprocessing retrieved nodes. 

83 

84 Implementations transform, filter, or enrich retrieved nodes 

85 after initial retrieval. 

86 """ 

87 

88 async def postprocess( 

89 self, nodes: list[RetrievedNode] 

90 ) -> Result[list[RetrievedNode], RetrieverError]: 

91 """Postprocess retrieved nodes. 

92 

93 Args: 

94 nodes: List of retrieved nodes to process. 

95 

96 Returns: 

97 Ok(list of processed RetrievedNode) on success. 

98 Err(RetrieverError) on failure. 

99 """ 

100 ...