fecf6decfb
* base requirements * autofix * Configure ruff for Python linting and formatting - Set up minimal ruff configuration with basic checks (E, W, F, I) - Add temporary ignores for common issues during migration - Configure pre-commit hooks to use ruff with pass_filenames - This enables gradual migration from black to ruff * Delete sdj * autofixed only * migrate lint action * more autofixed * more fixes * change precommit * try changing the hook * try this stuff
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
|