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

1"""Hybrid routing strategy combining multiple strategies.""" 

2 

3from __future__ import annotations 

4 

5from lexigram.ai.rag.routing.strategies.base import RoutingStrategy 

6from lexigram.ai.rag.routing.types import ( 

7 DataSource, 

8 QueryFeatures, 

9 RoutingDecision, 

10) 

11from lexigram.logging import ( 

12 get_logger, 

13) 

14 

15logger = get_logger(__name__) 

16 

17 

18class HybridRouter: 

19 """Hybrid routing strategy combining multiple routers. 

20 

21 Tries strategies in order until one returns a confident decision, 

22 or combines results from multiple strategies using ensemble voting. 

23 

24 Example: 

25 ```python 

26 from lexigram.ai.rag import ( 

27 HybridRouter, 

28 RuleBasedRouter, 

29 SemanticRouter, 

30 LLMRouter 

31 ) 

32 

33 hybrid = HybridRouter( 

34 strategies=[ 

35 RuleBasedRouter.with_defaults(), 

36 SemanticRouter.with_defaults(embed_fn=embed), 

37 LLMRouter(llm_fn=llm), 

38 ], 

39 confidence_threshold=0.7, 

40 use_ensemble=False # Try in order 

41 ) 

42 

43 decision = await hybrid.route(features, available_sources) 

44 ``` 

45 """ 

46 

47 def __init__( 

48 self, 

49 *, 

50 strategies: list[RoutingStrategy] | None = None, 

51 confidence_threshold: float = 0.7, 

52 use_ensemble: bool = False, 

53 ): 

54 """Initialize the hybrid router. 

55 

56 Args: 

57 strategies: List of routing strategies to use. 

58 confidence_threshold: Confidence threshold for accepting decisions. 

59 use_ensemble: Whether to use ensemble voting (combine all strategies). 

60 """ 

61 self.strategies = strategies or [] 

62 self.confidence_threshold = confidence_threshold 

63 self.use_ensemble = use_ensemble 

64 

65 def add_strategy(self, strategy: RoutingStrategy) -> None: 

66 """Add a routing strategy. 

67 

68 Args: 

69 strategy: Routing strategy to add. 

70 """ 

71 self.strategies.append(strategy) 

72 

73 async def route( 

74 self, 

75 features: QueryFeatures, 

76 available_sources: list[DataSource], 

77 ) -> RoutingDecision: 

78 """Route query using hybrid strategy. 

79 

80 Args: 

81 features: Extracted query features. 

82 available_sources: List of available data sources. 

83 

84 Returns: 

85 Routing decision from hybrid approach. 

86 """ 

87 if not self.strategies: 

88 # No strategies configured, return default 

89 if available_sources: 

90 return RoutingDecision( 

91 query=features.text, 

92 data_sources=[available_sources[0]], 

93 strategy="dense", 

94 confidence=0.3, 

95 reasoning="No routing strategies configured", 

96 features=features, 

97 metadata={"error": "no_strategies"}, 

98 ) 

99 return RoutingDecision( 

100 query=features.text, 

101 data_sources=[], 

102 strategy="none", 

103 confidence=0.0, 

104 reasoning="No strategies or sources available", 

105 features=features, 

106 metadata={"error": "no_config"}, 

107 ) 

108 

109 if self.use_ensemble: 

110 return await self._ensemble_route(features, available_sources) 

111 return await self._cascade_route(features, available_sources) 

112 

113 async def _cascade_route( 

114 self, 

115 features: QueryFeatures, 

116 available_sources: list[DataSource], 

117 ) -> RoutingDecision: 

118 """Try strategies in order until confident decision. 

119 

120 Args: 

121 features: Query features. 

122 available_sources: Available data sources. 

123 

124 Returns: 

125 First confident routing decision. 

126 """ 

127 last_decision = None 

128 

129 for strategy in self.strategies: 

130 decision = await strategy.route(features, available_sources) 

131 

132 # Return if confident 

133 if decision.confidence >= self.confidence_threshold: 

