Coverage for src/lilbee/providers/backends/liter_llm/client.py: 100%

1"""Import and construct the liter-llm client.""" 

2 

3from __future__ import annotations 

4 

5import functools 

6from typing import Any 

7 

8from lilbee.providers.availability import PROVIDER_SDK_MISSING_MESSAGE 

9from lilbee.providers.backends.liter_llm.errors import _PROVIDER_NAME 

10from lilbee.providers.base import ProviderError 

11from lilbee.providers.local_servers import local_server_for_key 

12 

13# Seconds a single provider request may run. liter-llm has no unbounded default, 

14# so a long non-streaming generation needs an explicit ceiling. 

15_REQUEST_TIMEOUT_S = 600 

16# Retries after a failed request. A local server answers or is down, so a retry only 

17# repeats a long load; a hosted provider gets two (three attempts). 

18_LOCAL_SERVER_RETRIES = 0 

19_HOSTED_RETRIES = 2 

20 

21 

22def _require_sdk() -> Any: 

23 """Import ``liter_llm`` or raise a user-facing ProviderError with install steps.""" 

24 try: 

25 import liter_llm 

26 except ImportError as exc: 

27 raise ProviderError(PROVIDER_SDK_MISSING_MESSAGE, provider=_PROVIDER_NAME) from exc 

28 return liter_llm 

29 

30 

31def _max_retries(model: str) -> int: 

32 """Return the retry count for a ``provider/model`` string: none for a local server.""" 

33 provider = model.partition("/")[0] 

34 return _LOCAL_SERVER_RETRIES if local_server_for_key(provider) is not None else _HOSTED_RETRIES 

35 

36 

37@functools.lru_cache(maxsize=32) 

38def _client(model: str, base_url: str | None, api_key: str | None) -> Any: 

39 """Return the liter-llm client for one model, endpoint and key (cached).""" 

40 liter_llm = _require_sdk() 

41 options = liter_llm.ClientOptions( 

42 api_key=api_key or "", 

43 base_url=base_url, 

44 timeout_secs=_REQUEST_TIMEOUT_S, 

45 max_retries=_max_retries(model), 

46 model_hint=model, 

47 ) 

48 return liter_llm.create_client_with_options(options)