Spaces:
Running
Running
| """ | |
| 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() |