refac: Factory implementation of LangChainTypes

src/backend/langflow/interface/agents.py

@@ -0,0 +1,28 @@
+from langchain.agents import loading
+from langflow.interface.base import LangChainTypeCreator
+from langflow.utils.util import build_template_from_class
+from langflow.settings import settings
+from typing import Dict, List
+
+
+class AgentCreator(LangChainTypeCreator):
+    type_name: str = "agents"
+
+    @property
+    def type_to_loader_dict(self) -> Dict:
+        return loading.AGENT_TO_CLASS
+
+    def get_signature(self, name: str) -> Dict | None:
+        try:
+            return build_template_from_class(
+                name, self.type_to_loader_dict, add_function=True
+            )
+        except ValueError as exc:
+            raise ValueError("Agent not found") from exc
+
+    def to_list(self) -> List[str]:
+        return [
+            agent.__name__
+            for agent in self.type_to_loader_dict.values()
+            if agent.__name__ in settings.agents or settings.dev
+        ]

src/backend/langflow/interface/base.py

@@ -0,0 +1,56 @@
+from typing import Dict, List
+from pydantic import BaseModel
+from abc import ABC, abstractmethod
+from langflow.template.template import Template, Field, FrontendNode
+
+
+# Assuming necessary imports for Field, Template, and FrontendNode classes
+
+
+class LangChainTypeCreator(BaseModel, ABC):
+    type_name: str
+
+    @property
+    @abstractmethod
+    def type_to_loader_dict(self) -> Dict:
+        pass
+
+    @abstractmethod
+    def get_signature(self, name: str) -> Dict:
+        pass
+
+    @abstractmethod
+    def to_list(self) -> List[str]:
+        pass
+
+    def to_dict(self):
+        result = {self.type_name: {}}  # type: Dict
+
+        for name in self.to_list():
+            result[self.type_name][name] = self.get_signature(name)
+
+        return result
+
+    def frontend_node(self, name) -> FrontendNode:
+        signature = self.get_signature(name)
+        fields = [
+            Field(
+                name=key,
+                field_type=value["type"],
+                required=value.get("required", False),
+                placeholder=value.get("placeholder", ""),
+                is_list=value.get("list", False),
+                show=value.get("show", True),
+                multiline=value.get("multiline", False),
+                value=value.get("value", None),
+            )
+            for key, value in signature["template"].items()
+            if key != "_type"
+        ]
+        template = Template(type_name=name, fields=fields)
+        return FrontendNode(
+            template=template,
+            description=signature["description"],
+            base_classes=signature["base_classes"],
+            name=name,
+        )

src/backend/langflow/interface/chains.py

@@ -0,0 +1,35 @@
+from typing import Dict, List
+from langflow.interface.base import LangChainTypeCreator
+from langflow.interface.signature import get_chain_signature
+from langflow.template.template import Field, FrontendNode, Template
+from langflow.utils.util import build_template_from_function
+from langflow.settings import settings
+from langchain.chains import loading as chains_loading
+
+# Assuming necessary imports for Field, Template, and FrontendNode classes
+
+
+class ChainCreator(LangChainTypeCreator):
+    type_name: str = "chains"
+
+    @property
+    def type_to_loader_dict(self) -> Dict:
+        return chains_loading.type_to_loader_dict
+
+    def get_signature(self, name: str) -> Dict | None:
+        try:
+            return build_template_from_function(
+                name, self.type_to_loader_dict, add_function=True
+            )
+        except ValueError as exc:
+            raise ValueError("Chain not found") from exc
+
+    def to_list(self) -> List[str]:
+        return [
+            chain.__annotations__["return"].__name__
+            for chain in self.type_to_loader_dict.values()
+            if (
+                chain.__annotations__["return"].__name__ in settings.chains
+                or settings.dev
+            )
+        ]

src/backend/langflow/interface/llms.py

@@ -0,0 +1,27 @@
+from langflow.interface.custom_lists import llm_type_to_cls_dict
+from langflow.settings import settings
+from langflow.interface.base import LangChainTypeCreator
+from langflow.utils.util import build_template_from_class
+from typing import Dict, List
+
+
+class LLMCreator(LangChainTypeCreator):
+    type_name: str = "llms"
+
+    @property
+    def type_to_loader_dict(self) -> Dict:
+        return llm_type_to_cls_dict
+
+    def get_signature(self, name: str) -> Dict | None:
+        """Get the signature of an llm."""
+        try:
+            return build_template_from_class(name, llm_type_to_cls_dict)
+        except ValueError as exc:
+            raise ValueError("LLM not found") from exc
+
+    def to_list(self) -> List[str]:
+        return [
+            llm.__name__
+            for llm in self.type_to_loader_dict.values()
+            if llm.__name__ in settings.llms or settings.dev
+        ]

