Coverage for haystack/components/generators/chat/fallback.py: 97%
102 statements
« prev ^ index » next coverage.py v7.16.1, created at 2026-10-09 14:34 +0000
1# SPDX-FileCopyrightText: 2022-present deepset GmbH <info@deepset.ai>
2#
3# SPDX-License-Identifier: Apache-2.0
5from __future__ import annotations
7from typing import Any
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, _trace_chat_generator_run
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
18logger = logging.getLogger(__name__)
21@component
22class FallbackChatGenerator:
23 """
24 A chat generator wrapper that tries multiple chat generators sequentially.
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.
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
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.
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 """
52 def __init__(self, chat_generators: list[ChatGenerator]) -> None:
53 """
54 Creates an instance of FallbackChatGenerator.
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)
62 self.chat_generators = list(chat_generators)
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 )
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)
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()
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()
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()
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()
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 generator_inputs = {
126 "messages": messages,
127 "generation_kwargs": generation_kwargs,
128 "tools": tools,
129 "streaming_callback": streaming_callback,
130 }
131 with _trace_chat_generator_run(chat_generator=gen, generator_inputs=generator_inputs):
132 return gen.run(**generator_inputs)
134 async def _run_single_async(
135 self,
136 gen: Any,
137 messages: list[ChatMessage],
138 generation_kwargs: dict[str, Any] | None,
139 tools: ToolsType | None,
140 streaming_callback: StreamingCallbackT | None,
141 ) -> dict[str, Any]:
142 generator_inputs = {
143 "messages": messages,
144 "generation_kwargs": generation_kwargs,
145 "tools": tools,
146 "streaming_callback": streaming_callback,
147 }
148 with _trace_chat_generator_run(chat_generator=gen, generator_inputs=generator_inputs):
149 return await _execute_component_async(component_instance=gen, **generator_inputs)
151 @component.output_types(replies=list[ChatMessage], meta=dict[str, Any])
152 def run(
153 self,
154 messages: list[ChatMessage] | str,
155 generation_kwargs: dict[str, Any] | None = None,
156 tools: ToolsType | None = None,
157 streaming_callback: StreamingCallbackT | None = None,
158 ) -> dict[str, list[ChatMessage] | dict[str, Any]]:
159 """
160 Execute chat generators sequentially until one succeeds.
162 :param messages: The conversation history as a list of ChatMessage instances.
163 :param generation_kwargs: Optional parameters for the chat generator (e.g., temperature, max_tokens).
164 :param tools: A list of Tool and/or Toolset objects, or a single Toolset for function calling capabilities.
165 :param streaming_callback: Optional callable for handling streaming responses.
166 :returns: A dictionary with:
167 - "replies": Generated ChatMessage instances from the first successful generator.
168 - "meta": Execution metadata including successful_chat_generator_index, successful_chat_generator_class,
169 total_attempts, failed_chat_generators, plus any metadata from the successful generator.
170 :raises RuntimeError: If all chat generators fail.
171 """
172 self.warm_up()
174 messages = _normalize_messages(messages=messages)
176 failed: list[str] = []
177 last_error: BaseException | None = None
179 for idx, gen in enumerate(self.chat_generators):
180 gen_name = gen.__class__.__name__
181 try:
182 result = self._run_single_sync(gen, messages, generation_kwargs, tools, streaming_callback)
183 replies = result.get("replies", [])
184 meta = dict(result.get("meta", {}))
185 meta.update(
186 {
187 "successful_chat_generator_index": idx,
188 "successful_chat_generator_class": gen_name,
189 "total_attempts": idx + 1,
190 "failed_chat_generators": failed,
191 }
192 )
193 return {"replies": replies, "meta": meta}
194 except Exception as e: # noqa: BLE001 - fallback logic should handle any exception
195 logger.warning(
196 "ChatGenerator {chat_generator} failed with error: {error}", chat_generator=gen_name, error=e
197 )
198 failed.append(gen_name)
199 last_error = e
201 failed_names = ", ".join(failed)
202 msg = (
203 f"All {len(self.chat_generators)} chat generators failed. "
204 f"Last error: {last_error}. Failed chat generators: [{failed_names}]"
205 )
206 raise RuntimeError(msg)
208 @component.output_types(replies=list[ChatMessage], meta=dict[str, Any])
209 async def run_async(
210 self,
211 messages: list[ChatMessage] | str,
212 generation_kwargs: dict[str, Any] | None = None,
213 tools: ToolsType | None = None,
214 streaming_callback: StreamingCallbackT | None = None,
215 ) -> dict[str, list[ChatMessage] | dict[str, Any]]:
216 """
217 Asynchronously execute chat generators sequentially until one succeeds.
219 :param messages: The conversation history as a list of ChatMessage instances.
220 :param generation_kwargs: Optional parameters for the chat generator (e.g., temperature, max_tokens).
221 :param tools: A list of Tool and/or Toolset objects, or a single Toolset for function calling capabilities.
222 :param streaming_callback: Optional callable for handling streaming responses.
223 :returns: A dictionary with:
224 - "replies": Generated ChatMessage instances from the first successful generator.
225 - "meta": Execution metadata including successful_chat_generator_index, successful_chat_generator_class,
226 total_attempts, failed_chat_generators, plus any metadata from the successful generator.
227 :raises RuntimeError: If all chat generators fail.
228 """
229 await self.warm_up_async()
231 messages = _normalize_messages(messages=messages)
233 failed: list[str] = []
234 last_error: BaseException | None = None
236 for idx, gen in enumerate(self.chat_generators):
237 gen_name = gen.__class__.__name__
238 try:
239 result = await self._run_single_async(gen, messages, generation_kwargs, tools, streaming_callback)
240 replies = result.get("replies", [])
241 meta = dict(result.get("meta", {}))
242 meta.update(
243 {
244 "successful_chat_generator_index": idx,
245 "successful_chat_generator_class": gen_name,
246 "total_attempts": idx + 1,
247 "failed_chat_generators": failed,
248 }
249 )
250 return {"replies": replies, "meta": meta}
251 except Exception as e: # noqa: BLE001 - fallback logic should handle any exception
252 logger.warning(
253 "ChatGenerator {chat_generator} failed with error: {error}", chat_generator=gen_name, error=e
254 )
255 failed.append(gen_name)
256 last_error = e
258 failed_names = ", ".join(failed)
259 msg = (
260 f"All {len(self.chat_generators)} chat generators failed. "
261 f"Last error: {last_error}. Failed chat generators: [{failed_names}]"
262 )
263 raise RuntimeError(msg)