🐛 fix(utils.py): move merge_nested_dicts function to the top of the file for better organization

✨ feat(utils.py): add merge_nested_dicts function to merge nested dictionaries recursively
🐛 fix(endpoints.py): import merge_nested_dicts function from the correct module
✨ feat(endpoints.py): use merge_nested_dicts function to merge dictionaries in build_langchain_custom_component_list_from_path function
🐛 fix(types.py): import merge_nested_dicts function from the correct module
✨ feat(types.py): use merge_nested_dicts function to merge dictionaries in build_langchain_custom_component_list_from_path function
🐛 fix(types.py): import merge_nested_dicts function from the correct module
✨ feat(types.py): use merge_nested_dicts function to merge valid and invalid menus in build_langchain_custom_component_list_from_path function
🐛 fix(tools.py): import Optional from typing module
✨ feat(tools.py): add CustomComponentEmptyNode class to represent an empty custom component template
This commit is contained in:
gustavoschaedler 2023-07-25 21:58:09 +01:00
commit 883f852b73
4 changed files with 76 additions and 20 deletions

View file

@ -57,3 +57,12 @@ def build_input_keys_response(langchain_object, artifacts):
input_keys_response["template"] = langchain_object.prompt.template input_keys_response["template"] = langchain_object.prompt.template
return input_keys_response return input_keys_response
def merge_nested_dicts(dict1, dict2):
for key, value in dict2.items():
if isinstance(value, dict) and isinstance(dict1.get(key), dict):
dict1[key] = merge_nested_dicts(dict1[key], value)
else:
dict1[key] = value
return dict1

View file

@ -21,6 +21,8 @@ from langflow.api.v1.schemas import (
CustomComponentCode, CustomComponentCode,
) )
from langflow.api.utils import merge_nested_dicts
from langflow.interface.types import ( from langflow.interface.types import (
build_langchain_types_dict, build_langchain_types_dict,
build_langchain_template_custom_component, build_langchain_template_custom_component,
@ -34,16 +36,6 @@ from sqlmodel import Session
router = APIRouter(tags=["Base"]) router = APIRouter(tags=["Base"])
# TODO: Move to correct local
def merge_nested_dicts(dict1, dict2):
for key, value in dict2.items():
if isinstance(value, dict) and isinstance(dict1.get(key), dict):
dict1[key] = merge_nested_dicts(dict1[key], value)
else:
dict1[key] = value
return dict1
@router.get("/all") @router.get("/all")
def get_all(): def get_all():
native_components = build_langchain_types_dict() native_components = build_langchain_types_dict()

View file

@ -18,16 +18,21 @@ from langflow.interface.custom.base import custom_component_creator
from langflow.interface.custom.custom_component import CustomComponent from langflow.interface.custom.custom_component import CustomComponent
from langflow.template.field.base import TemplateField from langflow.template.field.base import TemplateField
from langflow.template.frontend_node.tools import CustomComponentNode from langflow.template.frontend_node.tools import (
CustomComponentNode,
CustomComponentEmptyNode,
)
from langflow.interface.retrievers.base import retriever_creator from langflow.interface.retrievers.base import retriever_creator
from langflow.interface.custom.directory_reader import DirectoryReader from langflow.interface.custom.directory_reader import DirectoryReader
from langflow.utils.logger import logger from langflow.utils.logger import logger
from langflow.utils.util import get_base_classes
from langflow.api.utils import merge_nested_dicts
import re import re
import warnings import warnings
import traceback import traceback
from fastapi import HTTPException from fastapi import HTTPException
from langflow.utils.util import get_base_classes
# Used to get the base_classes list # Used to get the base_classes list
@ -250,15 +255,14 @@ def build_langchain_custom_component_list_from_path(path: str):
# Build and validate all files # Build and validate all files
data = reader.build_component_menu_list(file_list) data = reader.build_component_menu_list(file_list)
valid_components = reader.filter_loaded_components(data, False)
# TODO: Handle those invalid components valid_components = reader.filter_loaded_components(data=data, with_errors=False)
reader.filter_loaded_components(data, True) invalid_components = reader.filter_loaded_components(data=data, with_errors=True)
menu = {} valid_menu = {}
for menu_item in valid_components["menu"]: for menu_item in valid_components["menu"]:
menu_name = menu_item["name"] menu_name = menu_item["name"]
menu[menu_name] = {} valid_menu[menu_name] = {}
for component in menu_item["components"]: for component in menu_item["components"]:
try: try:
@ -271,8 +275,33 @@ def build_langchain_custom_component_list_from_path(path: str):
component_extractor component_extractor
) )
menu[menu_name][component_name] = component_template valid_menu[menu_name][component_name] = component_template
except Exception as exc: except Exception as exc:
logger.error(f"Error while building custom component: {exc}") logger.error(f"Error while building custom component: {exc}")
return menu invalid_menu = {}
for menu_item in invalid_components["menu"]:
menu_name = menu_item["name"]
invalid_menu[menu_name] = {}
for component in menu_item["components"]:
try:
component_name = component["name"]
component_code = component["code"]
component_template = (
CustomComponentNode(
description="ERROR - Check your Python Code",
display_name=f"ERROR - {component_name}",
)
.to_dict()
.get(type(CustomComponent()).__name__)
)
component_template.get("template").get("code")["value"] = component_code
invalid_menu[menu_name][component_name] = component_template
except Exception as exc:
logger.error(f"Error while creating custom component: {exc}")
return merge_nested_dicts(valid_menu, invalid_menu)

View file

@ -5,6 +5,7 @@ from langflow.template.template.base import Template
from langflow.utils.constants import ( from langflow.utils.constants import (
DEFAULT_PYTHON_FUNCTION, DEFAULT_PYTHON_FUNCTION,
) )
from typing import Optional
class ToolNode(FrontendNode): class ToolNode(FrontendNode):
@ -160,7 +161,32 @@ class CustomComponentNode(FrontendNode):
) )
], ],
) )
description: str = "Python Class to be executed." description: str = "Dynamic Python code to be executed."
base_classes: list[str] = []
def to_dict(self):
return super().to_dict()
class CustomComponentEmptyNode(FrontendNode):
name: str = "CustomComponent"
template: Template = Template(
type_name="CustomComponent",
fields=[
TemplateField(
field_type="code",
required=True,
placeholder="",
is_list=False,
show=True,
value="",
name="code",
advanced=False,
dynamic=True,
)
],
)
description: str = "Dynamic Python code to be executed."
base_classes: list[str] = [] base_classes: list[str] = []
def to_dict(self): def to_dict(self):