src/backend/langflow/interface/memories.py

@@ -0,0 +1,27 @@
+from langflow.interface.custom_lists import memory_type_to_cls_dict
+from langflow.settings import settings
+from langflow.interface.base import LangChainTypeCreator
+from langflow.utils.util import build_template_from_class
+from typing import Dict, List
+
+
+class MemoryCreator(LangChainTypeCreator):
+    type_name: str = "memories"
+
+    @property
+    def type_to_loader_dict(self) -> Dict:
+        return memory_type_to_cls_dict
+
+    def get_signature(self, name: str) -> Dict | None:
+        """Get the signature of a memory."""
+        try:
+            return build_template_from_class(name, memory_type_to_cls_dict)
+        except ValueError as exc:
+            raise ValueError("Memory not found") from exc
+
+    def to_list(self) -> List[str]:
+        return [
+            memory.__name__
+            for memory in self.type_to_loader_dict.values()
+            if memory.__name__ in settings.memories or settings.dev
+        ]

src/backend/langflow/interface/prompts.py

@@ -0,0 +1,32 @@
+from langchain.prompts import loading
+from langflow.interface.base import LangChainTypeCreator
+from langflow.utils.util import build_template_from_function
+from langflow.settings import settings
+from langflow.custom.customs import get_custom_nodes
+from typing import Dict, List
+
+
+class PromptCreator(LangChainTypeCreator):
+    type_name: str = "prompts"
+
+    @property
+    def type_to_loader_dict(self) -> Dict:
+        return loading.type_to_loader_dict
+
+    def get_signature(self, name: str) -> Dict | None:
+        try:
+            if name in get_custom_nodes("prompts").keys():
+                return get_custom_nodes("prompts")[name]
+            return build_template_from_function(name, self.type_to_loader_dict)
+        except ValueError as exc:
+            raise ValueError("Prompt not found") from exc
+
+    def to_list(self) -> List[str]:
+        custom_prompts = get_custom_nodes("prompts")
+        library_prompts = [
+            prompt.__annotations__["return"].__name__
+            for prompt in self.type_to_loader_dict.values()
+            if prompt.__annotations__["return"].__name__ in settings.prompts
+            or settings.dev
+        ]
+        return library_prompts + list(custom_prompts.keys())

src/backend/langflow/interface/tools.py

