b8e9a80d93
* wait I forgot to comit locally * cp the entire core directory and then rm the .git subdir
73 lines
2.0 KiB
Python
73 lines
2.0 KiB
Python
import importlib
|
|
from typing import Protocol, runtime_checkable
|
|
|
|
from letta.settings import settings
|
|
|
|
|
|
@runtime_checkable
|
|
class SummarizerProtocol(Protocol):
|
|
"""What a summarizer must implement"""
|
|
|
|
async def summarize(self, text: str) -> str: ...
|
|
def get_name(self) -> str: ...
|
|
|
|
|
|
# Currently this supports one of each plugin type. This can be expanded in the future.
|
|
DEFAULT_PLUGINS = {
|
|
"experimental_check": {
|
|
"protocol": None,
|
|
"target": "letta.plugins.defaults:is_experimental_enabled",
|
|
},
|
|
"summarizer": {
|
|
"protocol": SummarizerProtocol,
|
|
"target": "letta.services.summarizer.summarizer:Summarizer",
|
|
},
|
|
}
|
|
|
|
|
|
def get_plugin(plugin_type: str):
|
|
"""Get a plugin instance"""
|
|
plugin_register = dict(DEFAULT_PLUGINS, **settings.plugin_register_dict)
|
|
if plugin_type in plugin_register:
|
|
impl_path = plugin_register[plugin_type]["target"]
|
|
module_path, name = impl_path.split(":")
|
|
module = importlib.import_module(module_path)
|
|
plugin = getattr(module, name)
|
|
if type(plugin).__name__ == "function":
|
|
return plugin
|
|
elif type(plugin).__name__ == "class":
|
|
if plugin_register["protocol"] and not isinstance(plugin, type(plugin_register["protocol"])):
|
|
raise TypeError(f"{plugin} does not implement {type(plugin_register['protocol']).__name__}")
|
|
return plugin()
|
|
raise TypeError("Unknown plugin type")
|
|
|
|
|
|
_experimental_checker = None
|
|
_summarizer = None
|
|
|
|
|
|
# TODO handle coroutines
|
|
# Convenience functions
|
|
def get_experimental_checker():
|
|
global _experimental_checker
|
|
if _experimental_checker is None:
|
|
_experimental_checker = get_plugin("experimental_check")
|
|
return _experimental_checker
|
|
|
|
|
|
def get_summarizer():
|
|
global _summarizer
|
|
if _summarizer is None:
|
|
_summarizer = get_plugin("summarizer")
|
|
return _summarizer
|
|
|
|
|
|
def reset_experimental_checker():
|
|
global _experimental_checker
|
|
_experimental_checker = None
|
|
|
|
|
|
def reset_summarizer():
|
|
global _summarizer
|
|
_summarizer = None
|