wehub-resource-sync
e4dcfc49aa
Tests / Import Check (Python 3.13) (push) Has been cancelled
Tests / Import Check (Python 3.14) (push) Has been cancelled
Tests / Python Tests (Python 3.11) (push) Has been cancelled
Tests / Python Tests (Python 3.12) (push) Has been cancelled
Tests / Python Tests (Python 3.14) (push) Has been cancelled
Tests / Test Summary (push) Has been cancelled
Tests / Lint and Format (push) Has been cancelled
Tests / Web Node Tests (push) Has been cancelled
Tests / Import Check (Python 3.11) (push) Has been cancelled
Tests / Import Check (Python 3.12) (push) Has been cancelled
Tests / Python Tests (Python 3.13) (push) Has been cancelled
205 lines
7.3 KiB
Python
205 lines
7.3 KiB
Python
"""Base LLM provider with unified configuration and retries."""
|
|
|
|
from abc import ABC
|
|
from collections.abc import Awaitable, Callable
|
|
import logging
|
|
from typing import TypeVar
|
|
|
|
import tenacity
|
|
from tenacity import AsyncRetrying, retry_if_exception, stop_after_attempt
|
|
|
|
from deeptutor.utils.error_rate_tracker import record_provider_call
|
|
from deeptutor.utils.network.circuit_breaker import (
|
|
is_call_allowed,
|
|
record_call_failure,
|
|
record_call_success,
|
|
)
|
|
|
|
from ..config import LLMConfig
|
|
from ..error_mapping import map_error
|
|
from ..exceptions import (
|
|
LLMAPIError,
|
|
LLMCircuitBreakerError,
|
|
LLMError,
|
|
LLMRateLimitError,
|
|
LLMTimeoutError,
|
|
)
|
|
from ..traffic_control import TrafficController
|
|
from ..types import AsyncStreamGenerator, TutorResponse
|
|
|
|
T = TypeVar("T")
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# Cap retry delays to avoid excessive waits during outages.
|
|
MAX_RETRY_DELAY_SECONDS = 60.0
|
|
BASE_RETRY_DELAY_SECONDS = 1.0
|
|
|
|
|
|
class BaseLLMProvider(ABC):
|
|
"""Base class for all LLM providers with unified config and retries."""
|
|
|
|
def __init__(self, config: LLMConfig) -> None:
|
|
"""Initialize provider with shared configuration and traffic control."""
|
|
self.config = config
|
|
self.provider_name = config.provider_name
|
|
self.api_key = getattr(config, "get_api_key", lambda: config.api_key)()
|
|
self.base_url = config.base_url or config.effective_url
|
|
|
|
# Isolation: Each provider gets its own traffic controller instance
|
|
self.traffic_controller: TrafficController
|
|
traffic_controller = getattr(config, "traffic_controller", None)
|
|
if isinstance(traffic_controller, TrafficController):
|
|
self.traffic_controller = traffic_controller
|
|
else:
|
|
self.traffic_controller = TrafficController(
|
|
provider_name=self.provider_name,
|
|
max_concurrency=getattr(config, "max_concurrency", 20),
|
|
requests_per_minute=getattr(config, "requests_per_minute", 600),
|
|
)
|
|
|
|
async def complete(self, prompt: str, **kwargs: object) -> TutorResponse:
|
|
"""Run a completion call for the provider."""
|
|
raise NotImplementedError
|
|
|
|
def stream(self, prompt: str, **kwargs: object) -> AsyncStreamGenerator:
|
|
"""Return an async generator for streaming completions."""
|
|
raise NotImplementedError
|
|
|
|
def _map_exception(self, e: Exception) -> LLMError:
|
|
return map_error(e, provider=self.provider_name)
|
|
|
|
def calculate_cost(self, usage: dict[str, object]) -> float:
|
|
"""Calculate cost estimate for a provider call."""
|
|
return 0.0
|
|
|
|
def _check_circuit_breaker(self) -> None:
|
|
"""Raise when the circuit breaker is open for this provider."""
|
|
if not is_call_allowed(self.provider_name):
|
|
record_provider_call(self.provider_name, success=False)
|
|
error = LLMCircuitBreakerError(
|
|
f"Circuit breaker open for provider {self.provider_name}",
|
|
provider=self.provider_name,
|
|
)
|
|
setattr(error, "is_circuit_breaker", True)
|
|
raise error
|
|
|
|
def _should_record_failure(self, error: LLMError) -> bool:
|
|
"""Return True when failures should trip the circuit breaker."""
|
|
if isinstance(error, (LLMRateLimitError, LLMTimeoutError)):
|
|
return True
|
|
if isinstance(error, LLMAPIError):
|
|
status_code = error.status_code
|
|
if status_code is None:
|
|
return True
|
|
return status_code >= 500
|
|
return False
|
|
|
|
def _should_retry_error(self, error: BaseException) -> bool:
|
|
"""Return True when an error should trigger a retry."""
|
|
if isinstance(error, (LLMRateLimitError, LLMTimeoutError)):
|
|
return True
|
|
if isinstance(error, LLMAPIError):
|
|
status_code = error.status_code
|
|
if status_code is None:
|
|
return True
|
|
return status_code >= 500
|
|
return False
|
|
|
|
def _wait_strategy(self, retry_state: tenacity.RetryCallState) -> float:
|
|
"""Return the next retry delay based on error context."""
|
|
outcome = retry_state.outcome
|
|
if outcome is None:
|
|
return BASE_RETRY_DELAY_SECONDS
|
|
exc = outcome.exception()
|
|
if exc is None:
|
|
return BASE_RETRY_DELAY_SECONDS
|
|
if isinstance(exc, LLMRateLimitError):
|
|
retry_after = getattr(exc, "retry_after", None)
|
|
retry_after_value: float | None = None
|
|
if retry_after is not None:
|
|
try:
|
|
retry_after_value = float(retry_after)
|
|
except (TypeError, ValueError):
|
|
retry_after_value = None
|
|
if retry_after_value is not None:
|
|
return max(0.0, min(retry_after_value, MAX_RETRY_DELAY_SECONDS))
|
|
|
|
wait_fn = tenacity.wait_exponential(
|
|
multiplier=1.5,
|
|
min=BASE_RETRY_DELAY_SECONDS,
|
|
max=MAX_RETRY_DELAY_SECONDS,
|
|
)
|
|
return float(wait_fn(retry_state))
|
|
|
|
async def _execute_core(
|
|
self,
|
|
func: Callable[..., Awaitable[T]],
|
|
*args: object,
|
|
**kwargs: object,
|
|
) -> T:
|
|
"""
|
|
Core execution pipeline:
|
|
1) circuit breaker check
|
|
2) traffic control context
|
|
3) call execution
|
|
4) mapping + metrics
|
|
"""
|
|
self._check_circuit_breaker()
|
|
|
|
try:
|
|
async with self.traffic_controller:
|
|
result = await func(*args, **kwargs)
|
|
record_provider_call(self.provider_name, success=True)
|
|
record_call_success(self.provider_name)
|
|
return result
|
|
except Exception as exc:
|
|
mapped_exc = self._map_exception(exc)
|
|
record_provider_call(self.provider_name, success=False)
|
|
if isinstance(mapped_exc, LLMError):
|
|
if self._should_record_failure(mapped_exc):
|
|
record_call_failure(self.provider_name)
|
|
raise mapped_exc from exc
|
|
# Internal/runtime errors should bubble up without being rewrapped.
|
|
raise mapped_exc
|
|
|
|
async def execute(
|
|
self,
|
|
func: Callable[..., Awaitable[T]],
|
|
*args: object,
|
|
**kwargs: object,
|
|
) -> T:
|
|
"""Execute a single attempt without retry."""
|
|
return await self._execute_core(func, *args, **kwargs)
|
|
|
|
async def execute_with_retry(
|
|
self,
|
|
func: Callable[..., Awaitable[T]],
|
|
*args: object,
|
|
max_retries: int = 3,
|
|
sleep: Callable[[int | float], Awaitable[None] | None] | None = None,
|
|
**kwargs: object,
|
|
) -> T:
|
|
"""Execute with automatic retries using tenacity."""
|
|
|
|
def _default_sleep(_delay: int | float) -> None:
|
|
return None
|
|
|
|
sleep_fn: Callable[[int | float], Awaitable[None] | None]
|
|
sleep_fn = _default_sleep if sleep is None else sleep
|
|
|
|
retrying = AsyncRetrying(
|
|
stop=stop_after_attempt(max_retries + 1),
|
|
wait=self._wait_strategy,
|
|
retry=retry_if_exception(self._should_retry_error),
|
|
reraise=True,
|
|
before_sleep=tenacity.before_sleep_log(logger, logging.WARNING),
|
|
sleep=sleep_fn,
|
|
)
|
|
|
|
async for attempt in retrying:
|
|
with attempt:
|
|
return await self._execute_core(func, *args, **kwargs)
|
|
|
|
raise RuntimeError("Retry loop exited without returning")
|