diff --git a/src/backend/langflow/interface/custom/__init__.py b/src/backend/langflow/interface/custom/__init__.py new file mode 100644 index 000000000..48672e52b --- /dev/null +++ b/src/backend/langflow/interface/custom/__init__.py @@ -0,0 +1,4 @@ +from langflow.interface.custom.base import CustomComponentCreator +from langflow.interface.custom.custom import CustomComponent + +__all__ = ["CustomComponentCreator", "CustomComponent"] diff --git a/src/backend/langflow/interface/custom/base.py b/src/backend/langflow/interface/custom/base.py new file mode 100644 index 000000000..359b799ce --- /dev/null +++ b/src/backend/langflow/interface/custom/base.py @@ -0,0 +1,43 @@ +from typing import Any, Dict, List, Optional, Type + +from langflow.custom.customs import get_custom_nodes +from langflow.interface.base import LangChainTypeCreator +from langflow.interface.custom.custom import CustomComponent +from langflow.template.frontend_node.custom_components import ( + CustomComponentFrontendNode, +) +from langflow.utils.logger import logger + +# Assuming necessary imports for Field, Template, and FrontendNode classes + + +class CustomComponentCreator(LangChainTypeCreator): + type_name: str = "custom_components" + + @property + def frontend_node_class(self) -> Type[CustomComponentFrontendNode]: + return CustomComponentFrontendNode + + @property + def type_to_loader_dict(self) -> Dict: + if self.type_dict is None: + self.type_dict: dict[str, Any] = { + "CustomComponent": CustomComponent, + } + return self.type_dict + + def get_signature(self, name: str) -> Optional[Dict]: + try: + if name in get_custom_nodes(self.type_name).keys(): + return get_custom_nodes(self.type_name)[name] + except ValueError as exc: + raise ValueError(f"CustomComponent {name} not found: {exc}") from exc + except AttributeError as exc: + logger.error(f"CustomComponent {name} not loaded: {exc}") + return None + + def to_list(self) -> List[str]: + return list(self.type_to_loader_dict.keys()) + + +custom_component_creator = CustomComponentCreator()