134 decision.metadata["strategy_used"] = strategy.__class__.__name__ 

135 decision.metadata["cascade"] = True 

136 return decision 

137 

138 # Keep track of last decision 

139 last_decision = decision 

140 

141 # No confident decision, return last one 

142 if last_decision: 

143 last_decision.metadata["strategy_used"] = "last_fallback" 

144 last_decision.metadata["cascade"] = True 

145 last_decision.reasoning = f"No confident decision (best: {last_decision.confidence:.2f}). {last_decision.reasoning}" 

146 return last_decision 

147 

148 # Should not reach here 

149 return RoutingDecision( 

150 query=features.text, 

151 data_sources=[], 

152 strategy="none", 

153 confidence=0.0, 

154 reasoning="No routing decision made", 

155 features=features, 

156 metadata={"error": "no_decision"}, 

157 ) 

158 

159 async def _ensemble_route( 

160 self, 

161 features: QueryFeatures, 

162 available_sources: list[DataSource], 

163 ) -> RoutingDecision: 

164 """Combine decisions from all strategies using voting. 

165 

166 Args: 

167 features: Query features. 

168 available_sources: Available data sources. 

169 

170 Returns: 

171 Ensemble routing decision. 

172 """ 

173 # Get decisions from all strategies 

174 decisions = [] 

175 for strategy in self.strategies: 

176 try: 

177 decision = await strategy.route(features, available_sources) 

178 decisions.append(decision) 

179 except (RuntimeError, ValueError, TypeError, OSError) as e: 

180 logger.debug( 

181 "Strategy %s failed: %s", 

182 getattr(strategy, "name", str(strategy)), 

183 e, 

184 ) 

185 # Skip failed strategies 

186 continue 

187 

188 if not decisions: 

189 return RoutingDecision( 

190 query=features.text, 

191 data_sources=[], 

192 strategy="none", 

193 confidence=0.0, 

194 reasoning="All strategies failed", 

195 features=features, 

196 metadata={"error": "all_failed"}, 

197 ) 

198 

199 # Vote on data sources (weighted by confidence) 

200 source_votes: dict[str, float] = {} 

201 strategy_votes: dict[str, float] = {} 

202 

203 for decision in decisions: 

204 # Vote for data sources 

205 for source in decision.data_sources: 

206 source_votes[source.name] = ( 

207 source_votes.get(source.name, 0) + decision.confidence 

208 ) 

209 

210 # Vote for strategy 

211 strategy_votes[decision.strategy] = ( 

212 strategy_votes.get(decision.strategy, 0) + decision.confidence 

213 ) 

214 

215 # Select top data sources 

216 top_sources = sorted(source_votes.items(), key=lambda x: x[1], reverse=True) 

217 selected_source_names = [x[0] for x in top_sources[:3]] # Top 3 

218 

219 selected_sources = [ 

220 source 

221 for source in available_sources 

222 if source.name in selected_source_names 

223 ] 

224 

225 # Select top strategy 

226 top_strategy = ( 

227 max(strategy_votes.items(), key=lambda x: x[1])[0] 

228 if strategy_votes 

229 else "dense" 

230 ) 

231 

232 # Calculate ensemble confidence (average of top decisions) 

233 top_confidences = sorted( 

234 (d.confidence for d in decisions), 

235 reverse=True, 

236 )[:2] 

237 ensemble_confidence = sum(top_confidences) / len(top_confidences) 

238 

239 return RoutingDecision( 

240 query=features.text, 

241 data_sources=( 

242 selected_sources or [available_sources[0]] if available_sources else [] 

243 ), 

244 strategy=top_strategy, 

245 confidence=ensemble_confidence, 

246 reasoning=f"Ensemble decision from {len(decisions)} strategies", 

247 features=features, 

248 metadata={ 

249 "ensemble": True, 

250 "num_strategies": len(decisions), 

251 "source_votes": source_votes, 

252 "strategy_votes": strategy_votes, 

253 }, 

254 )