Coverage for src/local_deep_research/embeddings/providers/base.py: 92%

1"""Base class for embedding providers.""" 

2 

3from abc import ABC, abstractmethod 

4from typing import Any, Dict, List, Optional 

5 

6from langchain_core.embeddings import Embeddings 

7 

8from ...security.egress.classification import Exposure 

9 

10 

11class BaseEmbeddingProvider(ABC): 

12 """ 

13 Abstract base class for embedding providers. 

14 

15 All embedding providers should inherit from this class and implement 

16 the required methods. This provides a consistent interface similar to 

17 the LLM provider system. 

18 """ 

19 

20 # Override these in subclasses 

21 provider_name = "base" # Display name for logs/UI 

22 provider_key = "BASE" # Unique identifier (uppercase) 

23 requires_api_key = False # Whether this provider requires an API key 

24 supports_local = False # Whether this runs locally 

25 default_model = None # Default embedding model 

26 # Egress exposure (ADR-0007): declares whether this is an exposing 

27 # inference sink (cloud = exposing, local = contained). Declarative default 

28 # only — the egress resolver classifies the provider's ACTUAL endpoint at 

29 # run time (run_classification.embeddings_label via evaluate_embeddings), so 

30 # a URL-configurable provider is refined then; this documents intent. 

31 egress_exposure = Exposure.EXPOSING 

32 

33 @classmethod 

34 @abstractmethod 

35 def create_embeddings( 

36 cls, 

37 model: Optional[str] = None, 

38 settings_snapshot: Optional[Dict[str, Any]] = None, 

39 **kwargs, 

40 ) -> Embeddings: 

41 """ 

42 Create an embeddings instance for this provider. 

43 

44 Args: 

45 model: Name of the embedding model to use 

46 settings_snapshot: Optional settings snapshot for thread-safe access 

47 **kwargs: Additional provider-specific parameters 

48 

49 Returns: 

50 A LangChain Embeddings instance 

51 

52 Raises: 

53 ValueError: If required configuration is missing 

54 ImportError: If required dependencies are not installed 

55 """ 

56 pass 

57 

58 @classmethod 

59 @abstractmethod 

60 def is_available( 

61 cls, settings_snapshot: Optional[Dict[str, Any]] = None 

62 ) -> bool: 

63 """ 

64 Check if this embedding provider is available and properly configured. 

65 

66 Args: 

67 settings_snapshot: Optional settings snapshot for thread-safe access 

68 

69 Returns: 

70 True if the provider can be used, False otherwise 

71 """ 

72 pass 

73 

74 @classmethod 

75 def get_available_models( 

76 cls, settings_snapshot: Optional[Dict[str, Any]] = None 

77 ) -> List[Dict[str, Any]]: 

78 """ 

79 Get list of available embedding models for this provider. 

80 

81 Implementations should return every model the backend reports. 

82 Filtering by name is unreliable — users may load custom or 

83 renamed embedding models — so leave the choice to the user and 

84 only tag entries when a real capability signal is available. 

85 

86 Args: 

87 settings_snapshot: Optional settings snapshot 

88 

89 Returns: 

90 List of dicts with ``value`` and ``label`` string keys for 

91 each model. May include an optional ``is_embedding`` (bool) 

92 key when the provider can detect embedding capability from 

93 the backend (e.g. Ollama's ``/api/show`` capabilities). 

94 """ 

95 return [] 

96 

97 @classmethod 

98 def is_embedding_model( 

99 cls, 

100 model: str, 

101 settings_snapshot: Optional[Dict[str, Any]] = None, 

102 ) -> Optional[bool]: 

103 """ 

104 Check whether a specific model supports embeddings. 

105 

106 Providers that can distinguish embedding models from chat/LLM models 

107 should override this method. 

108 

109 Args: 

110 model: Model identifier 

111 settings_snapshot: Optional settings snapshot 

112 

113 Returns: 

114 True if the model supports embeddings, False if it does not, 

115 None if the provider cannot determine this. 

116 """ 

117 return None 

118 

119 @classmethod 

120 def get_model_info(cls, model: str) -> Optional[Dict[str, Any]]: 

121 """ 

122 Get information about a specific model. 

123 

124 Args: 

125 model: Model identifier 

126 

127 Returns: 

128 Dict with model metadata (dimensions, description, etc.) or None 

129 """ 

130 return None 

131 

132 @classmethod 

133 def validate_config( 

134 cls, settings_snapshot: Optional[Dict[str, Any]] = None 

135 ) -> tuple[bool, Optional[str]]: 

136 """ 

137 Validate the provider configuration. 

138 

139 Args: 

140 settings_snapshot: Optional settings snapshot 

141 

142 Returns: 

143 Tuple of (is_valid, error_message) 

144 """ 

145 if not cls.is_available(settings_snapshot): 

146 return ( 

147 False, 

148 f"{cls.provider_name} is not available or not configured", 

149 ) 

150 return True, None 

151 

152 @classmethod 

153 def get_provider_info(cls) -> Dict[str, Any]: 

154 """ 

155 Get metadata about this provider. 

156 

157 Returns: 

158 Dict with provider information 

159 """ 

160 return { 

161 "name": cls.provider_name, 

162 "key": cls.provider_key, 

163 "requires_api_key": cls.requires_api_key, 

164 "supports_local": cls.supports_local, 

165 "default_model": cls.default_model, 

166 }