Spaces:
Sleeping
Sleeping
File size: 4,300 Bytes
f871fed |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 |
"""
Models service layer using API.
"""
from typing import List, Optional
from loguru import logger
from api.client import api_client
from open_notebook.domain.models import DefaultModels, Model
class ModelsService:
"""Service layer for models operations using API."""
def __init__(self):
logger.info("Using API for models operations")
def get_all_models(self, model_type: Optional[str] = None) -> List[Model]:
"""Get all models with optional type filtering."""
models_data = api_client.get_models(model_type=model_type)
# Convert API response to Model objects
models = []
for model_data in models_data:
model = Model(
name=model_data["name"],
provider=model_data["provider"],
type=model_data["type"],
)
model.id = model_data["id"]
model.created = model_data["created"]
model.updated = model_data["updated"]
models.append(model)
return models
def create_model(self, name: str, provider: str, model_type: str) -> Model:
"""Create a new model."""
response = api_client.create_model(name, provider, model_type)
model_data = response if isinstance(response, dict) else response[0]
model = Model(
name=model_data["name"],
provider=model_data["provider"],
type=model_data["type"],
)
model.id = model_data["id"]
model.created = model_data["created"]
model.updated = model_data["updated"]
return model
def delete_model(self, model_id: str) -> bool:
"""Delete a model."""
api_client.delete_model(model_id)
return True
def get_default_models(self) -> DefaultModels:
"""Get default model assignments."""
response = api_client.get_default_models()
defaults_data = response if isinstance(response, dict) else response[0]
defaults = DefaultModels()
# Set the values from API response
defaults.default_chat_model = defaults_data.get("default_chat_model")
defaults.default_transformation_model = defaults_data.get("default_transformation_model")
defaults.large_context_model = defaults_data.get("large_context_model")
defaults.default_text_to_speech_model = defaults_data.get("default_text_to_speech_model")
defaults.default_speech_to_text_model = defaults_data.get("default_speech_to_text_model")
defaults.default_embedding_model = defaults_data.get("default_embedding_model")
defaults.default_tools_model = defaults_data.get("default_tools_model")
return defaults
def update_default_models(self, defaults: DefaultModels) -> DefaultModels:
"""Update default model assignments."""
updates = {
"default_chat_model": defaults.default_chat_model,
"default_transformation_model": defaults.default_transformation_model,
"large_context_model": defaults.large_context_model,
"default_text_to_speech_model": defaults.default_text_to_speech_model,
"default_speech_to_text_model": defaults.default_speech_to_text_model,
"default_embedding_model": defaults.default_embedding_model,
"default_tools_model": defaults.default_tools_model,
}
response = api_client.update_default_models(**updates)
defaults_data = response if isinstance(response, dict) else response[0]
# Update the defaults object with the response
defaults.default_chat_model = defaults_data.get("default_chat_model")
defaults.default_transformation_model = defaults_data.get("default_transformation_model")
defaults.large_context_model = defaults_data.get("large_context_model")
defaults.default_text_to_speech_model = defaults_data.get("default_text_to_speech_model")
defaults.default_speech_to_text_model = defaults_data.get("default_speech_to_text_model")
defaults.default_embedding_model = defaults_data.get("default_embedding_model")
defaults.default_tools_model = defaults_data.get("default_tools_model")
return defaults
# Global service instance
models_service = ModelsService() |