@@ -0,0 +1,128 @@
+from langflow.custom import customs
+from langflow.interface.listing import ALL_TOOLS_NAMES, CUSTOM_TOOLS
+from langflow.template.template import Field, Template
+from langflow.utils import util
+from langflow.settings import settings
+from langflow.interface.base import LangChainTypeCreator
+from typing import Dict, List
+from langchain.agents.load_tools import (
+    _BASE_TOOLS,
+    _EXTRA_LLM_TOOLS,
+    _EXTRA_OPTIONAL_TOOLS,
+    _LLM_TOOLS,
+)
+
+
+class ToolCreator(LangChainTypeCreator):
+    type_name: str = "tools"
+
+    @property
+    def type_to_loader_dict(self) -> Dict:
+        return ALL_TOOLS_NAMES
+
+    def get_signature(self, name: str) -> Dict | None:
+        """Get the signature of a tool."""
+
+        NODE_INPUTS = ["llm", "func"]
+        base_classes = ["Tool"]
+        all_tools = {}
+        for tool in ALL_TOOLS_NAMES:
+            if tool_params := util.get_tool_params(util.get_tool_by_name(tool)):
+                tool_name = tool_params.get("name") or str(tool)
+                all_tools[tool_name] = {"type": tool, "params": tool_params}
+
+        # Raise error if name is not in tools
+        if name not in all_tools.keys():
+            raise ValueError("Tool not found")
+
+        type_dict = {
+            "str": Field(
+                field_type="str",
+                required=True,
+                is_list=False,
+                show=True,
+                placeholder="",
+                value="",
+            ),
+            "llm": Field(field_type="BaseLLM", required=True, is_list=False, show=True),
+            "func": Field(
+                field_type="function",
+                required=True,
+                is_list=False,
+                show=True,
+                multiline=True,
+            ),
+            "code": Field(
+                field_type="str",
+                required=True,
+                is_list=False,
+                show=True,
+                value="",
+                multiline=True,
+            ),
+        }
+
+        tool_type: str = all_tools[name]["type"]  # type: ignore
+
+        if tool_type in _BASE_TOOLS:
+            params = []
+        elif tool_type in _LLM_TOOLS:
+            params = ["llm"]
+        elif tool_type in _EXTRA_LLM_TOOLS:
+            _, extra_keys = _EXTRA_LLM_TOOLS[tool_type]
+            params = ["llm"] + extra_keys
+        elif tool_type in _EXTRA_OPTIONAL_TOOLS:
+            _, extra_keys = _EXTRA_OPTIONAL_TOOLS[tool_type]
+            params = extra_keys
+        elif tool_type == "Tool":
+            params = ["name", "description", "func"]
+        elif tool_type in CUSTOM_TOOLS:
+            # Get custom tool params
+            params = all_tools[name]["params"]  # type: ignore
+            base_classes = ["function"]
+            if node := customs.get_custom_nodes("tools").get(tool_type):
+                return node
+
+        else:
+            params = []
+
+        # Copy the field and add the name
+        fields = []
+        for param in params:
+            if param in NODE_INPUTS:
+                field = type_dict[param].copy()
+            else:
+                field = type_dict["str"].copy()
+            field.name = param
+            if param == "aiosession":
+                field.show = False
+                field.required = False
+            fields.append(field)
+
+        template = Template(fields=fields, type_name=tool_type)
+
+        tool_params = util.get_tool_params(util.get_tool_by_name(tool_type))
+        if tool_params is None:
+            tool_params = {}
+        return {
+            "template": util.format_dict(template.to_dict()),
+            **tool_params,
+            "base_classes": base_classes,
+        }
+
+    def to_list(self) -> List[str]:
+        """List all load tools"""
+
+        tools = []
+
+        for tool in ALL_TOOLS_NAMES:
+            tool_params = util.get_tool_params(util.get_tool_by_name(tool))
+            if tool_params and (
+                tool_params.get("name") in settings.tools
+                or (tool_params.get("name") and settings.dev)
+            ):
+                tools.append(tool_params["name"])
+
+        # Add Tool
+        custom_tools = customs.get_custom_nodes("tools")
+        return tools + list(custom_tools.keys())

src/backend/langflow/interface/types.py

@@ -1,5 +1,12 @@
+from langflow.interface.agents import AgentCreator
 from langflow.interface.listing import list_type
+from langflow.interface.llms import LLMCreator
+from langflow.interface.memories import MemoryCreator
+from langflow.interface.prompts import PromptCreator
 from langflow.interface.signature import get_signature
+from langchain import chains
+from langflow.interface.chains import ChainCreator
+from langflow.interface.tools import ToolCreator
 
 
 def get_type_list():
@@ -16,21 +23,25 @@ def get_type_list():
 
 def build_langchain_types_dict():
     """Build a dictionary of all langchain types"""
+    chain_creator = ChainCreator()
+    agent_creator = AgentCreator()
+    prompt_creator = PromptCreator()
+    tool_creator = ToolCreator()
+    llm_creator = LLMCreator()
+    memory_creator = MemoryCreator()
 
-    return {
-        "chains": {
-            chain: get_signature(chain, "chains") for chain in list_type("chains")
-        },
-        "agents": {
-            agent: get_signature(agent, "agents") for agent in list_type("agents")
-        },
-        "prompts": {
-            prompt: get_signature(prompt, "prompts") for prompt in list_type("prompts")
-        },
-        "llms": {llm: get_signature(llm, "llms") for llm in list_type("llms")},
-        "memories": {
-            memory: get_signature(memory, "memories")
-            for memory in list_type("memories")
-        },
-        "tools": {tool: get_signature(tool, "tools") for tool in list_type("tools")},
-    }
+    all_types = {}
+
+    creators = [
+        chain_creator,
+        agent_creator,
+        prompt_creator,
+        llm_creator,
+        memory_creator,
+        tool_creator,
+    ]
+
+    all_types = {}
+    for creator in creators:
+        all_types.update(creator.to_dict())
+    return all_types