Coverage for /home/admin/Documents/AI/applications/lexigram-dev/lexigram/experimental/ai/lexigram-ai-llm/src/lexigram/ai/llm/model_manager/base.py: 70%

1"""Base classes for model management.""" 

2 

3from __future__ import annotations 

4 

5from abc import ABC, abstractmethod 

6from typing import Any 

7 

8from lexigram.ai.llm.http.client import ResilientHTTPClient 

9from lexigram.ai.llm.model_manager.types import ModelLoadResult 

10from lexigram.logging import ( 

11 get_logger, 

12) 

13 

14logger = get_logger(__name__) 

15 

16 

17class AbstractModelManager(ABC): 

18 """Abstract base class for LLM model managers.""" 

19 

20 def __init__(self, base_url: str, name: str = "model-manager"): 

21 self.base_url = base_url.rstrip("/") 

22 self.name = name 

23 self._client: ResilientHTTPClient | None = None 

24 

25 async def _get_client(self) -> ResilientHTTPClient: 

26 """Get or create HTTP client.""" 

27 if self._client is None: 

28 self._client = ResilientHTTPClient( 

29 base_url=self.base_url, 

30 timeout=30.0, 

31 name=self.name, 

32 ) 

33 return self._client 

34 

35 @abstractmethod 

36 async def list_models(self) -> list[dict[str, Any]]: 

37 """List available models.""" 

38 

39 @abstractmethod 

40 async def load_model(self, model_name: str, **kwargs: Any) -> ModelLoadResult: 

41 """Load a model.""" 

42 

43 @abstractmethod 

44 async def unload_model(self, model_name: str) -> bool: 

45 """Unload a model.""" 

46 

47 @abstractmethod 

48 async def switch_model(self, model_name: str, **kwargs: Any) -> ModelLoadResult: 

49 """Switch to a different model.""" 

50 

51 @abstractmethod 

52 async def get_loaded_models(self) -> list[str]: 

53 """Get currently loaded models.""" 

54 

55 async def close(self) -> None: 

56 """Close the manager.""" 

57 if self._client: 

58 await self._client.close() 

59 self._client = None