Coverage for haystack/components/generators/chat/fallback.py: 97%

1# SPDX-FileCopyrightText: 2022-present deepset GmbH <info@deepset.ai> 

2# 

3# SPDX-License-Identifier: Apache-2.0 

4 

5from __future__ import annotations 

6 

7from typing import Any 

8 

9from haystack import component, default_from_dict, default_to_dict, logging 

10from haystack.components.generators.chat.types import ChatGenerator 

11from haystack.components.generators.utils import _normalize_messages 

12from haystack.core.serialization import component_to_dict 

13from haystack.dataclasses import ChatMessage, StreamingCallbackT 

14from haystack.tools import ToolsType 

15from haystack.utils.async_utils import _execute_component_async 

16from haystack.utils.deserialization import deserialize_component_inplace 

17 

18logger = logging.getLogger(__name__) 

19 

20 

21@component 

22class FallbackChatGenerator: 

23 """ 

24 A chat generator wrapper that tries multiple chat generators sequentially. 

25 

26 It forwards all parameters transparently to the underlying chat generators and returns the first successful result. 

27 Calls chat generators sequentially until one succeeds. Falls back on any exception raised by a generator. 

28 If all chat generators fail, it raises a RuntimeError with details. 

29 

30 Timeout enforcement is fully delegated to the underlying chat generators. The fallback mechanism will only 

31 work correctly if the underlying chat generators implement proper timeout handling and raise exceptions 

32 when timeouts occur. For predictable latency guarantees, ensure your chat generators: 

33 - Support a `timeout` parameter in their initialization 

34 - Implement timeout as total wall-clock time (shared deadline for both streaming and non-streaming) 

35 - Raise timeout exceptions (e.g., TimeoutError, asyncio.TimeoutError, httpx.TimeoutException) when exceeded 

36 

37 Note: Most well-implemented chat generators (OpenAI, Anthropic, Cohere, etc.) support timeout parameters 

38 with consistent semantics. For HTTP-based LLM providers, a single timeout value (e.g., `timeout=30`) 

39 typically applies to all connection phases: connection setup, read, write, and pool. For streaming 

40 responses, read timeout is the maximum gap between chunks. For non-streaming, it's the time limit for 

41 receiving the complete response. 

42 

43 Fail over is automatically triggered when a generator raises any exception, including: 

44 - Timeout errors (if the generator implements and raises them) 

45 - Rate limit errors (429) 

46 - Authentication errors (401) 

47 - Context length errors (400) 

48 - Server errors (500+) 

49 - Any other exception 

50 """ 

51 

52 def __init__(self, chat_generators: list[ChatGenerator]) -> None: 

53 """ 

54 Creates an instance of FallbackChatGenerator. 

55 

56 :param chat_generators: A non-empty list of chat generator components to try in order. 

57 """ 

58 if not chat_generators: 

59 msg = "'chat_generators' must be a non-empty list" 

60 raise ValueError(msg) 

61 

62 self.chat_generators = list(chat_generators) 

63 

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

65 """Serialize the component, including nested chat generators.""" 

66 return default_to_dict( 

67 self, 

68 chat_generators=[ 

69 component_to_dict(gen, name=f"chat_generator_{idx}") for idx, gen in enumerate(self.chat_generators) 

70 ], 

71 ) 

72 

73 @classmethod 

74 def from_dict(cls, data: dict[str, Any]) -> FallbackChatGenerator: 

75 """Rebuild the component from a serialized representation, restoring nested chat generators.""" 

76 # Reconstruct nested chat generators from their serialized dicts 

77 init_params = data.get("init_parameters", {}) 

78 serialized = init_params.get("chat_generators") or [] 

79 deserialized: list[Any] = [] 

80 for g in serialized: 

81 # Use the generic component deserializer available in Haystack 

82 holder = {"component": g} 

83 deserialize_component_inplace(holder, key="component") 

84 deserialized.append(holder["component"]) 

85 init_params["chat_generators"] = deserialized 

86 data["init_parameters"] = init_params 

87 return default_from_dict(cls, data) 

88 

89 def warm_up(self) -> None: 

90 """Warm up all underlying chat generators.""" 

91 for gen in self.chat_generators: 

92 if hasattr(gen, "warm_up"): 

93 gen.warm_up() 

94 

95 async def warm_up_async(self) -> None: 

96 """Warm up all underlying chat generators on the serving event loop.""" 

97 for gen in self.chat_generators: 

98 if hasattr(gen, "warm_up_async"): 

99 await gen.warm_up_async() 

100 elif hasattr(gen, "warm_up"): 

101 gen.warm_up() 

102 

103 def close(self) -> None: 

104 """Release the underlying chat generators' resources.""" 

105 for gen in self.chat_generators: 

106 if hasattr(gen, "close"): 

107 gen.close() 

108 

109 async def close_async(self) -> None: 

110 """Release the underlying chat generators' async resources.""" 

111 for gen in self.chat_generators: 

112 if hasattr(gen, "close_async"): 

113 await gen.close_async() 

114 elif hasattr(gen, "close"): 

115 gen.close() 

116 

117 def _run_single_sync( 

118 self, 

119 gen: Any, 

120 messages: list[ChatMessage], 

121 generation_kwargs: dict[str, Any] | None, 

122 tools: ToolsType | None, 

123 streaming_callback: StreamingCallbackT | None, 

124 ) -> dict[str, Any]: 

125 return gen.run( 

126 messages=messages, generation_kwargs=generation_kwargs, tools=tools, streaming_callback=streaming_callback 

127 ) 

128 

