mirror of
https://github.com/HKUDS/nanobot.git
synced 2026-08-08 13:28:43 +03:00
nanobot-contributorandXubin Ren
c2c47f7a03
When the primary model (e.g. DeepSeek during peak hours) returns an empty
choices response with HTTP 200, the error carries no status code or
structured error metadata. The existing _FALLBACK_ERROR_TOKENS had no
matching token, so _should_fallback() returned False and fallback models
were never tried.
Changes:
- Add 'empty' token to _FALLBACK_ERROR_TOKENS so 'Error: API returned
empty choices.' text matches the fallback path
- Set error_kind='empty' in openai_compat_provider when returning
the empty-choices error, making the classification explicit
- Add test coverage for both text-only and error_kind matching paths
Fixes: glebov reported primary never falls back when DeepSeek returns
empty responses
315 lines
12 KiB
Python
315 lines
12 KiB
Python
"""Provider wrapper that transparently fails over to fallback models on error."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import time
|
|
from collections.abc import Awaitable, Callable
|
|
from typing import Any
|
|
|
|
from loguru import logger
|
|
|
|
from nanobot.providers.base import LLMProvider, LLMResponse
|
|
|
|
# Circuit breaker tuned to match OpenAICompatProvider's Responses API breaker.
|
|
_PRIMARY_FAILURE_THRESHOLD = 3
|
|
_PRIMARY_COOLDOWN_S = 60
|
|
_MISSING = object()
|
|
_FALLBACK_ERROR_KINDS = frozenset({
|
|
"timeout",
|
|
"connection",
|
|
"server_error",
|
|
"rate_limit",
|
|
"overloaded",
|
|
})
|
|
_NON_FALLBACK_ERROR_KINDS = frozenset({
|
|
"authentication",
|
|
"auth",
|
|
"permission",
|
|
"content_filter",
|
|
"refusal",
|
|
"context_length",
|
|
"invalid_request",
|
|
})
|
|
_FALLBACK_ERROR_TOKENS = (
|
|
"rate_limit",
|
|
"rate limit",
|
|
"too_many_requests",
|
|
"too many requests",
|
|
"overloaded",
|
|
"server_error",
|
|
"server error",
|
|
"temporarily unavailable",
|
|
"timeout",
|
|
"timed out",
|
|
"connection",
|
|
"empty", # API returned empty choices (e.g. DeepSeek peak hours), transient
|
|
"insufficient_quota",
|
|
"insufficient quota",
|
|
"quota_exceeded",
|
|
"quota exceeded",
|
|
"quota_exhausted",
|
|
"quota exhausted",
|
|
"billing_hard_limit",
|
|
"insufficient_balance",
|
|
"balance",
|
|
"out of credits",
|
|
)
|
|
|
|
|
|
class FallbackProvider(LLMProvider):
|
|
"""Wrap a primary provider and transparently failover to fallback models.
|
|
|
|
When the primary model returns a fallbackable error before content has been
|
|
streamed, the wrapper tries each fallback model in order. Streamed timeout
|
|
errors are the recovery exception: the caller may close the current stream
|
|
segment, then the wrapper continues failover with later deltas in a new
|
|
segment. Each fallback model may reside on a different provider — a factory
|
|
callable creates the underlying provider on-the-fly.
|
|
|
|
Key design:
|
|
- Failover is request-scoped (the wrapper itself is stateless between turns).
|
|
- Skipped when content was already streamed to avoid duplicate output,
|
|
except timeout recovery can resume in a new stream segment.
|
|
- Recursive failover is prevented by the factory returning plain providers.
|
|
- Primary provider is circuit-broken after repeated failures to avoid
|
|
wasting requests on a known-bad endpoint.
|
|
"""
|
|
|
|
supports_stream_recover_callback = True
|
|
|
|
def __init__(
|
|
self,
|
|
primary: LLMProvider,
|
|
fallback_presets: list[Any],
|
|
provider_factory: Callable[[Any], LLMProvider],
|
|
):
|
|
self._primary = primary
|
|
self._fallback_presets = list(fallback_presets)
|
|
self._provider_factory = provider_factory
|
|
self._has_fallbacks = bool(fallback_presets)
|
|
self._primary_failures = 0
|
|
self._primary_tripped_at: float | None = None
|
|
|
|
@property
|
|
def generation(self):
|
|
return self._primary.generation
|
|
|
|
@generation.setter
|
|
def generation(self, value):
|
|
self._primary.generation = value
|
|
|
|
def get_default_model(self) -> str:
|
|
return self._primary.get_default_model()
|
|
|
|
@property
|
|
def supports_progress_deltas(self) -> bool:
|
|
return bool(getattr(self._primary, "supports_progress_deltas", False))
|
|
|
|
def _primary_available(self) -> bool:
|
|
"""Return True if the primary provider is not currently tripped."""
|
|
if self._primary_tripped_at is None:
|
|
return True
|
|
if time.monotonic() - self._primary_tripped_at >= _PRIMARY_COOLDOWN_S:
|
|
# Half-open: allow one probe attempt.
|
|
return True
|
|
return False
|
|
|
|
async def chat(self, **kwargs: Any) -> LLMResponse:
|
|
if not self._has_fallbacks:
|
|
return await self._primary.chat(**kwargs)
|
|
return await self._try_with_fallback(
|
|
lambda p, kw: p.chat(**kw), kwargs, has_streamed=None
|
|
)
|
|
|
|
async def chat_stream(self, **kwargs: Any) -> LLMResponse:
|
|
on_stream_recover = kwargs.pop("on_stream_recover", None)
|
|
if not self._has_fallbacks:
|
|
return await self._primary.chat_stream(**kwargs)
|
|
|
|
has_streamed: list[bool] = [False]
|
|
original_delta = kwargs.get("on_content_delta")
|
|
|
|
async def _tracking_delta(text: str) -> None:
|
|
if text:
|
|
has_streamed[0] = True
|
|
if original_delta:
|
|
await original_delta(text)
|
|
|
|
kwargs["on_content_delta"] = _tracking_delta
|
|
return await self._try_with_fallback(
|
|
lambda p, kw: p.chat_stream(**kw),
|
|
kwargs,
|
|
has_streamed=has_streamed,
|
|
on_stream_recover=on_stream_recover,
|
|
)
|
|
|
|
async def _try_with_fallback(
|
|
self,
|
|
call: Callable[[LLMProvider, dict[str, Any]], Awaitable[LLMResponse]],
|
|
kwargs: dict[str, Any],
|
|
has_streamed: list[bool] | None,
|
|
on_stream_recover: Callable[[], Awaitable[None]] | None = None,
|
|
) -> LLMResponse:
|
|
primary_model = kwargs.get("model") or self._primary.get_default_model()
|
|
primary_was_attempted = False
|
|
primary_error = "unknown error"
|
|
|
|
if self._primary_available():
|
|
primary_was_attempted = True
|
|
response = await call(self._primary, kwargs)
|
|
if response.finish_reason != "error":
|
|
self._primary_failures = 0
|
|
self._primary_tripped_at = None
|
|
return response
|
|
primary_error = (response.content or primary_error)[:120]
|
|
|
|
if has_streamed is not None and has_streamed[0]:
|
|
is_timeout = (response.error_kind or "").lower() == "timeout"
|
|
if is_timeout:
|
|
logger.warning(
|
|
"Primary model '{}' stream stalled after content was emitted; "
|
|
"attempting failover anyway",
|
|
primary_model,
|
|
)
|
|
has_streamed[0] = False
|
|
if on_stream_recover:
|
|
await on_stream_recover()
|
|
else:
|
|
kwargs["on_content_delta"] = None
|
|
else:
|
|
logger.warning(
|
|
"Primary model error but content already streamed; skipping failover"
|
|
)
|
|
return response
|
|
|
|
if not self._should_fallback(response):
|
|
logger.warning(
|
|
"Primary model '{}' returned non-fallbackable error: {}",
|
|
primary_model,
|
|
(response.content or "")[:120],
|
|
)
|
|
return response
|
|
|
|
self._primary_failures += 1
|
|
if self._primary_failures >= _PRIMARY_FAILURE_THRESHOLD:
|
|
self._primary_tripped_at = time.monotonic()
|
|
logger.warning(
|
|
"Primary model '{}' circuit open after {} consecutive failures",
|
|
primary_model, self._primary_failures,
|
|
)
|
|
else:
|
|
logger.debug("Primary model '{}' circuit open; skipping", primary_model)
|
|
|
|
last_response: LLMResponse | None = None
|
|
primary_skipped = not primary_was_attempted
|
|
for idx, fallback in enumerate(self._fallback_presets):
|
|
fallback_model = fallback.model
|
|
if has_streamed is not None and has_streamed[0]:
|
|
is_timeout = (
|
|
last_response is not None
|
|
and (last_response.error_kind or "").lower() == "timeout"
|
|
)
|
|
if is_timeout and on_stream_recover:
|
|
logger.warning(
|
|
"Fallback model '{}' stream stalled after content was emitted; "
|
|
"starting a new stream segment and trying next fallback",
|
|
self._fallback_presets[idx - 1].model if idx > 0 else primary_model,
|
|
)
|
|
has_streamed[0] = False
|
|
await on_stream_recover()
|
|
else:
|
|
break
|
|
if idx == 0 and primary_skipped:
|
|
logger.info(
|
|
"Primary model '{}' circuit open, trying fallback '{}'",
|
|
primary_model, fallback_model,
|
|
)
|
|
elif idx == 0:
|
|
logger.info(
|
|
"Primary model '{}' failed: {}; trying fallback '{}'",
|
|
primary_model, primary_error, fallback_model,
|
|
)
|
|
else:
|
|
logger.info(
|
|
"Fallback '{}' also failed, trying next fallback '{}'",
|
|
self._fallback_presets[idx - 1].model, fallback_model,
|
|
)
|
|
try:
|
|
fallback_provider = self._provider_factory(fallback)
|
|
except Exception as exc:
|
|
logger.warning(
|
|
"Failed to create provider for fallback '{}': {}", fallback_model, exc
|
|
)
|
|
continue
|
|
|
|
original_values = {
|
|
name: kwargs.get(name, _MISSING)
|
|
for name in ("model", "max_tokens", "temperature", "reasoning_effort")
|
|
}
|
|
kwargs["model"] = fallback_model
|
|
kwargs["max_tokens"] = fallback.max_tokens
|
|
kwargs["temperature"] = fallback.temperature
|
|
if fallback.reasoning_effort is None:
|
|
kwargs.pop("reasoning_effort", None)
|
|
else:
|
|
kwargs["reasoning_effort"] = fallback.reasoning_effort
|
|
try:
|
|
fallback_response = await call(fallback_provider, kwargs)
|
|
finally:
|
|
for name, value in original_values.items():
|
|
if value is _MISSING:
|
|
kwargs.pop(name, None)
|
|
else:
|
|
kwargs[name] = value
|
|
|
|
if fallback_response.finish_reason != "error":
|
|
logger.info(
|
|
"Fallback '{}' succeeded after primary '{}' failed",
|
|
fallback_model, primary_model,
|
|
)
|
|
return fallback_response
|
|
|
|
last_response = fallback_response
|
|
logger.warning(
|
|
"Fallback '{}' also failed: {}",
|
|
fallback_model,
|
|
(fallback_response.content or "")[:120],
|
|
)
|
|
|
|
logger.warning(
|
|
"All {} fallback model(s) failed",
|
|
len(self._fallback_presets),
|
|
)
|
|
# Return the last error response we saw (primary or last fallback).
|
|
if last_response is not None:
|
|
return last_response
|
|
# Primary was tripped and we have no fallbacks — synthesize an error.
|
|
return LLMResponse(
|
|
content=f"Primary model '{primary_model}' circuit open and no fallbacks available",
|
|
finish_reason="error",
|
|
)
|
|
|
|
@staticmethod
|
|
def _should_fallback(response: LLMResponse) -> bool:
|
|
if response.error_should_retry is False:
|
|
return False
|
|
status = response.error_status_code
|
|
kind = (response.error_kind or "").lower()
|
|
error_type = (response.error_type or "").lower()
|
|
code = (response.error_code or "").lower()
|
|
text = (response.content or "").lower()
|
|
|
|
if status in {400, 401, 403, 404, 422}:
|
|
return False
|
|
if kind in _NON_FALLBACK_ERROR_KINDS:
|
|
return False
|
|
if any(token in value for value in (kind, error_type, code) for token in _NON_FALLBACK_ERROR_KINDS):
|
|
return False
|
|
if response.error_should_retry is True:
|
|
return True
|
|
if status is not None and (status in {408, 409, 429} or 500 <= status <= 599):
|
|
return True
|
|
if kind in _FALLBACK_ERROR_KINDS:
|
|
return True
|
|
return any(token in value for value in (kind, error_type, code, text) for token in _FALLBACK_ERROR_TOKENS)
|