"""LLMClientPlugin model."""
from typing import List, Optional, Type
from django.db import models
from smarter.apps.account.models import (
UserProfile,
)
from smarter.apps.plugin.manifest.controller import PluginController
from smarter.apps.plugin.manifest.models.common.plugin.model import SAMPluginCommon
from smarter.apps.plugin.models import PluginMeta
from smarter.apps.plugin.plugin.base import PluginBase
from smarter.common.exceptions import SmarterValueError
from smarter.lib import logging
from smarter.lib.cache import cache_results
from smarter.lib.django.models import TimestampedModel
from smarter.lib.django.waffle import SmarterWaffleSwitches
from smarter.lib.manifest.loader import SAMLoader
from .llmclient import LLMClient
logger = logging.getSmarterLogger(__name__, any_switches=[SmarterWaffleSwitches.LLM_CLIENT_LOGGING])
[docs]
class LLMClientPlugin(TimestampedModel):
"""
Represents the association between a LLMClient instance and its enabled plugins within the Smarter platform.
This model establishes a many-to-one relationship, where each plugin entry is linked to a specific LLMClient
and references metadata describing the plugin. By maintaining this mapping, the platform can manage which
plugins are available to each llmclient, enabling extensibility and customization of llmclient capabilities.
The LLMClientPlugin model supports use cases such as plugin activation, deactivation, and enumeration for
individual llmclients. It is essential for scenarios where llmclients require additional functionality
provided by external or internal plugins, such as integrations, enhanced processing, or custom behaviors.
**Model Relationships**
- Each LLMClientPlugin is linked to one :class:`LLMClient` instance.
- Each LLMClientPlugin references one :class:`PluginMeta` instance, which contains metadata about the plugin.
**Usage Example**
.. code-block:: python
# Add a plugin to an llmclient
plugin_meta = PluginMeta.objects.get(name="weather")
llmclient_plugin = LLMClientPlugin.objects.create(llmclient=my_llmclient, plugin_meta=plugin_meta)
# List all plugins for an llmclient
plugins = LLMClientPlugin.objects.filter(llmclient=my_llmclient)
**Notes**
- Plugin management and loading are handled via the PluginController and related infrastructure.
- This model is intended for internal use to support dynamic extension of llmclient features.
- Uniqueness is enforced for each (llmclient, plugin_meta) pair to prevent duplicate plugin assignments.
"""
# pylint: disable=C0115
class Meta:
verbose_name_plural = "LLMClient Plugins"
unique_together = ("llmclient", "plugin_meta")
#: The LLMClient instance associated with this plugin.
llmclient = models.ForeignKey(LLMClient, on_delete=models.CASCADE)
#: The metadata for the plugin associated with the LLMClient.
plugin_meta = models.ForeignKey(PluginMeta, on_delete=models.CASCADE)
def __str__(self):
try:
url = self.llmclient.url if self.llmclient else "undefined llmclient"
plugin_name = self.plugin_meta.name if self.plugin_meta else "undefined plugin"
except LLMClient.DoesNotExist:
url = "undefined llmclient"
except PluginMeta.DoesNotExist:
plugin_name = "undefined plugin"
return f"{url} - {plugin_name}"
@property
def plugin(self) -> Optional[PluginBase]:
"""
Returns the Plugin instance associated with this LLMClientPlugin.
:returns: Plugin instance or None
:rtype: Optional[PluginBase]
"""
if not self.llmclient:
return None
admin_user = UserProfile.admin_for_account(self.llmclient.user_profile.cached_account)
if admin_user is None:
raise SmarterValueError("LLMClientPlugin.plugin() failed to find admin user for llmclient account")
user_profile = UserProfile.get_cached_object(invalidate=False, user=admin_user)
@cache_results()
def get_cached_plugin_controller(
account_id: int,
user_id: int,
plugin_meta_id: int,
user_profile_id: int,
class_name: str = self.__class__.__name__,
) -> PluginController:
retval = PluginController(
account=self.llmclient.user_profile.cached_account,
user=admin_user,
plugin_meta=self.plugin_meta,
user_profile=user_profile,
)
logger.debug(
"%s.get_cached_plugin_controller() fetched and cached plugin controller for llmclient_id: %s, plugin_meta_id: %s",
class_name,
self.llmclient.id,
self.plugin_meta.id,
)
return retval
plugin_controller = get_cached_plugin_controller(
account_id=self.llmclient.user_profile.cached_account.id,
user_id=admin_user.id, # type: ignore[union-attr]
plugin_meta_id=self.plugin_meta.id,
user_profile_id=user_profile.id, # type: ignore[union-attr]
class_name=self.__class__.__name__,
)
this_plugin = plugin_controller.plugin
return this_plugin
[docs]
@classmethod
def load(cls: Type["LLMClientPlugin"], llmclient: LLMClient, data) -> "LLMClientPlugin":
"""
Load (aka import) a plugin from a data file in yaml or json format.
:param llmclient: The LLMClient instance to associate with the plugin.
:param data: The plugin manifest data in yaml or json format.
:returns: The created LLMClientPlugin instance.
:rtype: LLMClientPlugin
See Also:
- :py:class:`smarter.apps.plugin.manifest.controller.PluginController`
- :py:class:`smarter.lib.manifest.loader.SAMLoader`
"""
if not llmclient:
return None
admin_user = UserProfile.admin_for_account(llmclient.user_profile.cached_account)
if admin_user is None:
raise SmarterValueError("LLMClientPlugin.plugin() failed to find admin user for llmclient account")
user_profile = UserProfile.get_cached_object(invalidate=False, user=admin_user)
loader = SAMLoader(manifest=data)
manifest = SAMPluginCommon(**loader.json_data) # type: ignore[call-arg]
plugin_controller = PluginController(user_profile=user_profile, manifest=manifest)
plugin = plugin_controller.plugin
if not plugin or plugin.plugin_meta is None:
raise SmarterValueError("LLMClientPlugin.load() failed to load plugin from data file")
return cls.objects.create(llmclient=llmclient, plugin_meta=plugin.plugin_meta)
[docs]
@classmethod
def plugins(cls, llmclient: LLMClient) -> List[PluginBase]:
"""
Returns a list of Plugin instances associated with the given LLMClient.
:param llmclient: The LLMClient instance to retrieve plugins for.
:returns: List of Plugin instances.
:rtype: List[PluginBase]
:raises SmarterValueError: If admin user for llmclient account is not found
or if a plugin fails to load.
See Also:
- :py:class:`smarter.apps.plugin.controller.PluginController`
"""
if not llmclient:
return []
llmclient_plugins = cls.objects.filter(llmclient=llmclient)
admin_user = UserProfile.admin_for_account(llmclient.user_profile.cached_account)
if admin_user is None:
raise SmarterValueError("LLMClientPlugin.plugin() failed to find admin user for llmclient account")
user_profile = UserProfile.get_cached_object(invalidate=False, user=admin_user)
retval = []
for llmclient_plugin in llmclient_plugins:
plugin_controller = PluginController(
user_profile=user_profile,
plugin_meta=llmclient_plugin.plugin_meta,
)
if not plugin_controller or not plugin_controller.plugin:
raise SmarterValueError(
f"LLMClientPlugin.plugins() failed to load plugin for {llmclient_plugin.plugin_meta.name}"
)
retval.append(plugin_controller.plugin)
return retval
# pylint: disable=W0221
[docs]
@classmethod
def get_cached_objects(
cls, invalidate: Optional[bool] = False, llmclient: Optional[LLMClient] = None
) -> models.QuerySet["LLMClientPlugin"]:
"""
Retrieve a queryset of LLMClientPlugin instances associated with a LLMClient using caching.
:param invalidate: Whether to invalidate the cache for this retrieval.
:type invalidate: bool, optional
:param llmclient: The LLMClient instance for which to retrieve plugins.
:type llmclient: LLMClient, optional
:returns: A queryset of LLMClientPlugin instances associated with the LLMClient.
:rtype: models.QuerySet["LLMClientPlugin"]
"""
logger_prefix = logging.formatted_text(__name__ + "." + LLMClientPlugin.__name__ + ".get_cached_objects()")
@cache_results()
def _get_plugins_for_llmclient_id(
llmclient_id: int, class_name: str = cls.__name__
) -> models.QuerySet["LLMClientPlugin"]:
"""
Caches the plugins for an llmclient by llmclient_id to optimize.
performance and reduce database queries.
:param llmclient_id: The ID of the LLMClient for which to retrieve plugins.
:param class_name: The name of the class for cache key purposes.
:returns: A queryset of LLMClientPlugin instances associated with the LLMClient.
:rtype: models.QuerySet["LLMClientPlugin"]
"""
logger.debug("%s called with llmclient=%s, invalidate=%s", logger_prefix, llmclient, invalidate)
retval = cls.objects.filter(llmclient_id=llmclient_id).select_related(
"plugin_meta",
"plugin_meta__user_profile",
"plugin_meta__user_profile__user",
"plugin_meta__user_profile__account",
"llmclient__user_profile",
"llmclient__user_profile__user",
"llmclient__user_profile__account",
)
logger.debug(
"%s._get_plugins_for_llmclient_id() fetched and cached %s plugins for llmclient_id: %s",
logger_prefix,
len(retval),
llmclient_id,
)
return retval
if invalidate and llmclient:
_get_plugins_for_llmclient_id.invalidate(llmclient_id=llmclient.id, class_name=cls.__name__) # type: ignore[union-attr]
if llmclient:
return _get_plugins_for_llmclient_id(llmclient_id=llmclient.id, class_name=cls.__name__) # type: ignore[return-value]
return super().get_cached_objects(invalidate=invalidate) # type: ignore[return-value]
[docs]
@classmethod
def plugins_json(cls, llmclient: LLMClient) -> List[dict]:
retval = []
for plugin in cls.plugins(llmclient):
retval.append(plugin.to_json())
return retval
__all__ = [
"LLMClientPlugin",
]