Coverage for /home/admin/Documents/AI/applications/lexigram-dev/lexigram/experimental/ai/lexigram-ai-rag/src/lexigram/ai/rag/routing/types.py: 84%

1"""Core types for query routing.""" 

2 

3from __future__ import annotations 

4 

5from dataclasses import dataclass, field 

6from enum import StrEnum 

7from typing import Any 

8 

9from lexigram.ai.rag.multimodal.types import Modality 

10 

11 

12class QueryIntent(StrEnum): 

13 """Intent classification for queries. 

14 

15 Attributes: 

16 FACTUAL: Questions seeking factual information. 

17 CONVERSATIONAL: Casual conversation or greetings. 

18 ANALYTICAL: Queries requiring comparison or analysis. 

19 CREATIVE: Requests for creative content generation. 

20 PROCEDURAL: How-to questions or instructions. 

21 NAVIGATIONAL: Queries seeking specific pages or resources. 

22 """ 

23 

24 FACTUAL = "factual" 

25 CONVERSATIONAL = "conversational" 

26 ANALYTICAL = "analytical" 

27 CREATIVE = "creative" 

28 PROCEDURAL = "procedural" 

29 NAVIGATIONAL = "navigational" 

30 

31 

32class DataSourceType(StrEnum): 

33 """Types of data sources for routing. 

34 

35 Attributes: 

36 VECTOR_STORE: Dense vector embeddings store. 

37 KEYWORD_INDEX: Sparse keyword-based index (BM25, TF-IDF). 

38 KNOWLEDGE_GRAPH: Graph database for structured knowledge. 

39 SQL_DATABASE: Relational database for structured queries. 

40 EXTERNAL_API: External API or web service. 

41 MULTIMODAL_STORE: Multi-modal content store (images, audio, video). 

42 """ 

43 

44 VECTOR_STORE = "vector_store" 

45 KEYWORD_INDEX = "keyword_index" 

46 KNOWLEDGE_GRAPH = "knowledge_graph" 

47 SQL_DATABASE = "sql_database" 

48 EXTERNAL_API = "external_api" 

49 MULTIMODAL_STORE = "multimodal_store" 

50 

51 

52@dataclass 

53class QueryFeatures: 

54 """Features extracted from a query for routing decisions. 

55 

56 Attributes: 

57 text: Original query text. 

58 length: Character count of the query. 

59 intent: Classified intent of the query. 

60 language: Detected language code (e.g., 'en', 'es'). 

61 domain: Optional domain classification (e.g., 'technical', 'medical'). 

62 keywords: Extracted keywords from the query. 

63 has_entities: Whether named entities were detected. 

64 modalities: Detected modalities (text, image, audio, video). 

65 complexity: Query complexity score (0-1). 

66 metadata: Additional metadata for routing decisions. 

67 """ 

68 

69 text: str 

70 length: int 

71 intent: QueryIntent 

72 language: str = "en" 

73 domain: str | None = None 

74 keywords: list[str] = field(default_factory=list) 

75 has_entities: bool = False 

76 modalities: list[Modality] = field(default_factory=lambda: [Modality.TEXT]) 

77 complexity: float = 0.5 

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

79 

80 @property 

81 def is_simple(self) -> bool: 

82 """Check if query is simple (low complexity).""" 

83 return self.complexity < 0.3 

84 

85 @property 

86 def is_complex(self) -> bool: 

87 """Check if query is complex (high complexity).""" 

88 return self.complexity > 0.7 

89 

90 @property 

91 def is_multimodal(self) -> bool: 

92 """Check if query involves multiple modalities.""" 

93 return len(self.modalities) > 1 or Modality.TEXT not in self.modalities 

94 

95 @property 

96 def is_long(self) -> bool: 

97 """Check if query is long (>200 chars).""" 

98 return self.length > 200 

99 

100 

101@dataclass 

102class DataSource: 

103 """Represents a data source for query routing. 

104 

105 Attributes: 

106 name: Unique identifier for the data source. 

107 type: Type of data source (vector store, keyword index, etc.). 

108 description: Human-readable description. 

109 capabilities: List of supported capabilities. 

110 priority: Priority for routing (higher = more preferred). 

111 metadata: Additional metadata about the data source. 

112 """ 

113 

114 name: str 

115 type: DataSourceType 

116 description: str 

117 capabilities: list[str] = field(default_factory=list) 

118 priority: int = 0 

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

120 

121 def supports(self, capability: str) -> bool: 

122 """Check if data source supports a capability. 

123 

124 Args: 

125 capability: Capability to check (e.g., 'dense_search'). 

126 

127 Returns: 

128 True if capability is supported, False otherwise. 

129 """ 

130 return capability in self.capabilities 

131 

132 def __hash__(self) -> int: 

133 """Make DataSourceProtocol hashable for use in sets/dicts.""" 

134 return hash(self.name) 

135 

136 def __eq__(self, other: object) -> bool: 

137 """Check equality based on name.""" 

138 if not isinstance(other, DataSource): 

139 return NotImplemented 

140 return self.name == other.name 

141 

142 

143@dataclass 

144class RoutingDecision: 

145 """Result of a routing decision. 

146 

147 Attributes: 

148 query: Original query text. 

149 data_sources: Selected data sources for the query. 

150 strategy: Retrieval strategy to use. 

151 confidence: Confidence score (0-1) in the routing decision. 

152 reasoning: Human-readable explanation of the decision. 

153 features: Query features used for routing. 

154 metadata: Additional metadata about the routing decision. 

155 """ 

156 

157 query: str 

158 data_sources: list[DataSource] 

159 strategy: str 

160 confidence: float 

161 reasoning: str 

162 features: QueryFeatures | None = None 

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

164 

165 @property 

166 def primary_source(self) -> DataSource | None: 

167 """Get the primary (first) data source.""" 

168 return self.data_sources[0] if self.data_sources else None 

169 

170 @property 

171 def is_confident(self) -> bool: 

172 """Check if routing decision is confident (>0.7).""" 

173 return self.confidence > 0.7 

174 

175 @property 

176 def is_multimodal(self) -> bool: 

177 """Check if routing involves multimodal sources.""" 

178 return any( 

179 source.type == DataSourceType.MULTIMODAL_STORE 

180 for source in self.data_sources 

181 ) 

182 

183 def to_dict(self) -> dict[str, Any]: 

184 """Convert routing decision to dictionary. 

185 

186 Returns: 

187 Dictionary representation of the routing decision. 

188 """ 

189 return { 

190 "query": self.query, 

191 "data_sources": [ 

192 { 

193 "name": source.name, 

194 "type": source.type.value, 

195 "description": source.description, 

196 } 

197 for source in self.data_sources 

198 ], 

199 "strategy": self.strategy, 

200 "confidence": self.confidence, 

201 "reasoning": self.reasoning, 

202 "features": ( 

203 { 

204 "intent": self.features.intent.value, 

205 "language": self.features.language, 

206 "domain": self.features.domain, 

207 "modalities": [m.value for m in self.features.modalities], 

208 } 

209 if self.features 

210 else None 

211 ), 

212 "metadata": self.metadata, 

213 }