diff --git a/src/backend/langflow/custom/customs.py b/src/backend/langflow/custom/customs.py index 2b2ccc43f..ee266b0ee 100644 --- a/src/backend/langflow/custom/customs.py +++ b/src/backend/langflow/custom/customs.py @@ -1,24 +1,27 @@ -from langflow.template import nodes +from langflow.template import frontend_node # These should always be instantiated CUSTOM_NODES = { - "prompts": {"ZeroShotPrompt": nodes.ZeroShotPromptNode()}, - "tools": {"PythonFunction": nodes.PythonFunctionNode(), "Tool": nodes.ToolNode()}, + "prompts": {"ZeroShotPrompt": frontend_node.prompts.ZeroShotPromptNode()}, + "tools": { + "PythonFunction": frontend_node.tools.PythonFunctionNode(), + "Tool": frontend_node.tools.ToolNode(), + }, "agents": { - "JsonAgent": nodes.JsonAgentNode(), - "CSVAgent": nodes.CSVAgentNode(), - "initialize_agent": nodes.InitializeAgentNode(), - "VectorStoreAgent": nodes.VectorStoreAgentNode(), - "VectorStoreRouterAgent": nodes.VectorStoreRouterAgentNode(), - "SQLAgent": nodes.SQLAgentNode(), + "JsonAgent": frontend_node.agents.JsonAgentNode(), + "CSVAgent": frontend_node.agents.CSVAgentNode(), + "initialize_agent": frontend_node.agents.InitializeAgentNode(), + "VectorStoreAgent": frontend_node.agents.VectorStoreAgentNode(), + "VectorStoreRouterAgent": frontend_node.agents.VectorStoreRouterAgentNode(), + "SQLAgent": frontend_node.agents.SQLAgentNode(), }, "utilities": { - "SQLDatabase": nodes.SQLDatabaseNode(), + "SQLDatabase": frontend_node.agents.SQLDatabaseNode(), }, "chains": { - "SeriesCharacterChain": nodes.SeriesCharacterChainNode(), - "TimeTravelGuideChain": nodes.TimeTravelGuideChainNode(), - "MidJourneyPromptChain": nodes.MidJourneyPromptChainNode(), + "SeriesCharacterChain": frontend_node.chains.SeriesCharacterChainNode(), + "TimeTravelGuideChain": frontend_node.chains.TimeTravelGuideChainNode(), + "MidJourneyPromptChain": frontend_node.chains.MidJourneyPromptChainNode(), }, } diff --git a/src/backend/langflow/interface/base.py b/src/backend/langflow/interface/base.py index d51d81788..48d6dccc4 100644 --- a/src/backend/langflow/interface/base.py +++ b/src/backend/langflow/interface/base.py @@ -3,7 +3,9 @@ from typing import Any, Dict, List, Optional, Type, Union from pydantic import BaseModel -from langflow.template.base import FrontendNode, Template, TemplateField +from langflow.template.field.base import TemplateField +from langflow.template.frontend_node.base import FrontendNode +from langflow.template.template.base import Template from langflow.utils.logger import logger # Assuming necessary imports for Field, Template, and FrontendNode classes diff --git a/src/backend/langflow/interface/chains/base.py b/src/backend/langflow/interface/chains/base.py index 7b8da370b..ad66f07b2 100644 --- a/src/backend/langflow/interface/chains/base.py +++ b/src/backend/langflow/interface/chains/base.py @@ -4,7 +4,7 @@ from langflow.custom.customs import get_custom_nodes from langflow.interface.base import LangChainTypeCreator from langflow.interface.custom_lists import chain_type_to_cls_dict from langflow.settings import settings -from langflow.template.nodes import ChainFrontendNode +from langflow.template.frontend_node.chains import ChainFrontendNode from langflow.utils.logger import logger from langflow.utils.util import build_template_from_class diff --git a/src/backend/langflow/interface/embeddings/base.py b/src/backend/langflow/interface/embeddings/base.py index 933d8ad39..1dfa05a99 100644 --- a/src/backend/langflow/interface/embeddings/base.py +++ b/src/backend/langflow/interface/embeddings/base.py @@ -3,8 +3,8 @@ from typing import Dict, List, Optional, Type from langflow.interface.base import LangChainTypeCreator from langflow.interface.custom_lists import embedding_type_to_cls_dict from langflow.settings import settings -from langflow.template.base import FrontendNode -from langflow.template.nodes import EmbeddingFrontendNode +from langflow.template.frontend_node.base import FrontendNode +from langflow.template.frontend_node.embeddings import EmbeddingFrontendNode from langflow.utils.logger import logger from langflow.utils.util import build_template_from_class diff --git a/src/backend/langflow/interface/llms/base.py b/src/backend/langflow/interface/llms/base.py index 04a36eb2d..66e153880 100644 --- a/src/backend/langflow/interface/llms/base.py +++ b/src/backend/langflow/interface/llms/base.py @@ -3,7 +3,7 @@ from typing import Dict, List, Optional, Type from langflow.interface.base import LangChainTypeCreator from langflow.interface.custom_lists import llm_type_to_cls_dict from langflow.settings import settings -from langflow.template.nodes import LLMFrontendNode +from langflow.template.frontend_node.llms import LLMFrontendNode from langflow.utils.logger import logger from langflow.utils.util import build_template_from_class diff --git a/src/backend/langflow/interface/memories/base.py b/src/backend/langflow/interface/memories/base.py index f26b09351..f0d8f88f5 100644 --- a/src/backend/langflow/interface/memories/base.py +++ b/src/backend/langflow/interface/memories/base.py @@ -3,8 +3,8 @@ from typing import Dict, List, Optional, Type from langflow.interface.base import LangChainTypeCreator from langflow.interface.custom_lists import memory_type_to_cls_dict from langflow.settings import settings -from langflow.template.base import FrontendNode -from langflow.template.nodes import MemoryFrontendNode +from langflow.template.frontend_node.base import FrontendNode +from langflow.template.frontend_node.memories import MemoryFrontendNode from langflow.utils.logger import logger from langflow.utils.util import build_template_from_class diff --git a/src/backend/langflow/interface/prompts/base.py b/src/backend/langflow/interface/prompts/base.py index 5f83a5412..39bd94c5b 100644 --- a/src/backend/langflow/interface/prompts/base.py +++ b/src/backend/langflow/interface/prompts/base.py @@ -6,7 +6,7 @@ from langflow.custom.customs import get_custom_nodes from langflow.interface.base import LangChainTypeCreator from langflow.interface.importing.utils import import_class from langflow.settings import settings -from langflow.template.nodes import PromptFrontendNode +from langflow.template.frontend_node.prompts import PromptFrontendNode from langflow.utils.logger import logger from langflow.utils.util import build_template_from_class diff --git a/src/backend/langflow/interface/tools/base.py b/src/backend/langflow/interface/tools/base.py index 888accb29..a8e7045c0 100644 --- a/src/backend/langflow/interface/tools/base.py +++ b/src/backend/langflow/interface/tools/base.py @@ -16,7 +16,8 @@ from langflow.interface.tools.constants import ( ) from langflow.interface.tools.util import get_tool_params from langflow.settings import settings -from langflow.template.base import Template, TemplateField +from langflow.template.field.base import TemplateField +from langflow.template.template.base import Template from langflow.utils import util from langflow.utils.util import build_template_from_class diff --git a/src/backend/langflow/interface/vector_store/base.py b/src/backend/langflow/interface/vector_store/base.py index d30563d7b..7ec1e0f5b 100644 --- a/src/backend/langflow/interface/vector_store/base.py +++ b/src/backend/langflow/interface/vector_store/base.py @@ -5,7 +5,7 @@ from langchain import vectorstores from langflow.interface.base import LangChainTypeCreator from langflow.interface.importing.utils import import_class from langflow.settings import settings -from langflow.template.nodes import VectorStoreFrontendNode +from langflow.template.frontend_node.vectorstores import VectorStoreFrontendNode from langflow.utils.logger import logger from langflow.utils.util import build_template_from_method diff --git a/src/backend/langflow/template/base.py b/src/backend/langflow/template/base.py index 76b544974..8b1378917 100644 --- a/src/backend/langflow/template/base.py +++ b/src/backend/langflow/template/base.py @@ -1,255 +1 @@ -import re -from abc import ABC -from typing import Any, Callable, List, Optional, Union -from pydantic import BaseModel - -from langflow.template.constants import FORCE_SHOW_FIELDS -from langflow.utils import constants - - -class TemplateFieldCreator(BaseModel, ABC): - field_type: str = "str" - required: bool = False - placeholder: str = "" - is_list: bool = False - show: bool = True - multiline: bool = False - value: Any = None - suffixes: list[str] = [] - fileTypes: list[str] = [] - file_types: list[str] = [] - content: Union[str, None] = None - password: bool = False - options: list[str] = [] - name: str = "" - display_name: Optional[str] = None - advanced: bool = False - - def to_dict(self): - result = self.dict() - # Remove key if it is None - for key in list(result.keys()): - if result[key] is None or result[key] == []: - del result[key] - result["type"] = result.pop("field_type") - result["list"] = result.pop("is_list") - - if result.get("file_types"): - result["fileTypes"] = result.pop("file_types") - - if self.field_type == "file": - result["content"] = self.content - return result - - -class TemplateField(TemplateFieldCreator): - pass - - -class Template(BaseModel): - type_name: str - fields: list[TemplateField] - - def process_fields( - self, - name: Optional[str] = None, - format_field_func: Union[Callable, None] = None, - ): - if format_field_func: - for field in self.fields: - format_field_func(field, name) - - def to_dict(self, format_field_func=None): - self.process_fields(self.type_name, format_field_func) - result = {field.name: field.to_dict() for field in self.fields} - result["_type"] = self.type_name # type: ignore - return result - - -class FrontendNode(BaseModel): - template: Template - description: str - base_classes: List[str] - name: str = "" - - def to_dict(self) -> dict: - return { - self.name: { - "template": self.template.to_dict(self.format_field), - "description": self.description, - "base_classes": self.base_classes, - } - } - - @staticmethod - def format_field(field: TemplateField, name: Optional[str] = None) -> None: - """Formats a given field based on its attributes and value.""" - SPECIAL_FIELD_HANDLERS = { - "allowed_tools": lambda field: "Tool", - "max_value_length": lambda field: "int", - } - - key = field.name - value = field.to_dict() - _type = value["type"] - - _type = FrontendNode.remove_optional(_type) - _type, is_list = FrontendNode.check_for_list_type(_type) - field.is_list = is_list or field.is_list - _type = FrontendNode.replace_mapping_with_dict(_type) - _type = FrontendNode.handle_union_type(_type) - - field.field_type = FrontendNode.handle_special_field( - field, key, _type, SPECIAL_FIELD_HANDLERS - ) - field.field_type = FrontendNode.handle_dict_type(field, _type) - field.show = FrontendNode.should_show_field(key, field.required) - field.password = FrontendNode.should_be_password(key, field.show) - field.multiline = FrontendNode.should_be_multiline(key) - - FrontendNode.replace_default_value(field, value) - FrontendNode.handle_specific_field_values(field, key, name) - FrontendNode.handle_kwargs_field(field) - FrontendNode.handle_api_key_field(field, key) - - @staticmethod - def remove_optional(_type: str) -> str: - """Removes 'Optional' wrapper from the type if present.""" - return re.sub(r"Optional\[(.*)\]", r"\1", _type) - - @staticmethod - def check_for_list_type(_type: str) -> tuple: - """Checks for list type and returns the modified type and a boolean indicating if it's a list.""" - is_list = "List" in _type or "Sequence" in _type - if is_list: - _type = re.sub(r"(List|Sequence)\[(.*)\]", r"\2", _type) - return _type, is_list - - @staticmethod - def replace_mapping_with_dict(_type: str) -> str: - """Replaces 'Mapping' with 'dict'.""" - return _type.replace("Mapping", "dict") - - @staticmethod - def handle_union_type(_type: str) -> str: - """Simplifies the 'Union' type to the first type in the Union.""" - if "Union" in _type: - _type = _type.replace("Union[", "")[:-1] - _type = _type.split(",")[0] - _type = _type.replace("]", "").replace("[", "") - return _type - - @staticmethod - def handle_special_field( - field, key: str, _type: str, SPECIAL_FIELD_HANDLERS - ) -> str: - """Handles special field by using the respective handler if present.""" - handler = SPECIAL_FIELD_HANDLERS.get(key) - return handler(field) if handler else _type - - @staticmethod - def handle_dict_type(field: TemplateField, _type: str) -> str: - """Handles 'dict' type by replacing it with 'code' or 'file' based on the field name.""" - if "dict" in _type.lower(): - if field.name == "dict_": - field.field_type = "file" - field.suffixes = [".json", ".yaml", ".yml"] - field.file_types = ["json", "yaml", "yml"] - else: - field.field_type = "code" - return _type - - @staticmethod - def replace_default_value(field: TemplateField, value: dict) -> None: - """Replaces default value with actual value if 'default' is present in value.""" - if "default" in value: - field.value = value["default"] - - @staticmethod - def handle_specific_field_values( - field: TemplateField, key: str, name: Optional[str] = None - ) -> None: - """Handles specific field values for certain fields.""" - if key == "headers": - field.value = """{'Authorization': - 'Bearer '}""" - if name == "OpenAI" and key == "model_name": - field.options = constants.OPENAI_MODELS - field.is_list = True - elif name == "ChatOpenAI" and key == "model_name": - field.options = constants.CHAT_OPENAI_MODELS - field.is_list = True - if "api_key" in key and "OpenAI" in str(name): - field.display_name = "OpenAI API Key" - field.required = False - if field.value is None: - field.value = "" - - @staticmethod - def handle_kwargs_field(field: TemplateField) -> None: - """Handles kwargs field by setting certain attributes.""" - if "kwargs" in field.name.lower(): - field.advanced = True - field.required = False - field.show = False - - @staticmethod - def handle_api_key_field(field: TemplateField, key: str) -> None: - """Handles api key field by setting certain attributes.""" - if "api" in key.lower() and "key" in key.lower(): - field.required = False - field.advanced = False - - @staticmethod - def should_show_field(key: str, required: bool) -> bool: - """Determines whether the field should be shown.""" - return ( - (required and key not in ["input_variables"]) - or key in FORCE_SHOW_FIELDS - or "api" in key - or ("key" in key and "input" not in key and "output" not in key) - ) - - @staticmethod - def should_be_password(key: str, show: bool) -> bool: - """Determines whether the field should be a password field.""" - return ( - any(text in key.lower() for text in {"password", "token", "api", "key"}) - and show - ) - - @staticmethod - def should_be_multiline(key: str) -> bool: - """Determines whether the field should be multiline.""" - return key in { - "suffix", - "prefix", - "template", - "examples", - "code", - "headers", - "description", - } - - @staticmethod - def replace_dict_with_code_or_file( - field: TemplateField, _type: str, key: str - ) -> str: - """Replaces 'dict' type with 'code' or 'file'.""" - if "dict" in _type.lower(): - if key == "dict_": - field.field_type = "file" - field.suffixes = [".json", ".yaml", ".yml"] - field.file_types = ["json", "yaml", "yml"] - else: - field.field_type = "code" - return field.field_type - - @staticmethod - def set_field_default_value(field: TemplateField, value: dict, key: str) -> None: - """Sets the field value with the default value if present.""" - if "default" in value: - field.value = value["default"] - if key == "headers": - field.value = """{'Authorization': 'Bearer '}""" diff --git a/src/backend/langflow/template/field/__init__.py b/src/backend/langflow/template/field/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/src/backend/langflow/template/field/base.py b/src/backend/langflow/template/field/base.py new file mode 100644 index 000000000..a1de2c1b6 --- /dev/null +++ b/src/backend/langflow/template/field/base.py @@ -0,0 +1,43 @@ +from abc import ABC +from typing import Any, Optional, Union + +from pydantic import BaseModel + + +class TemplateFieldCreator(BaseModel, ABC): + field_type: str = "str" + required: bool = False + placeholder: str = "" + is_list: bool = False + show: bool = True + multiline: bool = False + value: Any = None + suffixes: list[str] = [] + fileTypes: list[str] = [] + file_types: list[str] = [] + content: Union[str, None] = None + password: bool = False + options: list[str] = [] + name: str = "" + display_name: Optional[str] = None + advanced: bool = False + + def to_dict(self): + result = self.dict() + # Remove key if it is None + for key in list(result.keys()): + if result[key] is None or result[key] == []: + del result[key] + result["type"] = result.pop("field_type") + result["list"] = result.pop("is_list") + + if result.get("file_types"): + result["fileTypes"] = result.pop("file_types") + + if self.field_type == "file": + result["content"] = self.content + return result + + +class TemplateField(TemplateFieldCreator): + pass diff --git a/src/backend/langflow/template/frontend_node/__init__.py b/src/backend/langflow/template/frontend_node/__init__.py new file mode 100644 index 000000000..1aa946d41 --- /dev/null +++ b/src/backend/langflow/template/frontend_node/__init__.py @@ -0,0 +1,21 @@ +from langflow.template.frontend_node import ( + agents, + chains, + embeddings, + llms, + memories, + prompts, + tools, + vectorstores, +) + +__all__ = [ + "agents", + "chains", + "embeddings", + "memories", + "tools", + "llms", + "prompts", + "vectorstores", +] diff --git a/src/backend/langflow/template/frontend_node/agents.py b/src/backend/langflow/template/frontend_node/agents.py new file mode 100644 index 000000000..e4fe40187 --- /dev/null +++ b/src/backend/langflow/template/frontend_node/agents.py @@ -0,0 +1,233 @@ +from typing import Optional + +from langchain.agents import types + +from langflow.template.field.base import TemplateField +from langflow.template.frontend_node.base import FrontendNode +from langflow.template.template.base import Template + +NON_CHAT_AGENTS = { + agent_type: agent_class + for agent_type, agent_class in types.AGENT_TO_CLASS.items() + if "chat" not in agent_type.value +} + + +class SQLAgentNode(FrontendNode): + name: str = "SQLAgent" + template: Template = Template( + type_name="sql_agent", + fields=[ + TemplateField( + field_type="str", + required=True, + placeholder="", + is_list=False, + show=True, + multiline=False, + value="", + name="database_uri", + ), + TemplateField( + field_type="BaseLanguageModel", + required=True, + show=True, + name="llm", + display_name="LLM", + ), + ], + ) + description: str = """Construct a sql agent from an LLM and tools.""" + base_classes: list[str] = ["AgentExecutor"] + + def to_dict(self): + return super().to_dict() + + +class VectorStoreRouterAgentNode(FrontendNode): + name: str = "VectorStoreRouterAgent" + template: Template = Template( + type_name="vectorstorerouter_agent", + fields=[ + TemplateField( + field_type="VectorStoreRouterToolkit", + required=True, + show=True, + name="vectorstoreroutertoolkit", + display_name="Vector Store Router Toolkit", + ), + TemplateField( + field_type="BaseLanguageModel", + required=True, + show=True, + name="llm", + display_name="LLM", + ), + ], + ) + description: str = """Construct an agent from a Vector Store Router.""" + base_classes: list[str] = ["AgentExecutor"] + + def to_dict(self): + return super().to_dict() + + +class VectorStoreAgentNode(FrontendNode): + name: str = "VectorStoreAgent" + template: Template = Template( + type_name="vectorstore_agent", + fields=[ + TemplateField( + field_type="VectorStoreInfo", + required=True, + show=True, + name="vectorstoreinfo", + display_name="Vector Store Info", + ), + TemplateField( + field_type="BaseLanguageModel", + required=True, + show=True, + name="llm", + display_name="LLM", + ), + ], + ) + description: str = """Construct an agent from a Vector Store.""" + base_classes: list[str] = ["AgentExecutor"] + + def to_dict(self): + return super().to_dict() + + +class SQLDatabaseNode(FrontendNode): + name: str = "SQLDatabase" + template: Template = Template( + type_name="sql_database", + fields=[ + TemplateField( + field_type="str", + required=True, + is_list=False, + show=True, + multiline=False, + value="", + name="uri", + ), + ], + ) + description: str = """SQLAlchemy wrapper around a database.""" + base_classes: list[str] = ["SQLDatabase"] + + def to_dict(self): + return super().to_dict() + + +class CSVAgentNode(FrontendNode): + name: str = "CSVAgent" + template: Template = Template( + type_name="csv_agent", + fields=[ + TemplateField( + field_type="file", + required=True, + show=True, + name="path", + value="", + suffixes=[".csv"], + fileTypes=["csv"], + ), + TemplateField( + field_type="BaseLanguageModel", + required=True, + show=True, + name="llm", + display_name="LLM", + ), + ], + ) + description: str = """Construct a json agent from a CSV and tools.""" + base_classes: list[str] = ["AgentExecutor"] + + def to_dict(self): + return super().to_dict() + + +class InitializeAgentNode(FrontendNode): + name: str = "initialize_agent" + template: Template = Template( + type_name="initailize_agent", + fields=[ + TemplateField( + field_type="str", + required=True, + is_list=True, + show=True, + multiline=False, + options=list(NON_CHAT_AGENTS.keys()), + value=list(NON_CHAT_AGENTS.keys())[0], + name="agent", + advanced=False, + ), + TemplateField( + field_type="BaseChatMemory", + required=False, + show=True, + name="memory", + advanced=False, + ), + TemplateField( + field_type="Tool", + required=False, + show=True, + name="tools", + is_list=True, + advanced=False, + ), + TemplateField( + field_type="BaseLanguageModel", + required=True, + show=True, + name="llm", + display_name="LLM", + advanced=False, + ), + ], + ) + description: str = """Construct a json agent from an LLM and tools.""" + base_classes: list[str] = ["AgentExecutor", "function"] + + def to_dict(self): + return super().to_dict() + + @staticmethod + def format_field(field: TemplateField, name: Optional[str] = None) -> None: + # do nothing and don't return anything + pass + + +class JsonAgentNode(FrontendNode): + name: str = "JsonAgent" + template: Template = Template( + type_name="json_agent", + fields=[ + TemplateField( + field_type="BaseToolkit", + required=True, + show=True, + name="toolkit", + ), + TemplateField( + field_type="BaseLanguageModel", + required=True, + show=True, + name="llm", + display_name="LLM", + ), + ], + ) + description: str = """Construct a json agent from an LLM and tools.""" + base_classes: list[str] = ["AgentExecutor"] + + def to_dict(self): + return super().to_dict() diff --git a/src/backend/langflow/template/frontend_node/base.py b/src/backend/langflow/template/frontend_node/base.py new file mode 100644 index 000000000..69fb07752 --- /dev/null +++ b/src/backend/langflow/template/frontend_node/base.py @@ -0,0 +1,197 @@ +import re +from typing import List, Optional + +from pydantic import BaseModel + +from langflow.template.constants import FORCE_SHOW_FIELDS +from langflow.template.field.base import TemplateField +from langflow.template.template.base import Template +from langflow.utils import constants + + +class FrontendNode(BaseModel): + template: Template + description: str + base_classes: List[str] + name: str = "" + + def to_dict(self) -> dict: + return { + self.name: { + "template": self.template.to_dict(self.format_field), + "description": self.description, + "base_classes": self.base_classes, + } + } + + @staticmethod + def format_field(field: TemplateField, name: Optional[str] = None) -> None: + """Formats a given field based on its attributes and value.""" + SPECIAL_FIELD_HANDLERS = { + "allowed_tools": lambda field: "Tool", + "max_value_length": lambda field: "int", + } + + key = field.name + value = field.to_dict() + _type = value["type"] + + _type = FrontendNode.remove_optional(_type) + _type, is_list = FrontendNode.check_for_list_type(_type) + field.is_list = is_list or field.is_list + _type = FrontendNode.replace_mapping_with_dict(_type) + _type = FrontendNode.handle_union_type(_type) + + field.field_type = FrontendNode.handle_special_field( + field, key, _type, SPECIAL_FIELD_HANDLERS + ) + field.field_type = FrontendNode.handle_dict_type(field, _type) + field.show = FrontendNode.should_show_field(key, field.required) + field.password = FrontendNode.should_be_password(key, field.show) + field.multiline = FrontendNode.should_be_multiline(key) + + FrontendNode.replace_default_value(field, value) + FrontendNode.handle_specific_field_values(field, key, name) + FrontendNode.handle_kwargs_field(field) + FrontendNode.handle_api_key_field(field, key) + + @staticmethod + def remove_optional(_type: str) -> str: + """Removes 'Optional' wrapper from the type if present.""" + return re.sub(r"Optional\[(.*)\]", r"\1", _type) + + @staticmethod + def check_for_list_type(_type: str) -> tuple: + """Checks for list type and returns the modified type and a boolean indicating if it's a list.""" + is_list = "List" in _type or "Sequence" in _type + if is_list: + _type = re.sub(r"(List|Sequence)\[(.*)\]", r"\2", _type) + return _type, is_list + + @staticmethod + def replace_mapping_with_dict(_type: str) -> str: + """Replaces 'Mapping' with 'dict'.""" + return _type.replace("Mapping", "dict") + + @staticmethod + def handle_union_type(_type: str) -> str: + """Simplifies the 'Union' type to the first type in the Union.""" + if "Union" in _type: + _type = _type.replace("Union[", "")[:-1] + _type = _type.split(",")[0] + _type = _type.replace("]", "").replace("[", "") + return _type + + @staticmethod + def handle_special_field( + field, key: str, _type: str, SPECIAL_FIELD_HANDLERS + ) -> str: + """Handles special field by using the respective handler if present.""" + handler = SPECIAL_FIELD_HANDLERS.get(key) + return handler(field) if handler else _type + + @staticmethod + def handle_dict_type(field: TemplateField, _type: str) -> str: + """Handles 'dict' type by replacing it with 'code' or 'file' based on the field name.""" + if "dict" in _type.lower(): + if field.name == "dict_": + field.field_type = "file" + field.suffixes = [".json", ".yaml", ".yml"] + field.file_types = ["json", "yaml", "yml"] + else: + field.field_type = "code" + return _type + + @staticmethod + def replace_default_value(field: TemplateField, value: dict) -> None: + """Replaces default value with actual value if 'default' is present in value.""" + if "default" in value: + field.value = value["default"] + + @staticmethod + def handle_specific_field_values( + field: TemplateField, key: str, name: Optional[str] = None + ) -> None: + """Handles specific field values for certain fields.""" + if key == "headers": + field.value = """{'Authorization': + 'Bearer '}""" + if name == "OpenAI" and key == "model_name": + field.options = constants.OPENAI_MODELS + field.is_list = True + elif name == "ChatOpenAI" and key == "model_name": + field.options = constants.CHAT_OPENAI_MODELS + field.is_list = True + if "api_key" in key and "OpenAI" in str(name): + field.display_name = "OpenAI API Key" + field.required = False + if field.value is None: + field.value = "" + + @staticmethod + def handle_kwargs_field(field: TemplateField) -> None: + """Handles kwargs field by setting certain attributes.""" + if "kwargs" in field.name.lower(): + field.advanced = True + field.required = False + field.show = False + + @staticmethod + def handle_api_key_field(field: TemplateField, key: str) -> None: + """Handles api key field by setting certain attributes.""" + if "api" in key.lower() and "key" in key.lower(): + field.required = False + field.advanced = False + + @staticmethod + def should_show_field(key: str, required: bool) -> bool: + """Determines whether the field should be shown.""" + return ( + (required and key not in ["input_variables"]) + or key in FORCE_SHOW_FIELDS + or "api" in key + or ("key" in key and "input" not in key and "output" not in key) + ) + + @staticmethod + def should_be_password(key: str, show: bool) -> bool: + """Determines whether the field should be a password field.""" + return ( + any(text in key.lower() for text in {"password", "token", "api", "key"}) + and show + ) + + @staticmethod + def should_be_multiline(key: str) -> bool: + """Determines whether the field should be multiline.""" + return key in { + "suffix", + "prefix", + "template", + "examples", + "code", + "headers", + "description", + } + + @staticmethod + def replace_dict_with_code_or_file( + field: TemplateField, _type: str, key: str + ) -> str: + """Replaces 'dict' type with 'code' or 'file'.""" + if "dict" in _type.lower(): + if key == "dict_": + field.field_type = "file" + field.suffixes = [".json", ".yaml", ".yml"] + field.file_types = ["json", "yaml", "yml"] + else: + field.field_type = "code" + return field.field_type + + @staticmethod + def set_field_default_value(field: TemplateField, value: dict, key: str) -> None: + """Sets the field value with the default value if present.""" + if "default" in value: + field.value = value["default"] + if key == "headers": + field.value = """{'Authorization': 'Bearer '}""" diff --git a/src/backend/langflow/template/frontend_node/chains.py b/src/backend/langflow/template/frontend_node/chains.py new file mode 100644 index 000000000..f12f98111 --- /dev/null +++ b/src/backend/langflow/template/frontend_node/chains.py @@ -0,0 +1,158 @@ +from typing import Optional + +from langflow.template.field.base import TemplateField +from langflow.template.frontend_node.base import FrontendNode +from langflow.template.template.base import Template + + +class ChainFrontendNode(FrontendNode): + @staticmethod + def format_field(field: TemplateField, name: Optional[str] = None) -> None: + FrontendNode.format_field(field, name) + + field.advanced = False + if "key" in field.name: + field.password = False + field.show = False + if field.name in ["input_key", "output_key"]: + field.required = True + field.show = True + field.advanced = True + + # Separated for possible future changes + if field.name == "prompt" and field.value is None: + # if no prompt is provided, use the default prompt + field.required = False + field.show = True + field.advanced = False + if field.name == "memory": + field.required = False + field.show = True + field.advanced = False + if field.name == "verbose": + field.required = False + field.show = True + field.advanced = True + if field.name == "llm": + field.required = True + field.show = True + field.advanced = False + + +class SeriesCharacterChainNode(FrontendNode): + name: str = "SeriesCharacterChain" + template: Template = Template( + type_name="SeriesCharacterChain", + fields=[ + TemplateField( + field_type="str", + required=True, + placeholder="", + is_list=False, + show=True, + advanced=False, + multiline=False, + name="character", + ), + TemplateField( + field_type="str", + required=True, + placeholder="", + is_list=False, + show=True, + advanced=False, + multiline=False, + name="series", + ), + TemplateField( + field_type="BaseLanguageModel", + required=True, + placeholder="", + is_list=False, + show=True, + advanced=False, + multiline=False, + name="llm", + display_name="LLM", + ), + ], + ) + description: str = "SeriesCharacterChain is a chain you can use to have a conversation with a character from a series." # noqa + base_classes: list[str] = [ + "LLMChain", + "BaseCustomChain", + "Chain", + "ConversationChain", + "SeriesCharacterChain", + "function", + ] + + +class TimeTravelGuideChainNode(FrontendNode): + name: str = "TimeTravelGuideChain" + template: Template = Template( + type_name="TimeTravelGuideChain", + fields=[ + TemplateField( + field_type="BaseLanguageModel", + required=True, + placeholder="", + is_list=False, + show=True, + advanced=False, + multiline=False, + name="llm", + display_name="LLM", + ), + TemplateField( + field_type="BaseChatMemory", + required=False, + show=True, + name="memory", + advanced=False, + ), + ], + ) + description: str = "Time travel guide chain to be used in the flow." + base_classes: list[str] = [ + "LLMChain", + "BaseCustomChain", + "TimeTravelGuideChain", + "Chain", + "ConversationChain", + ] + + +class MidJourneyPromptChainNode(FrontendNode): + name: str = "MidJourneyPromptChain" + template: Template = Template( + type_name="MidJourneyPromptChain", + fields=[ + TemplateField( + field_type="BaseLanguageModel", + required=True, + placeholder="", + is_list=False, + show=True, + advanced=False, + multiline=False, + name="llm", + display_name="LLM", + ), + TemplateField( + field_type="BaseChatMemory", + required=False, + show=True, + name="memory", + advanced=False, + ), + ], + ) + description: str = "MidJourneyPromptChain is a chain you can use to generate new MidJourney prompts." + base_classes: list[str] = [ + "LLMChain", + "BaseCustomChain", + "Chain", + "ConversationChain", + "MidJourneyPromptChain", + ] diff --git a/src/backend/langflow/template/frontend_node/embeddings.py b/src/backend/langflow/template/frontend_node/embeddings.py new file mode 100644 index 000000000..d21e12e73 --- /dev/null +++ b/src/backend/langflow/template/frontend_node/embeddings.py @@ -0,0 +1,38 @@ +from typing import Optional + +from langflow.template.field.base import TemplateField +from langflow.template.frontend_node.base import FrontendNode + + +class EmbeddingFrontendNode(FrontendNode): + @staticmethod + def format_jina_fields(field: TemplateField): + if "jina" in field.name: + field.show = True + field.advanced = False + + if "auth" in field.name or "token" in field.name: + field.password = True + field.show = True + field.advanced = False + + if field.name == "jina_api_url": + field.show = True + field.advanced = True + field.display_name = "Jina API URL" + field.password = False + + @staticmethod + def format_field(field: TemplateField, name: Optional[str] = None) -> None: + FrontendNode.format_field(field, name) + field.advanced = not field.required + field.show = True + if field.name == "headers": + field.show = False + + if "openai" in field.name: + field.show = True + field.advanced = "api_key" not in field.name + + # Format Jina fields + EmbeddingFrontendNode.format_jina_fields(field) diff --git a/src/backend/langflow/template/frontend_node/llms.py b/src/backend/langflow/template/frontend_node/llms.py new file mode 100644 index 000000000..044025ec1 --- /dev/null +++ b/src/backend/langflow/template/frontend_node/llms.py @@ -0,0 +1,41 @@ +from typing import Optional + +from langflow.template.field.base import TemplateField +from langflow.template.frontend_node.base import FrontendNode + + +class LLMFrontendNode(FrontendNode): + @staticmethod + def format_field(field: TemplateField, name: Optional[str] = None) -> None: + display_names_dict = { + "huggingfacehub_api_token": "HuggingFace Hub API Token", + } + FrontendNode.format_field(field, name) + SHOW_FIELDS = ["repo_id"] + if field.name in SHOW_FIELDS: + field.show = True + + if "api" in field.name and ("key" in field.name or "token" in field.name): + field.password = True + field.show = True + # Required should be False to support + # loading the API key from environment variables + field.required = False + field.advanced = False + + if field.name == "task": + field.required = True + field.show = True + field.is_list = True + field.options = ["text-generation", "text2text-generation"] + field.advanced = True + + if display_name := display_names_dict.get(field.name): + field.display_name = display_name + if field.name == "model_kwargs": + field.field_type = "code" + field.advanced = True + field.show = True + elif field.name in ["model_name", "temperature"]: + field.advanced = False + field.show = True diff --git a/src/backend/langflow/template/frontend_node/memories.py b/src/backend/langflow/template/frontend_node/memories.py new file mode 100644 index 000000000..91d892627 --- /dev/null +++ b/src/backend/langflow/template/frontend_node/memories.py @@ -0,0 +1,20 @@ +from typing import Optional + +from langflow.template.field.base import TemplateField +from langflow.template.frontend_node.base import FrontendNode + + +class MemoryFrontendNode(FrontendNode): + @staticmethod + def format_field(field: TemplateField, name: Optional[str] = None) -> None: + FrontendNode.format_field(field, name) + + if not isinstance(field.value, str): + field.value = None + if field.name == "k": + field.required = True + field.show = True + field.field_type = "int" + field.value = 10 + field.display_name = "Memory Size" + field.password = False diff --git a/src/backend/langflow/template/frontend_node/prompts.py b/src/backend/langflow/template/frontend_node/prompts.py new file mode 100644 index 000000000..830623a7e --- /dev/null +++ b/src/backend/langflow/template/frontend_node/prompts.py @@ -0,0 +1,111 @@ +from typing import Optional + +from langchain.agents.mrkl import prompt + +from langflow.template.constants import DEFAULT_PROMPT, HUMAN_PROMPT, SYSTEM_PROMPT +from langflow.template.field.base import TemplateField +from langflow.template.frontend_node.base import FrontendNode +from langflow.template.template.base import Template + + +class PromptFrontendNode(FrontendNode): + @staticmethod + def format_field(field: TemplateField, name: Optional[str] = None) -> None: + # if field.field_type == "StringPromptTemplate" + # change it to str + PROMPT_FIELDS = [ + "template", + "suffix", + "prefix", + "examples", + "format_instructions", + ] + if field.field_type == "StringPromptTemplate" and "Message" in str(name): + field.field_type = "prompt" + field.multiline = True + field.value = HUMAN_PROMPT if "Human" in field.name else SYSTEM_PROMPT + if field.name == "template" and field.value == "": + field.value = DEFAULT_PROMPT + + if field.name in PROMPT_FIELDS: + field.field_type = "prompt" + field.advanced = False + + if ( + "Union" in field.field_type + and "BaseMessagePromptTemplate" in field.field_type + ): + field.field_type = "BaseMessagePromptTemplate" + + # All prompt fields should be password=False + field.password = False + + +class PromptTemplateNode(FrontendNode): + name: str = "PromptTemplate" + template: Template + description: str + base_classes: list[str] = ["BasePromptTemplate"] + + def to_dict(self): + return super().to_dict() + + @staticmethod + def format_field(field: TemplateField, name: Optional[str] = None) -> None: + FrontendNode.format_field(field, name) + if field.name == "examples": + field.advanced = False + + +class BasePromptFrontendNode(FrontendNode): + name: str + template: Template + description: str + base_classes: list[str] + + def to_dict(self): + return super().to_dict() + + +class ZeroShotPromptNode(BasePromptFrontendNode): + name: str = "ZeroShotPrompt" + template: Template = Template( + type_name="zero_shot", + fields=[ + TemplateField( + field_type="str", + required=False, + placeholder="", + is_list=False, + show=True, + multiline=True, + value=prompt.PREFIX, + name="prefix", + ), + TemplateField( + field_type="str", + required=True, + placeholder="", + is_list=False, + show=True, + multiline=True, + value=prompt.SUFFIX, + name="suffix", + ), + TemplateField( + field_type="str", + required=False, + placeholder="", + is_list=False, + show=True, + multiline=True, + value=prompt.FORMAT_INSTRUCTIONS, + name="format_instructions", + ), + ], + ) + description: str = "Prompt template for Zero Shot Agent." + base_classes: list[str] = ["BasePromptTemplate"] + + def to_dict(self): + return super().to_dict() diff --git a/src/backend/langflow/template/frontend_node/tools.py b/src/backend/langflow/template/frontend_node/tools.py new file mode 100644 index 000000000..2819be4d9 --- /dev/null +++ b/src/backend/langflow/template/frontend_node/tools.py @@ -0,0 +1,83 @@ +from langflow.template.field.base import TemplateField +from langflow.template.frontend_node.base import FrontendNode +from langflow.template.template.base import Template +from langflow.utils.constants import DEFAULT_PYTHON_FUNCTION + + +class ToolNode(FrontendNode): + name: str = "Tool" + template: Template = Template( + type_name="Tool", + fields=[ + TemplateField( + field_type="str", + required=True, + placeholder="", + is_list=False, + show=True, + multiline=True, + value="", + name="name", + advanced=False, + ), + TemplateField( + field_type="str", + required=True, + placeholder="", + is_list=False, + show=True, + multiline=True, + value="", + name="description", + advanced=False, + ), + TemplateField( + name="func", + field_type="function", + required=True, + is_list=False, + show=True, + multiline=True, + advanced=False, + ), + TemplateField( + field_type="bool", + required=True, + placeholder="", + is_list=False, + show=True, + multiline=False, + value=False, + name="return_direct", + ), + ], + ) + description: str = "Tool to be used in the flow." + base_classes: list[str] = ["Tool"] + + def to_dict(self): + return super().to_dict() + + +class PythonFunctionNode(FrontendNode): + name: str = "PythonFunction" + template: Template = Template( + type_name="python_function", + fields=[ + TemplateField( + field_type="code", + required=True, + placeholder="", + is_list=False, + show=True, + value=DEFAULT_PYTHON_FUNCTION, + name="code", + advanced=False, + ) + ], + ) + description: str = "Python function to be executed." + base_classes: list[str] = ["function"] + + def to_dict(self): + return super().to_dict() diff --git a/src/backend/langflow/template/frontend_node/vectorstores.py b/src/backend/langflow/template/frontend_node/vectorstores.py new file mode 100644 index 000000000..76f623852 --- /dev/null +++ b/src/backend/langflow/template/frontend_node/vectorstores.py @@ -0,0 +1,64 @@ +from typing import Optional + +from langflow.template.field.base import TemplateField +from langflow.template.frontend_node.base import FrontendNode + + +class VectorStoreFrontendNode(FrontendNode): + @staticmethod + def format_field(field: TemplateField, name: Optional[str] = None) -> None: + FrontendNode.format_field(field, name) + # Define common field attributes + basic_fields = ["work_dir", "collection_name", "api_key", "location"] + advanced_fields = [ + "n_dim", + "key", + "prefix", + "distance_func", + "content_payload_key", + "metadata_payload_key", + "timeout", + "host", + "path", + "url", + "port", + "https", + "prefer_grpc", + "grpc_port", + ] + + # Check and set field attributes + if field.name == "texts": + field.name = "documents" + field.field_type = "TextSplitter" + field.display_name = "Text Splitter" + field.required = True + field.show = True + field.advanced = False + + elif "embedding" in field.name: + # for backwards compatibility + field.name = "embedding" + field.required = True + field.show = True + field.advanced = False + field.display_name = "Embedding" + field.field_type = "Embeddings" + + elif field.name in basic_fields: + field.show = True + field.advanced = False + if field.name == "api_key": + field.display_name = "API Key" + field.password = True + elif field.name == "location": + field.value = ":memory:" + field.placeholder = ":memory:" + + elif field.name in advanced_fields: + field.show = True + field.advanced = True + if "key" in field.name: + field.password = False + # TODO: Weaviate requires weaviate_url to be passed as it is not part of + # the class or from_texts method. We need the add_extra_fields to fix this diff --git a/src/backend/langflow/template/nodes.py b/src/backend/langflow/template/nodes.py index 9b31bf884..8b1378917 100644 --- a/src/backend/langflow/template/nodes.py +++ b/src/backend/langflow/template/nodes.py @@ -1,691 +1 @@ -from typing import Optional -from langchain.agents import loading -from langchain.agents.mrkl import prompt - -from langflow.template.base import FrontendNode, Template, TemplateField -from langflow.template.constants import DEFAULT_PROMPT, HUMAN_PROMPT, SYSTEM_PROMPT -from langflow.utils.constants import DEFAULT_PYTHON_FUNCTION - -NON_CHAT_AGENTS = { - agent_type: agent_class - for agent_type, agent_class in loading.AGENT_TO_CLASS.items() - if "chat" not in agent_type.value -} - - -class BasePromptFrontendNode(FrontendNode): - name: str - template: Template - description: str - base_classes: list[str] - - def to_dict(self): - return super().to_dict() - - -class ZeroShotPromptNode(BasePromptFrontendNode): - name: str = "ZeroShotPrompt" - template: Template = Template( - type_name="zero_shot", - fields=[ - TemplateField( - field_type="str", - required=False, - placeholder="", - is_list=False, - show=True, - multiline=True, - value=prompt.PREFIX, - name="prefix", - ), - TemplateField( - field_type="str", - required=True, - placeholder="", - is_list=False, - show=True, - multiline=True, - value=prompt.SUFFIX, - name="suffix", - ), - TemplateField( - field_type="str", - required=False, - placeholder="", - is_list=False, - show=True, - multiline=True, - value=prompt.FORMAT_INSTRUCTIONS, - name="format_instructions", - ), - ], - ) - description: str = "Prompt template for Zero Shot Agent." - base_classes: list[str] = ["BasePromptTemplate"] - - def to_dict(self): - return super().to_dict() - - -class PromptTemplateNode(FrontendNode): - name: str = "PromptTemplate" - template: Template - description: str - base_classes: list[str] = ["BasePromptTemplate"] - - def to_dict(self): - return super().to_dict() - - @staticmethod - def format_field(field: TemplateField, name: Optional[str] = None) -> None: - FrontendNode.format_field(field, name) - if field.name == "examples": - field.advanced = False - - -class PythonFunctionNode(FrontendNode): - name: str = "PythonFunction" - template: Template = Template( - type_name="python_function", - fields=[ - TemplateField( - field_type="code", - required=True, - placeholder="", - is_list=False, - show=True, - value=DEFAULT_PYTHON_FUNCTION, - name="code", - advanced=False, - ) - ], - ) - description: str = "Python function to be executed." - base_classes: list[str] = ["function"] - - def to_dict(self): - return super().to_dict() - - -class MidJourneyPromptChainNode(FrontendNode): - name: str = "MidJourneyPromptChain" - template: Template = Template( - type_name="MidJourneyPromptChain", - fields=[ - TemplateField( - field_type="BaseLanguageModel", - required=True, - placeholder="", - is_list=False, - show=True, - advanced=False, - multiline=False, - name="llm", - display_name="LLM", - ), - TemplateField( - field_type="BaseChatMemory", - required=False, - show=True, - name="memory", - advanced=False, - ), - ], - ) - description: str = "MidJourneyPromptChain is a chain you can use to generate new MidJourney prompts." - base_classes: list[str] = [ - "LLMChain", - "BaseCustomChain", - "Chain", - "ConversationChain", - "MidJourneyPromptChain", - ] - - -class TimeTravelGuideChainNode(FrontendNode): - name: str = "TimeTravelGuideChain" - template: Template = Template( - type_name="TimeTravelGuideChain", - fields=[ - TemplateField( - field_type="BaseLanguageModel", - required=True, - placeholder="", - is_list=False, - show=True, - advanced=False, - multiline=False, - name="llm", - display_name="LLM", - ), - TemplateField( - field_type="BaseChatMemory", - required=False, - show=True, - name="memory", - advanced=False, - ), - ], - ) - description: str = "Time travel guide chain to be used in the flow." - base_classes: list[str] = [ - "LLMChain", - "BaseCustomChain", - "TimeTravelGuideChain", - "Chain", - "ConversationChain", - ] - - -class SeriesCharacterChainNode(FrontendNode): - name: str = "SeriesCharacterChain" - template: Template = Template( - type_name="SeriesCharacterChain", - fields=[ - TemplateField( - field_type="str", - required=True, - placeholder="", - is_list=False, - show=True, - advanced=False, - multiline=False, - name="character", - ), - TemplateField( - field_type="str", - required=True, - placeholder="", - is_list=False, - show=True, - advanced=False, - multiline=False, - name="series", - ), - TemplateField( - field_type="BaseLanguageModel", - required=True, - placeholder="", - is_list=False, - show=True, - advanced=False, - multiline=False, - name="llm", - display_name="LLM", - ), - ], - ) - description: str = "SeriesCharacterChain is a chain you can use to have a conversation with a character from a series." # noqa - base_classes: list[str] = [ - "LLMChain", - "BaseCustomChain", - "Chain", - "ConversationChain", - "SeriesCharacterChain", - "function", - ] - - -class ToolNode(FrontendNode): - name: str = "Tool" - template: Template = Template( - type_name="Tool", - fields=[ - TemplateField( - field_type="str", - required=True, - placeholder="", - is_list=False, - show=True, - multiline=True, - value="", - name="name", - advanced=False, - ), - TemplateField( - field_type="str", - required=True, - placeholder="", - is_list=False, - show=True, - multiline=True, - value="", - name="description", - advanced=False, - ), - TemplateField( - name="func", - field_type="function", - required=True, - is_list=False, - show=True, - multiline=True, - advanced=False, - ), - TemplateField( - field_type="bool", - required=True, - placeholder="", - is_list=False, - show=True, - multiline=False, - value=False, - name="return_direct", - ), - ], - ) - description: str = "Tool to be used in the flow." - base_classes: list[str] = ["Tool"] - - def to_dict(self): - return super().to_dict() - - -class JsonAgentNode(FrontendNode): - name: str = "JsonAgent" - template: Template = Template( - type_name="json_agent", - fields=[ - TemplateField( - field_type="BaseToolkit", - required=True, - show=True, - name="toolkit", - ), - TemplateField( - field_type="BaseLanguageModel", - required=True, - show=True, - name="llm", - display_name="LLM", - ), - ], - ) - description: str = """Construct a json agent from an LLM and tools.""" - base_classes: list[str] = ["AgentExecutor"] - - def to_dict(self): - return super().to_dict() - - -class InitializeAgentNode(FrontendNode): - name: str = "initialize_agent" - template: Template = Template( - type_name="initailize_agent", - fields=[ - TemplateField( - field_type="str", - required=True, - is_list=True, - show=True, - multiline=False, - options=list(NON_CHAT_AGENTS.keys()), - value=list(NON_CHAT_AGENTS.keys())[0], - name="agent", - advanced=False, - ), - TemplateField( - field_type="BaseChatMemory", - required=False, - show=True, - name="memory", - advanced=False, - ), - TemplateField( - field_type="Tool", - required=False, - show=True, - name="tools", - is_list=True, - advanced=False, - ), - TemplateField( - field_type="BaseLanguageModel", - required=True, - show=True, - name="llm", - display_name="LLM", - advanced=False, - ), - ], - ) - description: str = """Construct a json agent from an LLM and tools.""" - base_classes: list[str] = ["AgentExecutor", "function"] - - def to_dict(self): - return super().to_dict() - - @staticmethod - def format_field(field: TemplateField, name: Optional[str] = None) -> None: - # do nothing and don't return anything - pass - - -class CSVAgentNode(FrontendNode): - name: str = "CSVAgent" - template: Template = Template( - type_name="csv_agent", - fields=[ - TemplateField( - field_type="file", - required=True, - show=True, - name="path", - value="", - suffixes=[".csv"], - fileTypes=["csv"], - ), - TemplateField( - field_type="BaseLanguageModel", - required=True, - show=True, - name="llm", - display_name="LLM", - ), - ], - ) - description: str = """Construct a json agent from a CSV and tools.""" - base_classes: list[str] = ["AgentExecutor"] - - def to_dict(self): - return super().to_dict() - - -class SQLDatabaseNode(FrontendNode): - name: str = "SQLDatabase" - template: Template = Template( - type_name="sql_database", - fields=[ - TemplateField( - field_type="str", - required=True, - is_list=False, - show=True, - multiline=False, - value="", - name="uri", - ), - ], - ) - description: str = """SQLAlchemy wrapper around a database.""" - base_classes: list[str] = ["SQLDatabase"] - - def to_dict(self): - return super().to_dict() - - -class VectorStoreAgentNode(FrontendNode): - name: str = "VectorStoreAgent" - template: Template = Template( - type_name="vectorstore_agent", - fields=[ - TemplateField( - field_type="VectorStoreInfo", - required=True, - show=True, - name="vectorstoreinfo", - display_name="Vector Store Info", - ), - TemplateField( - field_type="BaseLanguageModel", - required=True, - show=True, - name="llm", - display_name="LLM", - ), - ], - ) - description: str = """Construct an agent from a Vector Store.""" - base_classes: list[str] = ["AgentExecutor"] - - def to_dict(self): - return super().to_dict() - - -class VectorStoreRouterAgentNode(FrontendNode): - name: str = "VectorStoreRouterAgent" - template: Template = Template( - type_name="vectorstorerouter_agent", - fields=[ - TemplateField( - field_type="VectorStoreRouterToolkit", - required=True, - show=True, - name="vectorstoreroutertoolkit", - display_name="Vector Store Router Toolkit", - ), - TemplateField( - field_type="BaseLanguageModel", - required=True, - show=True, - name="llm", - display_name="LLM", - ), - ], - ) - description: str = """Construct an agent from a Vector Store Router.""" - base_classes: list[str] = ["AgentExecutor"] - - def to_dict(self): - return super().to_dict() - - -class SQLAgentNode(FrontendNode): - name: str = "SQLAgent" - template: Template = Template( - type_name="sql_agent", - fields=[ - TemplateField( - field_type="str", - required=True, - placeholder="", - is_list=False, - show=True, - multiline=False, - value="", - name="database_uri", - ), - TemplateField( - field_type="BaseLanguageModel", - required=True, - show=True, - name="llm", - display_name="LLM", - ), - ], - ) - description: str = """Construct a sql agent from an LLM and tools.""" - base_classes: list[str] = ["AgentExecutor"] - - def to_dict(self): - return super().to_dict() - - -class PromptFrontendNode(FrontendNode): - @staticmethod - def format_field(field: TemplateField, name: Optional[str] = None) -> None: - # if field.field_type == "StringPromptTemplate" - # change it to str - PROMPT_FIELDS = [ - "template", - "suffix", - "prefix", - "examples", - "format_instructions", - ] - if field.field_type == "StringPromptTemplate" and "Message" in str(name): - field.field_type = "prompt" - field.multiline = True - field.value = HUMAN_PROMPT if "Human" in field.name else SYSTEM_PROMPT - if field.name == "template" and field.value == "": - field.value = DEFAULT_PROMPT - - if field.name in PROMPT_FIELDS: - field.field_type = "prompt" - field.advanced = False - - if ( - "Union" in field.field_type - and "BaseMessagePromptTemplate" in field.field_type - ): - field.field_type = "BaseMessagePromptTemplate" - - # All prompt fields should be password=False - field.password = False - - -class MemoryFrontendNode(FrontendNode): - @staticmethod - def format_field(field: TemplateField, name: Optional[str] = None) -> None: - FrontendNode.format_field(field, name) - - if not isinstance(field.value, str): - field.value = None - if field.name == "k": - field.required = True - field.show = True - field.field_type = "int" - field.value = 10 - field.display_name = "Memory Size" - field.password = False - - -class ChainFrontendNode(FrontendNode): - @staticmethod - def format_field(field: TemplateField, name: Optional[str] = None) -> None: - FrontendNode.format_field(field, name) - - field.advanced = False - if "key" in field.name: - field.password = False - field.show = False - if field.name in ["input_key", "output_key"]: - field.required = True - field.show = True - field.advanced = True - - # Separated for possible future changes - if field.name == "prompt" and field.value is None: - # if no prompt is provided, use the default prompt - field.required = False - field.show = True - field.advanced = False - if field.name == "memory": - field.required = False - field.show = True - field.advanced = False - if field.name == "verbose": - field.required = False - field.show = True - field.advanced = True - if field.name == "llm": - field.required = True - field.show = True - field.advanced = False - - -class LLMFrontendNode(FrontendNode): - @staticmethod - def format_field(field: TemplateField, name: Optional[str] = None) -> None: - display_names_dict = { - "huggingfacehub_api_token": "HuggingFace Hub API Token", - } - FrontendNode.format_field(field, name) - SHOW_FIELDS = ["repo_id"] - if field.name in SHOW_FIELDS: - field.show = True - - if "api" in field.name and ("key" in field.name or "token" in field.name): - field.password = True - field.show = True - # Required should be False to support - # loading the API key from environment variables - field.required = False - field.advanced = False - - if field.name == "task": - field.required = True - field.show = True - field.is_list = True - field.options = ["text-generation", "text2text-generation"] - field.advanced = True - - if display_name := display_names_dict.get(field.name): - field.display_name = display_name - if field.name == "model_kwargs": - field.field_type = "code" - field.advanced = True - field.show = True - elif field.name in ["model_name", "temperature"]: - field.advanced = False - field.show = True - - -class EmbeddingFrontendNode(FrontendNode): - @staticmethod - def format_field(field: TemplateField, name: Optional[str] = None) -> None: - FrontendNode.format_field(field, name) - if field.name == "headers": - field.show = False - - -class VectorStoreFrontendNode(FrontendNode): - @staticmethod - def format_field(field: TemplateField, name: Optional[str] = None) -> None: - FrontendNode.format_field(field, name) - # Define common field attributes - basic_fields = ["work_dir", "collection_name", "api_key", "location"] - advanced_fields = [ - "n_dim", - "key", - "prefix", - "distance_func", - "content_payload_key", - "metadata_payload_key", - "timeout", - "host", - "path", - "url", - "port", - "https", - "prefer_grpc", - "grpc_port", - ] - - # Check and set field attributes - if field.name == "texts": - field.name = "documents" - field.field_type = "TextSplitter" - field.display_name = "Text Splitter" - field.required = True - field.show = True - field.advanced = False - - elif "embedding" in field.name: - # for backwards compatibility - field.name = "embedding" - field.required = True - field.show = True - field.advanced = False - field.display_name = "Embedding" - field.field_type = "Embeddings" - - elif field.name in basic_fields: - field.show = True - field.advanced = False - if field.name == "api_key": - field.display_name = "API Key" - field.password = True - elif field.name == "location": - field.value = ":memory:" - field.placeholder = ":memory:" - - elif field.name in advanced_fields: - field.show = True - field.advanced = True - if "key" in field.name: - field.password = False - - # TODO: Weaviate requires weaviate_url to be passed as it is not part of - # the class or from_texts method. We need the add_extra_fields to fix this diff --git a/src/backend/langflow/template/template/__init__.py b/src/backend/langflow/template/template/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/src/backend/langflow/template/template/base.py b/src/backend/langflow/template/template/base.py new file mode 100644 index 000000000..9279f5efb --- /dev/null +++ b/src/backend/langflow/template/template/base.py @@ -0,0 +1,25 @@ +from typing import Callable, Optional, Union + +from pydantic import BaseModel + +from langflow.template.field.base import TemplateField + + +class Template(BaseModel): + type_name: str + fields: list[TemplateField] + + def process_fields( + self, + name: Optional[str] = None, + format_field_func: Union[Callable, None] = None, + ): + if format_field_func: + for field in self.fields: + format_field_func(field, name) + + def to_dict(self, format_field_func=None): + self.process_fields(self.type_name, format_field_func) + result = {field.name: field.to_dict() for field in self.fields} + result["_type"] = self.type_name # type: ignore + return result diff --git a/tests/test_frontend_nodes.py b/tests/test_frontend_nodes.py index 673bdaec0..54fc783e8 100644 --- a/tests/test_frontend_nodes.py +++ b/tests/test_frontend_nodes.py @@ -1,5 +1,7 @@ import pytest -from langflow.template.base import FrontendNode, Template, TemplateField +from langflow.template.field.base import TemplateField +from langflow.template.frontend_node.base import FrontendNode +from langflow.template.template.base import Template @pytest.fixture