129 async def _run_single_async( 

130 self, 

131 gen: Any, 

132 messages: list[ChatMessage], 

133 generation_kwargs: dict[str, Any] | None, 

134 tools: ToolsType | None, 

135 streaming_callback: StreamingCallbackT | None, 

136 ) -> dict[str, Any]: 

137 return await _execute_component_async( 

138 gen, 

139 messages=messages, 

140 generation_kwargs=generation_kwargs, 

141 tools=tools, 

142 streaming_callback=streaming_callback, 

143 ) 

144 

145 @component.output_types(replies=list[ChatMessage], meta=dict[str, Any]) 

146 def run( 

147 self, 

148 messages: list[ChatMessage] | str, 

149 generation_kwargs: dict[str, Any] | None = None, 

150 tools: ToolsType | None = None, 

151 streaming_callback: StreamingCallbackT | None = None, 

152 ) -> dict[str, list[ChatMessage] | dict[str, Any]]: 

153 """ 

154 Execute chat generators sequentially until one succeeds. 

155 

156 :param messages: The conversation history as a list of ChatMessage instances. 

157 :param generation_kwargs: Optional parameters for the chat generator (e.g., temperature, max_tokens). 

158 :param tools: A list of Tool and/or Toolset objects, or a single Toolset for function calling capabilities. 

159 :param streaming_callback: Optional callable for handling streaming responses. 

160 :returns: A dictionary with: 

161 - "replies": Generated ChatMessage instances from the first successful generator. 

162 - "meta": Execution metadata including successful_chat_generator_index, successful_chat_generator_class, 

163 total_attempts, failed_chat_generators, plus any metadata from the successful generator. 

164 :raises RuntimeError: If all chat generators fail. 

165 """ 

166 self.warm_up() 

167 

168 messages = _normalize_messages(messages) 

169 

170 failed: list[str] = [] 

171 last_error: BaseException | None = None 

172 

173 for idx, gen in enumerate(self.chat_generators): 

174 gen_name = gen.__class__.__name__ 

175 try: 

176 result = self._run_single_sync(gen, messages, generation_kwargs, tools, streaming_callback) 

177 replies = result.get("replies", []) 

178 meta = dict(result.get("meta", {})) 

179 meta.update( 

180 { 

181 "successful_chat_generator_index": idx, 

182 "successful_chat_generator_class": gen_name, 

183 "total_attempts": idx + 1, 

184 "failed_chat_generators": failed, 

185 } 

186 ) 

187 return {"replies": replies, "meta": meta} 

188 except Exception as e: # noqa: BLE001 - fallback logic should handle any exception 

189 logger.warning( 

190 "ChatGenerator {chat_generator} failed with error: {error}", chat_generator=gen_name, error=e 

191 ) 

192 failed.append(gen_name) 

193 last_error = e 

194 

195 failed_names = ", ".join(failed) 

196 msg = ( 

197 f"All {len(self.chat_generators)} chat generators failed. " 

198 f"Last error: {last_error}. Failed chat generators: [{failed_names}]" 

199 ) 

200 raise RuntimeError(msg) 

201 

202 @component.output_types(replies=list[ChatMessage], meta=dict[str, Any]) 

203 async def run_async( 

204 self, 

205 messages: list[ChatMessage] | str, 

206 generation_kwargs: dict[str, Any] | None = None, 

207 tools: ToolsType | None = None, 

208 streaming_callback: StreamingCallbackT | None = None, 

209 ) -> dict[str, list[ChatMessage] | dict[str, Any]]: 

210 """ 

211 Asynchronously execute chat generators sequentially until one succeeds. 

212 

213 :param messages: The conversation history as a list of ChatMessage instances. 

214 :param generation_kwargs: Optional parameters for the chat generator (e.g., temperature, max_tokens). 

215 :param tools: A list of Tool and/or Toolset objects, or a single Toolset for function calling capabilities. 

216 :param streaming_callback: Optional callable for handling streaming responses. 

217 :returns: A dictionary with: 

218 - "replies": Generated ChatMessage instances from the first successful generator. 

219 - "meta": Execution metadata including successful_chat_generator_index, successful_chat_generator_class, 

220 total_attempts, failed_chat_generators, plus any metadata from the successful generator. 

221 :raises RuntimeError: If all chat generators fail. 

222 """ 

223 await self.warm_up_async() 

224 

225 messages = _normalize_messages(messages) 

226 

227 failed: list[str] = [] 

228 last_error: BaseException | None = None 

229 

230 for idx, gen in enumerate(self.chat_generators): 

231 gen_name = gen.__class__.__name__ 

232 try: 

233 result = await self._run_single_async(gen, messages, generation_kwargs, tools, streaming_callback) 

234 replies = result.get("replies", []) 

235 meta = dict(result.get("meta", {})) 

236 meta.update( 

237 { 

238 "successful_chat_generator_index": idx, 

239 "successful_chat_generator_class": gen_name, 

240 "total_attempts": idx + 1, 

241 "failed_chat_generators": failed, 

242 } 

243 ) 

244 return {"replies": replies, "meta": meta} 

245 except Exception as e: # noqa: BLE001 - fallback logic should handle any exception 

246 logger.warning( 

247 "ChatGenerator {chat_generator} failed with error: {error}", chat_generator=gen_name, error=e 

248 ) 

249 failed.append(gen_name) 

250 last_error = e 

251 

252 failed_names = ", ".join(failed) 

253 msg = ( 

254 f"All {len(self.chat_generators)} chat generators failed. " 

255 f"Last error: {last_error}. Failed chat generators: [{failed_names}]" 

256 ) 

257 raise RuntimeError(msg)