Refactor custom component module (#1210)
This commit is contained in:
Gabriel Luiz Freitas Almeida
GitHub
commit
865ac0306a
14 changed files with 622 additions and 619 deletions
5
Makefile
5
Makefile
|
|
@ -18,9 +18,12 @@ coverage:
|
|||
--cov-report xml \
|
||||
--cov-report term-missing:skip-covered
|
||||
|
||||
# allow passing arguments to pytest
|
||||
tests:
|
||||
@make install_backend
|
||||
poetry run pytest tests --instafail
|
||||
|
||||
poetry run pytest tests --instafail $(args)
|
||||
# Use like:
|
||||
|
||||
format:
|
||||
poetry run ruff . --fix
|
||||
|
|
|
|||
|
|
@ -3,26 +3,24 @@ from typing import Annotated, Optional, Union
|
|||
|
||||
import sqlalchemy as sa
|
||||
from fastapi import APIRouter, Body, Depends, HTTPException, UploadFile, status
|
||||
from loguru import logger
|
||||
from sqlmodel import select
|
||||
|
||||
from langflow.api.utils import update_frontend_node_with_template_values
|
||||
from langflow.api.v1.schemas import (
|
||||
CustomComponentCode,
|
||||
ProcessResponse,
|
||||
TaskResponse,
|
||||
TaskStatusResponse,
|
||||
UploadFileResponse,
|
||||
)
|
||||
from langflow.api.v1.schemas import (CustomComponentCode, ProcessResponse,
|
||||
TaskResponse, TaskStatusResponse,
|
||||
UploadFileResponse)
|
||||
from langflow.interface.custom.custom_component import CustomComponent
|
||||
from langflow.interface.custom.directory_reader import DirectoryReader
|
||||
from langflow.interface.types import build_custom_component_template, create_and_validate_component
|
||||
from langflow.interface.custom.utils import (build_custom_component_template,
|
||||
create_and_validate_component)
|
||||
from langflow.processing.process import process_graph_cached, process_tweaks
|
||||
from langflow.services.auth.utils import api_key_security, get_current_active_user
|
||||
from langflow.services.auth.utils import (api_key_security,
|
||||
get_current_active_user)
|
||||
from langflow.services.cache.utils import save_uploaded_file
|
||||
from langflow.services.database.models.flow import Flow
|
||||
from langflow.services.database.models.user.model import User
|
||||
from langflow.services.deps import get_session, get_session_service, get_settings_service, get_task_service
|
||||
from langflow.services.deps import (get_session, get_session_service,
|
||||
get_settings_service, get_task_service)
|
||||
from loguru import logger
|
||||
from sqlmodel import select
|
||||
|
||||
try:
|
||||
from langflow.worker import process_graph_cached_task
|
||||
|
|
@ -32,9 +30,8 @@ except ImportError:
|
|||
raise NotImplementedError("Celery is not installed")
|
||||
|
||||
|
||||
from sqlmodel import Session
|
||||
|
||||
from langflow.services.task.service import TaskService
|
||||
from sqlmodel import Session
|
||||
|
||||
# build router
|
||||
router = APIRouter(tags=["Base"])
|
||||
|
|
@ -226,7 +223,7 @@ async def custom_component(
|
|||
|
||||
@router.post("/custom_component/reload", status_code=HTTPStatus.OK)
|
||||
async def reload_custom_component(path: str, user: User = Depends(get_current_active_user)):
|
||||
from langflow.interface.types import build_custom_component_template
|
||||
from langflow.interface.custom.utils import build_custom_component_template
|
||||
|
||||
try:
|
||||
reader = DirectoryReader("")
|
||||
|
|
|
|||
|
|
@ -0,0 +1,3 @@
|
|||
from .code_parser import CodeParser
|
||||
|
||||
__all__ = ["CodeParser"]
|
||||
320
src/backend/langflow/interface/custom/code_parser/code_parser.py
Normal file
320
src/backend/langflow/interface/custom/code_parser/code_parser.py
Normal file
|
|
@ -0,0 +1,320 @@
|
|||
import ast
|
||||
import inspect
|
||||
import operator
|
||||
import traceback
|
||||
from typing import Any, Dict, List, Type, Union
|
||||
|
||||
from cachetools import TTLCache, cachedmethod, keys
|
||||
from fastapi import HTTPException
|
||||
from langflow.interface.custom.schema import (CallableCodeDetails,
|
||||
ClassCodeDetails)
|
||||
|
||||
|
||||
class CodeSyntaxError(HTTPException):
|
||||
pass
|
||||
|
||||
|
||||
def get_data_type():
|
||||
from langflow.field_typing import Data
|
||||
|
||||
return Data
|
||||
|
||||
|
||||
def imports_key(*args, **kwargs):
|
||||
imports = kwargs.pop("imports")
|
||||
key = keys.methodkey(*args, **kwargs)
|
||||
key += tuple(imports)
|
||||
return key
|
||||
|
||||
|
||||
class CodeParser:
|
||||
"""
|
||||
A parser for Python source code, extracting code details.
|
||||
"""
|
||||
|
||||
def __init__(self, code: Union[str, Type]) -> None:
|
||||
"""
|
||||
Initializes the parser with the provided code.
|
||||
"""
|
||||
self.cache: TTLCache = TTLCache(maxsize=1024, ttl=60)
|
||||
if isinstance(code, type):
|
||||
if not inspect.isclass(code):
|
||||
raise ValueError("The provided code must be a class.")
|
||||
# If the code is a class, get its source code
|
||||
code = inspect.getsource(code)
|
||||
self.code = code
|
||||
self.data: Dict[str, Any] = {
|
||||
"imports": [],
|
||||
"functions": [],
|
||||
"classes": [],
|
||||
"global_vars": [],
|
||||
}
|
||||
self.handlers = {
|
||||
ast.Import: self.parse_imports,
|
||||
ast.ImportFrom: self.parse_imports,
|
||||
ast.FunctionDef: self.parse_functions,
|
||||
ast.ClassDef: self.parse_classes,
|
||||
ast.Assign: self.parse_global_vars,
|
||||
}
|
||||
|
||||
def get_tree(self):
|
||||
"""
|
||||
Parses the provided code to validate its syntax.
|
||||
It tries to parse the code into an abstract syntax tree (AST).
|
||||
"""
|
||||
try:
|
||||
tree = ast.parse(self.code)
|
||||
except SyntaxError as err:
|
||||
raise CodeSyntaxError(
|
||||
status_code=400,
|
||||
detail={"error": err.msg, "traceback": traceback.format_exc()},
|
||||
) from err
|
||||
|
||||
return tree
|
||||
|
||||
def parse_node(self, node: Union[ast.stmt, ast.AST]) -> None:
|
||||
"""
|
||||
Parses an AST node and updates the data
|
||||
dictionary with the relevant information.
|
||||
"""
|
||||
if handler := self.handlers.get(type(node)): # type: ignore
|
||||
handler(node) # type: ignore
|
||||
|
||||
def parse_imports(self, node: Union[ast.Import, ast.ImportFrom]) -> None:
|
||||
"""
|
||||
Extracts "imports" from the code, including aliases.
|
||||
"""
|
||||
if isinstance(node, ast.Import):
|
||||
for alias in node.names:
|
||||
if alias.asname:
|
||||
self.data["imports"].append(f"{alias.name} as {alias.asname}")
|
||||
else:
|
||||
self.data["imports"].append(alias.name)
|
||||
elif isinstance(node, ast.ImportFrom):
|
||||
for alias in node.names:
|
||||
if alias.asname:
|
||||
self.data["imports"].append((node.module, f"{alias.name} as {alias.asname}"))
|
||||
else:
|
||||
self.data["imports"].append((node.module, alias.name))
|
||||
|
||||
def parse_functions(self, node: ast.FunctionDef) -> None:
|
||||
"""
|
||||
Extracts "functions" from the code.
|
||||
"""
|
||||
self.data["functions"].append(self.parse_callable_details(node))
|
||||
|
||||
def parse_arg(self, arg, default):
|
||||
"""
|
||||
Parses an argument and its default value.
|
||||
"""
|
||||
arg_dict = {"name": arg.arg, "default": default}
|
||||
if arg.annotation:
|
||||
arg_dict["type"] = ast.unparse(arg.annotation)
|
||||
return arg_dict
|
||||
|
||||
@cachedmethod(operator.attrgetter("cache"))
|
||||
def construct_eval_env(self, return_type_str: str, imports) -> dict:
|
||||
"""
|
||||
Constructs an evaluation environment with the necessary imports for the return type,
|
||||
taking into account module aliases.
|
||||
"""
|
||||
eval_env: dict = {}
|
||||
for import_entry in imports:
|
||||
if isinstance(import_entry, tuple): # from module import name
|
||||
module, name = import_entry
|
||||
if name in return_type_str:
|
||||
exec(f"import {module}", eval_env)
|
||||
exec(f"from {module} import {name}", eval_env)
|
||||
else: # import module
|
||||
module = import_entry
|
||||
alias = None
|
||||
if " as " in module:
|
||||
module, alias = module.split(" as ")
|
||||
if module in return_type_str or (alias and alias in return_type_str):
|
||||
exec(f"import {module} as {alias if alias else module}", eval_env)
|
||||
return eval_env
|
||||
|
||||
@cachedmethod(cache=operator.attrgetter("cache"))
|
||||
def parse_callable_details(self, node: ast.FunctionDef) -> Dict[str, Any]:
|
||||
"""
|
||||
Extracts details from a single function or method node.
|
||||
"""
|
||||
return_type = None
|
||||
if node.returns:
|
||||
return_type_str = ast.unparse(node.returns)
|
||||
eval_env = self.construct_eval_env(return_type_str, tuple(self.data["imports"]))
|
||||
|
||||
try:
|
||||
return_type = eval(return_type_str, eval_env)
|
||||
except NameError:
|
||||
# Handle cases where the type is not found in the constructed environment
|
||||
pass
|
||||
|
||||
func = CallableCodeDetails(
|
||||
name=node.name,
|
||||
doc=ast.get_docstring(node),
|
||||
args=self.parse_function_args(node),
|
||||
body=self.parse_function_body(node),
|
||||
return_type=return_type or get_data_type(),
|
||||
has_return=self.parse_return_statement(node),
|
||||
)
|
||||
|
||||
return func.model_dump()
|
||||
|
||||
def parse_function_args(self, node: ast.FunctionDef) -> List[Dict[str, Any]]:
|
||||
"""
|
||||
Parses the arguments of a function or method node.
|
||||
"""
|
||||
args = []
|
||||
|
||||
args += self.parse_positional_args(node)
|
||||
args += self.parse_varargs(node)
|
||||
args += self.parse_keyword_args(node)
|
||||
# Commented out because we don't want kwargs
|
||||
# showing up as fields in the frontend
|
||||
# args += self.parse_kwargs(node)
|
||||
|
||||
return args
|
||||
|
||||
def parse_positional_args(self, node: ast.FunctionDef) -> List[Dict[str, Any]]:
|
||||
"""
|
||||
Parses the positional arguments of a function or method node.
|
||||
"""
|
||||
num_args = len(node.args.args)
|
||||
num_defaults = len(node.args.defaults)
|
||||
num_missing_defaults = num_args - num_defaults
|
||||
missing_defaults = [None] * num_missing_defaults
|
||||
default_values = [ast.unparse(default).strip("'") if default else None for default in node.args.defaults]
|
||||
# Now check all default values to see if there
|
||||
# are any "None" values in the middle
|
||||
default_values = [None if value == "None" else value for value in default_values]
|
||||
|
||||
defaults = missing_defaults + default_values
|
||||
|
||||
args = [self.parse_arg(arg, default) for arg, default in zip(node.args.args, defaults)]
|
||||
return args
|
||||
|
||||
def parse_varargs(self, node: ast.FunctionDef) -> List[Dict[str, Any]]:
|
||||
"""
|
||||
Parses the *args argument of a function or method node.
|
||||
"""
|
||||
args = []
|
||||
|
||||
if node.args.vararg:
|
||||
args.append(self.parse_arg(node.args.vararg, None))
|
||||
|
||||
return args
|
||||
|
||||
def parse_keyword_args(self, node: ast.FunctionDef) -> List[Dict[str, Any]]:
|
||||
"""
|
||||
Parses the keyword-only arguments of a function or method node.
|
||||
"""
|
||||
kw_defaults = [None] * (len(node.args.kwonlyargs) - len(node.args.kw_defaults)) + [
|
||||
ast.unparse(default) if default else None for default in node.args.kw_defaults
|
||||
]
|
||||
|
||||
args = [self.parse_arg(arg, default) for arg, default in zip(node.args.kwonlyargs, kw_defaults)]
|
||||
return args
|
||||
|
||||
def parse_kwargs(self, node: ast.FunctionDef) -> List[Dict[str, Any]]:
|
||||
"""
|
||||
Parses the **kwargs argument of a function or method node.
|
||||
"""
|
||||
args = []
|
||||
|
||||
if node.args.kwarg:
|
||||
args.append(self.parse_arg(node.args.kwarg, None))
|
||||
|
||||
return args
|
||||
|
||||
def parse_function_body(self, node: ast.FunctionDef) -> List[str]:
|
||||
"""
|
||||
Parses the body of a function or method node.
|
||||
"""
|
||||
return [ast.unparse(line) for line in node.body]
|
||||
|
||||
def parse_return_statement(self, node: ast.FunctionDef) -> bool:
|
||||
"""
|
||||
Parses the return statement of a function or method node.
|
||||
"""
|
||||
|
||||
return any(isinstance(n, ast.Return) for n in node.body)
|
||||
|
||||
def parse_assign(self, stmt):
|
||||
"""
|
||||
Parses an Assign statement and returns a dictionary
|
||||
with the target's name and value.
|
||||
"""
|
||||
for target in stmt.targets:
|
||||
if isinstance(target, ast.Name):
|
||||
return {"name": target.id, "value": ast.unparse(stmt.value)}
|
||||
|
||||
def parse_ann_assign(self, stmt):
|
||||
"""
|
||||
Parses an AnnAssign statement and returns a dictionary
|
||||
with the target's name, value, and annotation.
|
||||
"""
|
||||
if isinstance(stmt.target, ast.Name):
|
||||
return {
|
||||
"name": stmt.target.id,
|
||||
"value": ast.unparse(stmt.value) if stmt.value else None,
|
||||
"annotation": ast.unparse(stmt.annotation),
|
||||
}
|
||||
|
||||
def parse_function_def(self, stmt):
|
||||
"""
|
||||
Parses a FunctionDef statement and returns the parsed
|
||||
method and a boolean indicating if it's an __init__ method.
|
||||
"""
|
||||
method = self.parse_callable_details(stmt)
|
||||
return (method, True) if stmt.name == "__init__" else (method, False)
|
||||
|
||||
def parse_classes(self, node: ast.ClassDef) -> None:
|
||||
"""
|
||||
Extracts "classes" from the code, including inheritance and init methods.
|
||||
"""
|
||||
|
||||
class_details = ClassCodeDetails(
|
||||
name=node.name,
|
||||
doc=ast.get_docstring(node),
|
||||
bases=[ast.unparse(base) for base in node.bases],
|
||||
attributes=[],
|
||||
methods=[],
|
||||
init=None,
|
||||
)
|
||||
|
||||
for stmt in node.body:
|
||||
if isinstance(stmt, ast.Assign):
|
||||
if attr := self.parse_assign(stmt):
|
||||
class_details.attributes.append(attr)
|
||||
elif isinstance(stmt, ast.AnnAssign):
|
||||
if attr := self.parse_ann_assign(stmt):
|
||||
class_details.attributes.append(attr)
|
||||
elif isinstance(stmt, (ast.FunctionDef, ast.AsyncFunctionDef)):
|
||||
method, is_init = self.parse_function_def(stmt)
|
||||
if is_init:
|
||||
class_details.init = method
|
||||
else:
|
||||
class_details.methods.append(method)
|
||||
|
||||
self.data["classes"].append(class_details.model_dump())
|
||||
|
||||
def parse_global_vars(self, node: ast.Assign) -> None:
|
||||
"""
|
||||
Extracts global variables from the code.
|
||||
"""
|
||||
global_var = {
|
||||
"targets": [t.id if hasattr(t, "id") else ast.dump(t) for t in node.targets],
|
||||
"value": ast.unparse(node.value),
|
||||
}
|
||||
self.data["global_vars"].append(global_var)
|
||||
|
||||
def parse_code(self) -> Dict[str, Any]:
|
||||
"""
|
||||
Runs all parsing operations and returns the resulting data.
|
||||
"""
|
||||
tree = self.get_tree()
|
||||
|
||||
for node in ast.walk(tree):
|
||||
self.parse_node(node)
|
||||
return self.data
|
||||
39
src/backend/langflow/interface/custom/code_parser/utils.py
Normal file
39
src/backend/langflow/interface/custom/code_parser/utils.py
Normal file
|
|
@ -0,0 +1,39 @@
|
|||
import re
|
||||
from types import GenericAlias
|
||||
from typing import Any
|
||||
|
||||
|
||||
def extract_inner_type(return_type: str) -> str:
|
||||
"""
|
||||
Extracts the inner type from a type hint that is a list.
|
||||
"""
|
||||
if match := re.match(r"list\[(.*)\]", return_type, re.IGNORECASE):
|
||||
return match[1]
|
||||
return return_type
|
||||
|
||||
|
||||
def extract_inner_type_from_generic_alias(return_type: GenericAlias) -> Any:
|
||||
"""
|
||||
Extracts the inner type from a type hint that is a list.
|
||||
"""
|
||||
if return_type.__origin__ == list:
|
||||
return list(return_type.__args__)
|
||||
|
||||
return return_type
|
||||
|
||||
|
||||
def extract_union_types(return_type: str) -> list[str]:
|
||||
"""
|
||||
Extracts the inner type from a type hint that is a list.
|
||||
"""
|
||||
# If the return type is a Union, then we need to parse it
|
||||
return_type = return_type.replace("Union", "").replace("[", "").replace("]", "")
|
||||
return_types = return_type.split(",")
|
||||
return [item.strip() for item in return_types]
|
||||
|
||||
|
||||
def extract_union_types_from_generic_alias(return_type: GenericAlias) -> list:
|
||||
"""
|
||||
Extracts the inner type from a type hint that is a Union.
|
||||
"""
|
||||
return list(return_type.__args__)
|
||||
|
|
@ -0,0 +1,3 @@
|
|||
from .custom_component import CustomComponent
|
||||
|
||||
__all__ = ["CustomComponent"]
|
||||
|
|
@ -0,0 +1,88 @@
|
|||
import ast
|
||||
import operator
|
||||
import warnings
|
||||
from typing import Any, ClassVar, Optional
|
||||
|
||||
from cachetools import TTLCache, cachedmethod
|
||||
from fastapi import HTTPException
|
||||
|
||||
from langflow.interface.custom.code_parser import CodeParser
|
||||
from langflow.utils import validate
|
||||
|
||||
|
||||
class ComponentCodeNullError(HTTPException):
|
||||
pass
|
||||
|
||||
|
||||
class ComponentFunctionEntrypointNameNullError(HTTPException):
|
||||
pass
|
||||
|
||||
|
||||
class Component:
|
||||
ERROR_CODE_NULL: ClassVar[str] = "Python code must be provided."
|
||||
ERROR_FUNCTION_ENTRYPOINT_NAME_NULL: ClassVar[str] = "The name of the entrypoint function must be provided."
|
||||
|
||||
code: Optional[str] = None
|
||||
_function_entrypoint_name: str = "build"
|
||||
field_config: dict = {}
|
||||
_user_id: Optional[str]
|
||||
|
||||
def __init__(self, **data):
|
||||
self.cache = TTLCache(maxsize=1024, ttl=60)
|
||||
for key, value in data.items():
|
||||
if key == "user_id":
|
||||
setattr(self, "_user_id", value)
|
||||
else:
|
||||
setattr(self, key, value)
|
||||
|
||||
def __setattr__(self, key, value):
|
||||
if key == "_user_id" and hasattr(self, "_user_id"):
|
||||
warnings.warn("user_id is immutable and cannot be changed.")
|
||||
super().__setattr__(key, value)
|
||||
|
||||
@cachedmethod(cache=operator.attrgetter("cache"))
|
||||
def get_code_tree(self, code: str):
|
||||
parser = CodeParser(code)
|
||||
return parser.parse_code()
|
||||
|
||||
def get_function(self):
|
||||
if not self.code:
|
||||
raise ComponentCodeNullError(
|
||||
status_code=400,
|
||||
detail={"error": self.ERROR_CODE_NULL, "traceback": ""},
|
||||
)
|
||||
|
||||
if not self._function_entrypoint_name:
|
||||
raise ComponentFunctionEntrypointNameNullError(
|
||||
status_code=400,
|
||||
detail={
|
||||
"error": self.ERROR_FUNCTION_ENTRYPOINT_NAME_NULL,
|
||||
"traceback": "",
|
||||
},
|
||||
)
|
||||
|
||||
return validate.create_function(self.code, self._function_entrypoint_name)
|
||||
|
||||
def build_template_config(self, attributes) -> dict:
|
||||
template_config = {}
|
||||
|
||||
for item in attributes:
|
||||
item_name = item.get("name")
|
||||
|
||||
if item_value := item.get("value"):
|
||||
if "display_name" in item_name:
|
||||
template_config["display_name"] = ast.literal_eval(item_value)
|
||||
|
||||
elif "description" in item_name:
|
||||
template_config["description"] = ast.literal_eval(item_value)
|
||||
|
||||
elif "beta" in item_name:
|
||||
template_config["beta"] = ast.literal_eval(item_value)
|
||||
|
||||
elif "documentation" in item_name:
|
||||
template_config["documentation"] = ast.literal_eval(item_value)
|
||||
|
||||
return template_config
|
||||
|
||||
def build(self, *args: Any, **kwargs: Any) -> Any:
|
||||
raise NotImplementedError
|
||||
|
|
@ -0,0 +1,256 @@
|
|||
import operator
|
||||
from typing import Any, Callable, ClassVar, List, Optional, Union
|
||||
from uuid import UUID
|
||||
|
||||
import yaml
|
||||
from cachetools import TTLCache, cachedmethod
|
||||
from fastapi import HTTPException
|
||||
from langflow.interface.custom.code_parser.utils import (
|
||||
extract_inner_type_from_generic_alias,
|
||||
extract_union_types_from_generic_alias)
|
||||
from langflow.interface.custom.directory_reader import DirectoryReader
|
||||
from langflow.services.database.models.flow import Flow
|
||||
from langflow.services.database.utils import session_getter
|
||||
from langflow.services.deps import get_credential_service, get_db_service
|
||||
from langflow.utils import validate
|
||||
|
||||
from .component import Component
|
||||
|
||||
|
||||
class CustomComponent(Component):
|
||||
display_name: Optional[str] = None
|
||||
description: Optional[str] = None
|
||||
code: Optional[str] = None
|
||||
field_config: dict = {}
|
||||
code_class_base_inheritance: ClassVar[str] = "CustomComponent"
|
||||
function_entrypoint_name: ClassVar[str] = "build"
|
||||
function: Optional[Callable] = None
|
||||
repr_value: Optional[Any] = ""
|
||||
user_id: Optional[Union[UUID, str]] = None
|
||||
status: Optional[Any] = None
|
||||
_tree: Optional[dict] = None
|
||||
|
||||
def __init__(self, **data):
|
||||
self.cache = TTLCache(maxsize=1024, ttl=60)
|
||||
super().__init__(**data)
|
||||
|
||||
def custom_repr(self):
|
||||
if self.repr_value == "":
|
||||
self.repr_value = self.status
|
||||
if isinstance(self.repr_value, dict):
|
||||
return yaml.dump(self.repr_value)
|
||||
if isinstance(self.repr_value, str):
|
||||
return self.repr_value
|
||||
return str(self.repr_value)
|
||||
|
||||
def build_config(self):
|
||||
return self.field_config
|
||||
|
||||
def _class_template_validation(self, code: str):
|
||||
TYPE_HINT_LIST = ["Optional", "Prompt", "PromptTemplate", "LLMChain"]
|
||||
|
||||
if not code:
|
||||
raise HTTPException(
|
||||
status_code=400,
|
||||
detail={
|
||||
"error": self.ERROR_CODE_NULL,
|
||||
"traceback": "",
|
||||
},
|
||||
)
|
||||
|
||||
reader = DirectoryReader("", False)
|
||||
|
||||
for type_hint in TYPE_HINT_LIST:
|
||||
if reader._is_type_hint_used_in_args(type_hint, code) and not reader._is_type_hint_imported(
|
||||
type_hint, code
|
||||
):
|
||||
error_detail = {
|
||||
"error": "Type hint Error",
|
||||
"traceback": f"Type hint '{type_hint}' is used but not imported in the code.",
|
||||
}
|
||||
raise HTTPException(status_code=400, detail=error_detail)
|
||||
return True
|
||||
|
||||
def validate(self) -> bool:
|
||||
return self._class_template_validation(self.code) if self.code else False
|
||||
|
||||
@property
|
||||
def tree(self):
|
||||
return self.get_code_tree(self.code or "")
|
||||
|
||||
@property
|
||||
def get_function_entrypoint_args(self) -> list:
|
||||
build_method = self.get_build_method()
|
||||
if not build_method:
|
||||
return []
|
||||
|
||||
args = build_method["args"]
|
||||
for arg in args:
|
||||
if arg.get("type") == "prompt":
|
||||
raise HTTPException(
|
||||
status_code=400,
|
||||
detail={
|
||||
"error": "Type hint Error",
|
||||
"traceback": (
|
||||
"Prompt type is not supported in the build method." " Try using PromptTemplate instead."
|
||||
),
|
||||
},
|
||||
)
|
||||
elif not arg.get("type") and arg.get("name") != "self":
|
||||
# Set the type to Data
|
||||
arg["type"] = "Data"
|
||||
return args
|
||||
|
||||
@cachedmethod(operator.attrgetter("cache"))
|
||||
def get_build_method(self):
|
||||
if not self.code:
|
||||
return {}
|
||||
|
||||
component_classes = [cls for cls in self.tree["classes"] if self.code_class_base_inheritance in cls["bases"]]
|
||||
if not component_classes:
|
||||
return {}
|
||||
|
||||
# Assume the first Component class is the one we're interested in
|
||||
component_class = component_classes[0]
|
||||
build_methods = [
|
||||
method for method in component_class["methods"] if method["name"] == self.function_entrypoint_name
|
||||
]
|
||||
|
||||
return build_methods[0] if build_methods else {}
|
||||
|
||||
@property
|
||||
def get_function_entrypoint_return_type(self) -> List[Any]:
|
||||
build_method = self.get_build_method()
|
||||
if not build_method or not build_method.get("has_return"):
|
||||
return []
|
||||
return_type = build_method["return_type"]
|
||||
|
||||
# If list or List is in the return type, then we remove it and return the inner type
|
||||
if hasattr(return_type, "__origin__") and return_type.__origin__ in [list, List]:
|
||||
return_type = extract_inner_type_from_generic_alias(return_type)
|
||||
|
||||
# If the return type is not a Union, then we just return it as a list
|
||||
if not hasattr(return_type, "__origin__") or return_type.__origin__ != Union:
|
||||
return return_type if isinstance(return_type, list) else [return_type]
|
||||
# If the return type is a Union, then we need to parse itx
|
||||
return_type = extract_union_types_from_generic_alias(return_type)
|
||||
return return_type
|
||||
|
||||
@property
|
||||
def get_main_class_name(self):
|
||||
if not self.code:
|
||||
return ""
|
||||
|
||||
base_name = self.code_class_base_inheritance
|
||||
method_name = self.function_entrypoint_name
|
||||
|
||||
classes = []
|
||||
for item in self.tree.get("classes", []):
|
||||
if base_name in item["bases"]:
|
||||
method_names = [method["name"] for method in item["methods"]]
|
||||
if method_name in method_names:
|
||||
classes.append(item["name"])
|
||||
|
||||
# Get just the first item
|
||||
return next(iter(classes), "")
|
||||
|
||||
@property
|
||||
def template_config(self):
|
||||
return self.build_template_config()
|
||||
|
||||
def build_template_config(self):
|
||||
if not self.code:
|
||||
return {}
|
||||
|
||||
attributes = [
|
||||
main_class["attributes"]
|
||||
for main_class in self.tree.get("classes", [])
|
||||
if main_class["name"] == self.get_main_class_name
|
||||
]
|
||||
# Get just the first item
|
||||
attributes = next(iter(attributes), [])
|
||||
|
||||
return super().build_template_config(attributes)
|
||||
|
||||
@property
|
||||
def keys(self):
|
||||
def get_credential(name: str):
|
||||
if hasattr(self, "_user_id") and not self._user_id:
|
||||
raise ValueError(f"User id is not set for {self.__class__.__name__}")
|
||||
credential_service = get_credential_service() # Get service instance
|
||||
# Retrieve and decrypt the credential by name for the current user
|
||||
db_service = get_db_service()
|
||||
with session_getter(db_service) as session:
|
||||
return credential_service.get_credential(user_id=self._user_id or "", name=name, session=session)
|
||||
|
||||
return get_credential
|
||||
|
||||
def list_key_names(self):
|
||||
if hasattr(self, "_user_id") and not self._user_id:
|
||||
raise ValueError(f"User id is not set for {self.__class__.__name__}")
|
||||
credential_service = get_credential_service()
|
||||
db_service = get_db_service()
|
||||
with session_getter(db_service) as session:
|
||||
return credential_service.list_credentials(user_id=self._user_id, session=session)
|
||||
|
||||
def index(self, value: int = 0):
|
||||
"""Returns a function that returns the value at the given index in the iterable."""
|
||||
|
||||
def get_index(iterable: List[Any]):
|
||||
return iterable[value] if iterable else iterable
|
||||
|
||||
return get_index
|
||||
|
||||
@property
|
||||
def get_function(self):
|
||||
return validate.create_function(self.code, self.function_entrypoint_name)
|
||||
|
||||
async def load_flow(self, flow_id: str, tweaks: Optional[dict] = None) -> Any:
|
||||
from langflow.processing.process import (build_sorted_vertices,
|
||||
process_tweaks)
|
||||
|
||||
db_service = get_db_service()
|
||||
with session_getter(db_service) as session:
|
||||
graph_data = flow.data if (flow := session.get(Flow, flow_id)) else None
|
||||
if not graph_data:
|
||||
raise ValueError(f"Flow {flow_id} not found")
|
||||
if tweaks:
|
||||
graph_data = process_tweaks(graph_data=graph_data, tweaks=tweaks)
|
||||
return await build_sorted_vertices(graph_data, self.user_id)
|
||||
|
||||
def list_flows(self, *, get_session: Optional[Callable] = None) -> List[Flow]:
|
||||
if not self._user_id:
|
||||
raise ValueError("Session is invalid")
|
||||
try:
|
||||
get_session = get_session or session_getter
|
||||
db_service = get_db_service()
|
||||
with get_session(db_service) as session:
|
||||
flows = session.query(Flow).filter(Flow.user_id == self.user_id).all()
|
||||
return flows
|
||||
except Exception as e:
|
||||
raise ValueError("Session is invalid") from e
|
||||
|
||||
async def get_flow(
|
||||
self,
|
||||
*,
|
||||
flow_name: Optional[str] = None,
|
||||
flow_id: Optional[str] = None,
|
||||
tweaks: Optional[dict] = None,
|
||||
get_session: Optional[Callable] = None,
|
||||
) -> Flow:
|
||||
get_session = get_session or session_getter
|
||||
db_service = get_db_service()
|
||||
with get_session(db_service) as session:
|
||||
if flow_id:
|
||||
flow = session.query(Flow).get(flow_id)
|
||||
elif flow_name:
|
||||
flow = (session.query(Flow).filter(Flow.name == flow_name).filter(Flow.user_id == self.user_id)).first()
|
||||
else:
|
||||
raise ValueError("Either flow_name or flow_id must be provided")
|
||||
|
||||
if not flow:
|
||||
raise ValueError(f"Flow {flow_name or flow_id} not found")
|
||||
return await self.load_flow(flow.id, tweaks)
|
||||
|
||||
def build(self, *args: Any, **kwargs: Any) -> Any:
|
||||
raise NotImplementedError
|
||||
|
|
@ -0,0 +1,3 @@
|
|||
from .directory_reader import DirectoryReader
|
||||
|
||||
__all__ = ["DirectoryReader"]
|
||||
|
|
@ -0,0 +1,260 @@
|
|||
import ast
|
||||
import os
|
||||
import zlib
|
||||
|
||||
from loguru import logger
|
||||
|
||||
|
||||
class CustomComponentPathValueError(ValueError):
|
||||
pass
|
||||
|
||||
|
||||
class StringCompressor:
|
||||
def __init__(self, input_string):
|
||||
"""Initialize StringCompressor with a string to compress."""
|
||||
self.input_string = input_string
|
||||
|
||||
def compress_string(self):
|
||||
"""
|
||||
Compress the initial string and return the compressed data.
|
||||
"""
|
||||
# Convert string to bytes
|
||||
byte_data = self.input_string.encode("utf-8")
|
||||
# Compress the bytes
|
||||
self.compressed_data = zlib.compress(byte_data)
|
||||
|
||||
return self.compressed_data
|
||||
|
||||
def decompress_string(self):
|
||||
"""
|
||||
Decompress the compressed data and return the original string.
|
||||
"""
|
||||
# Decompress the bytes
|
||||
decompressed_data = zlib.decompress(self.compressed_data)
|
||||
# Convert bytes back to string
|
||||
return decompressed_data.decode("utf-8")
|
||||
|
||||
|
||||
class DirectoryReader:
|
||||
# Ensure the base path to read the files that contain
|
||||
# the custom components from this directory.
|
||||
base_path = ""
|
||||
|
||||
def __init__(self, directory_path, compress_code_field=False):
|
||||
"""
|
||||
Initialize DirectoryReader with a directory path
|
||||
and a flag indicating whether to compress the code.
|
||||
"""
|
||||
self.directory_path = directory_path
|
||||
self.compress_code_field = compress_code_field
|
||||
|
||||
def get_safe_path(self):
|
||||
"""Check if the path is valid and return it, or None if it's not."""
|
||||
return self.directory_path if self.is_valid_path() else None
|
||||
|
||||
def is_valid_path(self) -> bool:
|
||||
"""Check if the directory path is valid by comparing it to the base path."""
|
||||
fullpath = os.path.normpath(os.path.join(self.directory_path))
|
||||
return fullpath.startswith(self.base_path)
|
||||
|
||||
def is_empty_file(self, file_content):
|
||||
"""
|
||||
Check if the file content is empty.
|
||||
"""
|
||||
return len(file_content.strip()) == 0
|
||||
|
||||
def filter_loaded_components(self, data: dict, with_errors: bool) -> dict:
|
||||
from langflow.interface.custom.utils import build_component
|
||||
items = [
|
||||
{
|
||||
"name": menu["name"],
|
||||
"path": menu["path"],
|
||||
"components": [
|
||||
(*build_component(component),component)
|
||||
for component in menu["components"]
|
||||
if (component["error"] if with_errors else not component["error"])
|
||||
],
|
||||
}
|
||||
for menu in data["menu"]
|
||||
]
|
||||
filtered = [menu for menu in items if menu["components"]]
|
||||
logger.debug(f'Filtered components {"with errors" if with_errors else ""}: {len(filtered)}')
|
||||
return {"menu": filtered}
|
||||
|
||||
def validate_code(self, file_content):
|
||||
"""
|
||||
Validate the Python code by trying to parse it with ast.parse.
|
||||
"""
|
||||
try:
|
||||
ast.parse(file_content)
|
||||
return True
|
||||
except SyntaxError:
|
||||
return False
|
||||
|
||||
def validate_build(self, file_content):
|
||||
"""
|
||||
Check if the file content contains a function named 'build'.
|
||||
"""
|
||||
return "def build" in file_content
|
||||
|
||||
def read_file_content(self, file_path):
|
||||
"""
|
||||
Read and return the content of a file.
|
||||
"""
|
||||
if not os.path.isfile(file_path):
|
||||
return None
|
||||
with open(file_path, "r") as file:
|
||||
return file.read()
|
||||
|
||||
def get_files(self):
|
||||
"""
|
||||
Walk through the directory path and return a list of all .py files.
|
||||
"""
|
||||
if not (safe_path := self.get_safe_path()):
|
||||
raise CustomComponentPathValueError(f"The path needs to start with '{self.base_path}'.")
|
||||
|
||||
file_list = []
|
||||
for root, _, files in os.walk(safe_path):
|
||||
file_list.extend(
|
||||
os.path.join(root, filename)
|
||||
for filename in files
|
||||
if filename.endswith(".py") and not filename.startswith("__")
|
||||
)
|
||||
return file_list
|
||||
|
||||
def find_menu(self, response, menu_name):
|
||||
"""
|
||||
Find and return a menu by its name in the response.
|
||||
"""
|
||||
return next(
|
||||
(menu for menu in response["menu"] if menu["name"] == menu_name),
|
||||
None,
|
||||
)
|
||||
|
||||
def _is_type_hint_imported(self, type_hint_name: str, code: str) -> bool:
|
||||
"""
|
||||
Check if a specific type hint is imported
|
||||
from the typing module in the given code.
|
||||
"""
|
||||
module = ast.parse(code)
|
||||
|
||||
return any(
|
||||
isinstance(node, ast.ImportFrom)
|
||||
and node.module == "typing"
|
||||
and any(alias.name == type_hint_name for alias in node.names)
|
||||
for node in ast.walk(module)
|
||||
)
|
||||
|
||||
def _is_type_hint_used_in_args(self, type_hint_name: str, code: str) -> bool:
|
||||
"""
|
||||
Check if a specific type hint is used in the
|
||||
function definitions within the given code.
|
||||
"""
|
||||
try:
|
||||
module = ast.parse(code)
|
||||
|
||||
for node in ast.walk(module):
|
||||
if isinstance(node, ast.FunctionDef):
|
||||
for arg in node.args.args:
|
||||
if self._is_type_hint_in_arg_annotation(arg.annotation, type_hint_name):
|
||||
return True
|
||||
except SyntaxError:
|
||||
# Returns False if the code is not valid Python
|
||||
return False
|
||||
return False
|
||||
|
||||
def _is_type_hint_in_arg_annotation(self, annotation, type_hint_name: str) -> bool:
|
||||
"""
|
||||
Helper function to check if a type hint exists in an annotation.
|
||||
"""
|
||||
return (
|
||||
annotation is not None
|
||||
and isinstance(annotation, ast.Subscript)
|
||||
and isinstance(annotation.value, ast.Name)
|
||||
and annotation.value.id == type_hint_name
|
||||
)
|
||||
|
||||
def is_type_hint_used_but_not_imported(self, type_hint_name: str, code: str) -> bool:
|
||||
"""
|
||||
Check if a type hint is used but not imported in the given code.
|
||||
"""
|
||||
try:
|
||||
return self._is_type_hint_used_in_args(type_hint_name, code) and not self._is_type_hint_imported(
|
||||
type_hint_name, code
|
||||
)
|
||||
except SyntaxError:
|
||||
# Returns True if there's something wrong with the code
|
||||
# TODO : Find a better way to handle this
|
||||
return True
|
||||
|
||||
def process_file(self, file_path):
|
||||
"""
|
||||
Process a file by validating its content and
|
||||
returning the result and content/error message.
|
||||
"""
|
||||
file_content = self.read_file_content(file_path)
|
||||
|
||||
if file_content is None:
|
||||
return False, f"Could not read {file_path}"
|
||||
elif self.is_empty_file(file_content):
|
||||
return False, "Empty file"
|
||||
elif not self.validate_code(file_content):
|
||||
return False, "Syntax error"
|
||||
elif not self.validate_build(file_content):
|
||||
return False, "Missing build function"
|
||||
elif self._is_type_hint_used_in_args("Optional", file_content) and not self._is_type_hint_imported(
|
||||
"Optional", file_content
|
||||
):
|
||||
return (
|
||||
False,
|
||||
"Type hint 'Optional' is used but not imported in the code.",
|
||||
)
|
||||
else:
|
||||
if self.compress_code_field:
|
||||
file_content = str(StringCompressor(file_content).compress_string())
|
||||
return True, file_content
|
||||
|
||||
def build_component_menu_list(self, file_paths):
|
||||
"""
|
||||
Build a list of menus with their components
|
||||
from the .py files in the directory.
|
||||
"""
|
||||
response = {"menu": []}
|
||||
logger.debug("-------------------- Building component menu list --------------------")
|
||||
|
||||
for file_path in file_paths:
|
||||
menu_name = os.path.basename(os.path.dirname(file_path))
|
||||
logger.debug(f"Menu name: {menu_name}")
|
||||
filename = os.path.basename(file_path)
|
||||
validation_result, result_content = self.process_file(file_path)
|
||||
logger.debug(f"Validation result: {validation_result}")
|
||||
|
||||
menu_result = self.find_menu(response, menu_name) or {
|
||||
"name": menu_name,
|
||||
"path": os.path.dirname(file_path),
|
||||
"components": [],
|
||||
}
|
||||
component_name = filename.split(".")[0]
|
||||
# This is the name of the file which will be displayed in the UI
|
||||
# We need to change it from snake_case to CamelCase
|
||||
|
||||
# first check if it's already CamelCase
|
||||
if "_" in component_name:
|
||||
component_name_camelcase = " ".join(word.title() for word in component_name.split("_"))
|
||||
else:
|
||||
component_name_camelcase = component_name
|
||||
|
||||
component_info = {
|
||||
"name": "CustomComponent",
|
||||
"output_types": [component_name_camelcase],
|
||||
"file": filename,
|
||||
"code": result_content if validation_result else "",
|
||||
"error": "" if validation_result else result_content,
|
||||
}
|
||||
menu_result["components"].append(component_info)
|
||||
|
||||
logger.debug(f"Component info: {component_info}")
|
||||
if menu_result not in response["menu"]:
|
||||
response["menu"].append(menu_result)
|
||||
logger.debug("-------------------- Component menu list built --------------------")
|
||||
return response
|
||||
147
src/backend/langflow/interface/custom/directory_reader/utils.py
Normal file
147
src/backend/langflow/interface/custom/directory_reader/utils.py
Normal file
|
|
@ -0,0 +1,147 @@
|
|||
from langflow.interface.custom.directory_reader import DirectoryReader
|
||||
from langflow.template.frontend_node.custom_components import \
|
||||
CustomComponentFrontendNode
|
||||
from loguru import logger
|
||||
|
||||
|
||||
def merge_nested_dicts_with_renaming(dict1, dict2):
|
||||
for key, value in dict2.items():
|
||||
if key in dict1 and isinstance(value, dict) and isinstance(dict1.get(key), dict):
|
||||
for sub_key, sub_value in value.items():
|
||||
# if sub_key in dict1[key]:
|
||||
# new_key = get_new_key(dict1[key], sub_key)
|
||||
# dict1[key][new_key] = sub_value
|
||||
# else:
|
||||
dict1[key][sub_key] = sub_value
|
||||
else:
|
||||
dict1[key] = value
|
||||
return dict1
|
||||
|
||||
|
||||
def build_invalid_menu(invalid_components):
|
||||
"""Build the invalid menu."""
|
||||
if not invalid_components.get("menu"):
|
||||
return {}
|
||||
|
||||
logger.debug("------------------- INVALID COMPONENTS -------------------")
|
||||
invalid_menu = {}
|
||||
for menu_item in invalid_components["menu"]:
|
||||
menu_name = menu_item["name"]
|
||||
invalid_menu[menu_name] = build_invalid_menu_items(menu_item)
|
||||
return invalid_menu
|
||||
|
||||
|
||||
def build_valid_menu(valid_components):
|
||||
"""Build the valid menu."""
|
||||
valid_menu = {}
|
||||
logger.debug("------------------- VALID COMPONENTS -------------------")
|
||||
for menu_item in valid_components["menu"]:
|
||||
menu_name = menu_item["name"]
|
||||
valid_menu[menu_name] = build_menu_items(menu_item)
|
||||
return valid_menu
|
||||
|
||||
|
||||
def build_and_validate_all_files(reader: DirectoryReader, file_list):
|
||||
"""Build and validate all files"""
|
||||
data = reader.build_component_menu_list(file_list)
|
||||
|
||||
valid_components = reader.filter_loaded_components(data=data, with_errors=False)
|
||||
invalid_components = reader.filter_loaded_components(data=data, with_errors=True)
|
||||
|
||||
return valid_components, invalid_components
|
||||
|
||||
|
||||
def load_files_from_path(path: str):
|
||||
"""Load all files from a given path"""
|
||||
reader = DirectoryReader(path, False)
|
||||
|
||||
return reader.get_files()
|
||||
|
||||
|
||||
def build_custom_component_list_from_path(path: str):
|
||||
"""Build a list of custom components for the langchain from a given path"""
|
||||
file_list = load_files_from_path(path)
|
||||
reader = DirectoryReader(path, False)
|
||||
|
||||
valid_components, invalid_components = build_and_validate_all_files(reader, file_list)
|
||||
|
||||
valid_menu = build_valid_menu(valid_components)
|
||||
invalid_menu = build_invalid_menu(invalid_components)
|
||||
|
||||
return merge_nested_dicts_with_renaming(valid_menu, invalid_menu)
|
||||
|
||||
|
||||
def create_invalid_component_template(component, component_name):
|
||||
"""Create a template for an invalid component."""
|
||||
component_code = component["code"]
|
||||
component_frontend_node = CustomComponentFrontendNode(
|
||||
description="ERROR - Check your Python Code",
|
||||
display_name=f"ERROR - {component_name}",
|
||||
)
|
||||
|
||||
|
||||
component_frontend_node.error = component.get("error", None)
|
||||
field = component_frontend_node.template.get_field("code")
|
||||
field.value = component_code
|
||||
component_frontend_node.template.update_field("code", field)
|
||||
return component_frontend_node.model_dump(by_alias=True, exclude_none=True)
|
||||
|
||||
|
||||
def log_invalid_component_details(component):
|
||||
"""Log details of an invalid component."""
|
||||
logger.debug(component)
|
||||
logger.debug(f"Component Path: {component.get('path', None)}")
|
||||
logger.debug(f"Component Error: {component.get('error', None)}")
|
||||
|
||||
|
||||
def build_invalid_component(component):
|
||||
"""Build a single invalid component."""
|
||||
component_name = component["name"]
|
||||
component_template = create_invalid_component_template(component, component_name)
|
||||
log_invalid_component_details(component)
|
||||
return component_name, component_template
|
||||
|
||||
|
||||
def build_invalid_menu_items(menu_item):
|
||||
"""Build invalid menu items for a given menu."""
|
||||
menu_items = {}
|
||||
for component in menu_item["components"]:
|
||||
try:
|
||||
component_name, component_template = build_invalid_component(component)
|
||||
menu_items[component_name] = component_template
|
||||
logger.debug(f"Added {component_name} to invalid menu.")
|
||||
except Exception as exc:
|
||||
logger.exception(f"Error while creating custom component [{component_name}]: {str(exc)}")
|
||||
return menu_items
|
||||
|
||||
|
||||
def get_new_key(dictionary, original_key):
|
||||
counter = 1
|
||||
new_key = original_key + " (" + str(counter) + ")"
|
||||
while new_key in dictionary:
|
||||
counter += 1
|
||||
new_key = original_key + " (" + str(counter) + ")"
|
||||
return new_key
|
||||
|
||||
|
||||
def determine_component_name(component):
|
||||
"""Determine the name of the component."""
|
||||
component_output_types = component["output_types"]
|
||||
if len(component_output_types) == 1:
|
||||
return component_output_types[0]
|
||||
else:
|
||||
file_name = component.get("file").split(".")[0]
|
||||
return "".join(word.capitalize() for word in file_name.split("_")) if "_" in file_name else file_name
|
||||
|
||||
|
||||
def build_menu_items(menu_item):
|
||||
"""Build menu items for a given menu."""
|
||||
menu_items = {}
|
||||
for component_name, component_template, component in menu_item["components"]:
|
||||
try:
|
||||
menu_items[component_name] = component_template
|
||||
logger.debug(f"Added {component_name} to valid menu.")
|
||||
except Exception as exc:
|
||||
logger.error(f"Error loading Component: {component['output_types']}")
|
||||
logger.exception(f"Error while building custom component {component['output_types']}: {exc}")
|
||||
return menu_items
|
||||
|
|
@ -1,40 +1,383 @@
|
|||
import ast
|
||||
import contextlib
|
||||
import re
|
||||
from types import GenericAlias
|
||||
from typing import Any
|
||||
import traceback
|
||||
import warnings
|
||||
from typing import Any, Dict, List, Optional, Union
|
||||
from uuid import UUID
|
||||
|
||||
from fastapi import HTTPException
|
||||
from langflow.field_typing.range_spec import RangeSpec
|
||||
from langflow.interface.custom.code_parser.utils import extract_inner_type
|
||||
from langflow.interface.custom.custom_component import CustomComponent
|
||||
from langflow.interface.custom.directory_reader.utils import (
|
||||
build_custom_component_list_from_path, determine_component_name,
|
||||
merge_nested_dicts_with_renaming)
|
||||
from langflow.interface.importing.utils import eval_custom_component_code
|
||||
from langflow.template.field.base import TemplateField
|
||||
from langflow.template.frontend_node.custom_components import \
|
||||
CustomComponentFrontendNode
|
||||
from langflow.utils.util import get_base_classes
|
||||
from loguru import logger
|
||||
|
||||
|
||||
def extract_inner_type(return_type: str) -> str:
|
||||
def add_output_types(frontend_node: CustomComponentFrontendNode, return_types: List[str]):
|
||||
"""Add output types to the frontend node"""
|
||||
for return_type in return_types:
|
||||
if return_type is None:
|
||||
raise HTTPException(
|
||||
status_code=400,
|
||||
detail={
|
||||
"error": ("Invalid return type. Please check your code and try again."),
|
||||
"traceback": traceback.format_exc(),
|
||||
},
|
||||
)
|
||||
if hasattr(return_type, "__name__"):
|
||||
return_type = return_type.__name__
|
||||
elif hasattr(return_type, "__class__"):
|
||||
return_type = return_type.__class__.__name__
|
||||
else:
|
||||
return_type = str(return_type)
|
||||
|
||||
frontend_node.add_output_type(return_type)
|
||||
|
||||
|
||||
def add_base_classes(frontend_node: CustomComponentFrontendNode, return_types: List[str]):
|
||||
"""Add base classes to the frontend node"""
|
||||
for return_type_instance in return_types:
|
||||
if return_type_instance is None:
|
||||
raise HTTPException(
|
||||
status_code=400,
|
||||
detail={
|
||||
"error": ("Invalid return type. Please check your code and try again."),
|
||||
"traceback": traceback.format_exc(),
|
||||
},
|
||||
)
|
||||
|
||||
base_classes = get_base_classes(return_type_instance)
|
||||
|
||||
for base_class in base_classes:
|
||||
frontend_node.add_base_class(base_class)
|
||||
|
||||
|
||||
def extract_type_from_optional(field_type):
|
||||
"""
|
||||
Extracts the inner type from a type hint that is a list.
|
||||
Extract the type from a string formatted as "Optional[<type>]".
|
||||
|
||||
Parameters:
|
||||
field_type (str): The string from which to extract the type.
|
||||
|
||||
Returns:
|
||||
str: The extracted type, or an empty string if no type was found.
|
||||
"""
|
||||
if match := re.match(r"list\[(.*)\]", return_type, re.IGNORECASE):
|
||||
return match[1]
|
||||
return return_type
|
||||
match = re.search(r"\[(.*?)\]$", field_type)
|
||||
return match[1] if match else None
|
||||
|
||||
|
||||
def extract_inner_type_from_generic_alias(return_type: GenericAlias) -> Any:
|
||||
"""
|
||||
Extracts the inner type from a type hint that is a list.
|
||||
"""
|
||||
if return_type.__origin__ == list:
|
||||
return list(return_type.__args__)
|
||||
def get_field_properties(extra_field):
|
||||
"""Get the properties of an extra field"""
|
||||
field_name = extra_field["name"]
|
||||
field_type = extra_field.get("type", "str")
|
||||
field_value = extra_field.get("default", "")
|
||||
field_required = "optional" not in field_type.lower()
|
||||
|
||||
return return_type
|
||||
if not field_required:
|
||||
field_type = extract_type_from_optional(field_type)
|
||||
if field_value is not None:
|
||||
with contextlib.suppress(Exception):
|
||||
field_value = ast.literal_eval(field_value)
|
||||
return field_name, field_type, field_value, field_required
|
||||
|
||||
|
||||
def extract_union_types_from_generic_alias(return_type: GenericAlias) -> list:
|
||||
"""
|
||||
Extracts the inner type from a type hint that is a Union.
|
||||
"""
|
||||
return list(return_type.__args__)
|
||||
def process_type(field_type: str):
|
||||
if field_type.startswith("list") or field_type.startswith("List"):
|
||||
return extract_inner_type(field_type)
|
||||
return "prompt" if field_type == "Prompt" else field_type
|
||||
|
||||
|
||||
def add_new_custom_field(
|
||||
frontend_node: CustomComponentFrontendNode,
|
||||
field_name: str,
|
||||
field_type: str,
|
||||
field_value: Any,
|
||||
field_required: bool,
|
||||
field_config: dict,
|
||||
):
|
||||
# Check field_config if any of the keys are in it
|
||||
# if it is, update the value
|
||||
display_name = field_config.pop("display_name", field_name)
|
||||
field_type = field_config.pop("field_type", field_type)
|
||||
field_contains_list = "list" in field_type.lower()
|
||||
field_type = process_type(field_type)
|
||||
field_value = field_config.pop("value", field_value)
|
||||
field_advanced = field_config.pop("advanced", False)
|
||||
|
||||
if field_type == "bool" and field_value is None:
|
||||
field_value = False
|
||||
|
||||
# If options is a list, then it's a dropdown
|
||||
# If options is None, then it's a list of strings
|
||||
is_list = isinstance(field_config.get("options"), list)
|
||||
field_config["is_list"] = is_list or field_config.get("is_list", False) or field_contains_list
|
||||
|
||||
if "name" in field_config:
|
||||
warnings.warn("The 'name' key in field_config is used to build the object and can't be changed.")
|
||||
required = field_config.pop("required", field_required)
|
||||
placeholder = field_config.pop("placeholder", "")
|
||||
|
||||
new_field = TemplateField(
|
||||
name=field_name,
|
||||
field_type=field_type,
|
||||
value=field_value,
|
||||
show=True,
|
||||
required=required,
|
||||
advanced=field_advanced,
|
||||
placeholder=placeholder,
|
||||
display_name=display_name,
|
||||
**sanitize_field_config(field_config),
|
||||
)
|
||||
frontend_node.template.upsert_field(field_name, new_field)
|
||||
if isinstance(frontend_node.custom_fields, dict):
|
||||
frontend_node.custom_fields[field_name] = None
|
||||
|
||||
return frontend_node
|
||||
|
||||
|
||||
def add_extra_fields(frontend_node, field_config, function_args):
|
||||
"""Add extra fields to the frontend node"""
|
||||
if not function_args:
|
||||
return
|
||||
|
||||
# sort function_args which is a list of dicts
|
||||
function_args.sort(key=lambda x: x["name"])
|
||||
|
||||
for extra_field in function_args:
|
||||
if "name" not in extra_field or extra_field["name"] == "self":
|
||||
continue
|
||||
|
||||
field_name, field_type, field_value, field_required = get_field_properties(extra_field)
|
||||
config = field_config.get(field_name, {})
|
||||
frontend_node = add_new_custom_field(
|
||||
frontend_node,
|
||||
field_name,
|
||||
field_type,
|
||||
field_value,
|
||||
field_required,
|
||||
config,
|
||||
)
|
||||
|
||||
|
||||
def get_field_dict(field: Union[TemplateField, dict]):
|
||||
"""Get the field dictionary from a TemplateField or a dict"""
|
||||
if isinstance(field, TemplateField):
|
||||
return field.model_dump(by_alias=True, exclude_none=True)
|
||||
return field
|
||||
|
||||
|
||||
def run_build_config(
|
||||
custom_component: CustomComponent, user_id: Optional[Union[str, UUID]] = None, update_field=None
|
||||
):
|
||||
"""Build the field configuration for a custom component"""
|
||||
|
||||
try:
|
||||
if custom_component.code is None:
|
||||
return {}
|
||||
elif isinstance(custom_component.code, str):
|
||||
custom_class = eval_custom_component_code(custom_component.code)
|
||||
else:
|
||||
raise ValueError("Invalid code type")
|
||||
except Exception as exc:
|
||||
logger.error(f"Error while evaluating custom component code: {str(exc)}")
|
||||
raise HTTPException(
|
||||
status_code=400,
|
||||
detail={
|
||||
"error": ("Invalid type convertion. Please check your code and try again."),
|
||||
"traceback": traceback.format_exc(),
|
||||
},
|
||||
) from exc
|
||||
|
||||
try:
|
||||
build_config: Dict = custom_class(user_id=user_id).build_config()
|
||||
|
||||
for field_name, field in build_config.items():
|
||||
# Allow user to build TemplateField as well
|
||||
# as a dict with the same keys as TemplateField
|
||||
field_dict = get_field_dict(field)
|
||||
if update_field is not None and field_name != update_field:
|
||||
continue
|
||||
try:
|
||||
update_field_dict(field_dict)
|
||||
build_config[field_name] = field_dict
|
||||
except Exception as exc:
|
||||
logger.error(f"Error while getting build_config: {str(exc)}")
|
||||
|
||||
return build_config
|
||||
|
||||
except Exception as exc:
|
||||
logger.error(f"Error while building field config: {str(exc)}")
|
||||
raise HTTPException(
|
||||
status_code=400,
|
||||
detail={
|
||||
"error": ("Invalid type convertion. Please check your code and try again."),
|
||||
"traceback": traceback.format_exc(),
|
||||
},
|
||||
) from exc
|
||||
|
||||
|
||||
def sanitize_template_config(template_config):
|
||||
"""Sanitize the template config"""
|
||||
attributes = {
|
||||
"display_name",
|
||||
"description",
|
||||
"beta",
|
||||
"documentation",
|
||||
"output_types",
|
||||
}
|
||||
for key in template_config.copy():
|
||||
if key not in attributes:
|
||||
template_config.pop(key, None)
|
||||
|
||||
return template_config
|
||||
|
||||
|
||||
def build_frontend_node(template_config):
|
||||
"""Build a frontend node for a custom component"""
|
||||
try:
|
||||
sanitized_template_config = sanitize_template_config(template_config)
|
||||
return CustomComponentFrontendNode(**sanitized_template_config)
|
||||
except Exception as exc:
|
||||
logger.error(f"Error while building base frontend node: {exc}")
|
||||
raise exc
|
||||
|
||||
|
||||
def add_code_field(frontend_node: CustomComponentFrontendNode, raw_code, field_config):
|
||||
code_field = TemplateField(
|
||||
dynamic=True,
|
||||
required=True,
|
||||
placeholder="",
|
||||
multiline=True,
|
||||
value=raw_code,
|
||||
password=False,
|
||||
name="code",
|
||||
advanced=field_config.pop("advanced", False),
|
||||
field_type="code",
|
||||
is_list=False,
|
||||
)
|
||||
frontend_node.template.add_field(code_field)
|
||||
|
||||
return frontend_node
|
||||
|
||||
|
||||
def build_custom_component_template(
|
||||
custom_component: CustomComponent,
|
||||
user_id: Optional[Union[str, UUID]] = None,
|
||||
update_field: Optional[str] = None,
|
||||
) -> Optional[Dict[str, Any]]:
|
||||
"""Build a custom component template for the langchain"""
|
||||
try:
|
||||
logger.debug("Building custom component template")
|
||||
frontend_node = build_frontend_node(custom_component.template_config)
|
||||
|
||||
logger.debug("Built base frontend node")
|
||||
|
||||
logger.debug("Updated attributes")
|
||||
field_config = run_build_config(custom_component, user_id=user_id, update_field=update_field)
|
||||
logger.debug("Built field config")
|
||||
entrypoint_args = custom_component.get_function_entrypoint_args
|
||||
|
||||
add_extra_fields(frontend_node, field_config, entrypoint_args)
|
||||
logger.debug("Added extra fields")
|
||||
frontend_node = add_code_field(frontend_node, custom_component.code, field_config.get("code", {}))
|
||||
logger.debug("Added code field")
|
||||
add_base_classes(frontend_node, custom_component.get_function_entrypoint_return_type)
|
||||
add_output_types(frontend_node, custom_component.get_function_entrypoint_return_type)
|
||||
logger.debug("Added base classes")
|
||||
return frontend_node.to_dict(add_name=False)
|
||||
except Exception as exc:
|
||||
if isinstance(exc, HTTPException):
|
||||
raise exc
|
||||
raise HTTPException(
|
||||
status_code=400,
|
||||
detail={
|
||||
"error": ("Invalid type convertion. Please check your code and try again."),
|
||||
"traceback": traceback.format_exc(),
|
||||
},
|
||||
) from exc
|
||||
|
||||
|
||||
def create_component_template(component):
|
||||
"""Create a template for a component."""
|
||||
component_code = component["code"]
|
||||
component_output_types = component["output_types"]
|
||||
|
||||
component_extractor = CustomComponent(code=component_code)
|
||||
component_extractor.validate()
|
||||
|
||||
component_template = build_custom_component_template(component_extractor)
|
||||
component_template["output_types"] = component_output_types
|
||||
return component_template
|
||||
|
||||
|
||||
def build_custom_components(settings_service):
|
||||
"""Build custom components from the specified paths."""
|
||||
if not settings_service.settings.COMPONENTS_PATH:
|
||||
return {}
|
||||
|
||||
logger.info(f"Building custom components from {settings_service.settings.COMPONENTS_PATH}")
|
||||
custom_components_from_file = {}
|
||||
processed_paths = set()
|
||||
for path in settings_service.settings.COMPONENTS_PATH:
|
||||
path_str = str(path)
|
||||
if path_str in processed_paths:
|
||||
continue
|
||||
|
||||
custom_component_dict = build_custom_component_list_from_path(path_str)
|
||||
if custom_component_dict:
|
||||
category = next(iter(custom_component_dict))
|
||||
logger.info(f"Loading {len(custom_component_dict[category])} component(s) from category {category}")
|
||||
custom_components_from_file = merge_nested_dicts_with_renaming(
|
||||
custom_components_from_file, custom_component_dict
|
||||
)
|
||||
processed_paths.add(path_str)
|
||||
|
||||
return custom_components_from_file
|
||||
|
||||
|
||||
def create_and_validate_component(code: str) -> CustomComponent:
|
||||
component = CustomComponent(code=code)
|
||||
component.validate()
|
||||
return component
|
||||
|
||||
|
||||
def update_field_dict(field_dict):
|
||||
"""Update the field dictionary by calling options() or value() if they are callable"""
|
||||
if "options" in field_dict and callable(field_dict["options"]):
|
||||
field_dict["options"] = field_dict["options"]()
|
||||
# Also update the "refresh" key
|
||||
field_dict["refresh"] = True
|
||||
|
||||
if "value" in field_dict and callable(field_dict["value"]):
|
||||
field_dict["value"] = field_dict["value"](field_dict.get("options", []))
|
||||
field_dict["refresh"] = True
|
||||
|
||||
# Let's check if "range_spec" is a RangeSpec object
|
||||
if "rangeSpec" in field_dict and isinstance(field_dict["rangeSpec"], RangeSpec):
|
||||
field_dict["rangeSpec"] = field_dict["rangeSpec"].model_dump()
|
||||
|
||||
|
||||
def sanitize_field_config(field_config: Dict):
|
||||
# If any of the already existing keys are in field_config, remove them
|
||||
for key in ["name", "field_type", "value", "required", "placeholder", "display_name", "advanced", "show"]:
|
||||
field_config.pop(key, None)
|
||||
return field_config
|
||||
|
||||
|
||||
def build_component(component):
|
||||
"""Build a single component."""
|
||||
logger.debug(f"Building component: {component.get('name'), component.get('output_types')}")
|
||||
component_name = determine_component_name(component)
|
||||
component_template = create_component_template(component)
|
||||
return component_name, component_template
|
||||
|
||||
|
||||
|
||||
def extract_union_types(return_type: str) -> list[str]:
|
||||
"""
|
||||
Extracts the inner type from a type hint that is a list.
|
||||
"""
|
||||
# If the return type is a Union, then we need to parse it
|
||||
return_type = return_type.replace("Union", "").replace("[", "").replace("]", "")
|
||||
return_types = return_type.split(",")
|
||||
return_types = [item.strip() for item in return_types]
|
||||
return return_types
|
||||
|
|
|
|||
|
|
@ -1,24 +1,12 @@
|
|||
import ast
|
||||
import contextlib
|
||||
import re
|
||||
import traceback
|
||||
import warnings
|
||||
from typing import Any, Dict, List, Optional, Union
|
||||
from uuid import UUID
|
||||
|
||||
from cachetools import LRUCache, cached
|
||||
from fastapi import HTTPException
|
||||
from loguru import logger
|
||||
|
||||
from langflow.field_typing.range_spec import RangeSpec
|
||||
from langflow.interface.agents.base import agent_creator
|
||||
from langflow.interface.chains.base import chain_creator
|
||||
from langflow.interface.custom.custom_component import CustomComponent
|
||||
from langflow.interface.custom.directory_reader import DirectoryReader
|
||||
from langflow.interface.custom.utils import extract_inner_type
|
||||
from langflow.interface.custom.directory_reader.utils import \
|
||||
merge_nested_dicts_with_renaming
|
||||
from langflow.interface.custom.utils import build_custom_components
|
||||
from langflow.interface.document_loaders.base import documentloader_creator
|
||||
from langflow.interface.embeddings.base import embedding_creator
|
||||
from langflow.interface.importing.utils import eval_custom_component_code
|
||||
from langflow.interface.llms.base import llm_creator
|
||||
from langflow.interface.memories.base import memory_creator
|
||||
from langflow.interface.output_parsers.base import output_parser_creator
|
||||
|
|
@ -30,9 +18,6 @@ from langflow.interface.tools.base import tool_creator
|
|||
from langflow.interface.utilities.base import utility_creator
|
||||
from langflow.interface.vector_store.base import vectorstore_creator
|
||||
from langflow.interface.wrappers.base import wrapper_creator
|
||||
from langflow.template.field.base import TemplateField
|
||||
from langflow.template.frontend_node.custom_components import CustomComponentFrontendNode
|
||||
from langflow.utils.util import get_base_classes
|
||||
|
||||
|
||||
# Used to get the base_classes list
|
||||
|
|
@ -80,481 +65,6 @@ def build_langchain_types_dict(): # sourcery skip: dict-assign-update-to-union
|
|||
return all_types
|
||||
|
||||
|
||||
def process_type(field_type: str):
|
||||
if field_type.startswith("list") or field_type.startswith("List"):
|
||||
return extract_inner_type(field_type)
|
||||
return "prompt" if field_type == "Prompt" else field_type
|
||||
|
||||
|
||||
# TODO: Move to correct place
|
||||
def add_new_custom_field(
|
||||
frontend_node: CustomComponentFrontendNode,
|
||||
field_name: str,
|
||||
field_type: str,
|
||||
field_value: Any,
|
||||
field_required: bool,
|
||||
field_config: dict,
|
||||
):
|
||||
# Check field_config if any of the keys are in it
|
||||
# if it is, update the value
|
||||
display_name = field_config.pop("display_name", field_name)
|
||||
field_type = field_config.pop("field_type", field_type)
|
||||
field_contains_list = "list" in field_type.lower()
|
||||
field_type = process_type(field_type)
|
||||
field_value = field_config.pop("value", field_value)
|
||||
field_advanced = field_config.pop("advanced", False)
|
||||
|
||||
if field_type == "bool" and field_value is None:
|
||||
field_value = False
|
||||
|
||||
# If options is a list, then it's a dropdown
|
||||
# If options is None, then it's a list of strings
|
||||
is_list = isinstance(field_config.get("options"), list)
|
||||
field_config["is_list"] = is_list or field_config.get("is_list", False) or field_contains_list
|
||||
|
||||
if "name" in field_config:
|
||||
warnings.warn("The 'name' key in field_config is used to build the object and can't be changed.")
|
||||
required = field_config.pop("required", field_required)
|
||||
placeholder = field_config.pop("placeholder", "")
|
||||
|
||||
new_field = TemplateField(
|
||||
name=field_name,
|
||||
field_type=field_type,
|
||||
value=field_value,
|
||||
show=True,
|
||||
required=required,
|
||||
advanced=field_advanced,
|
||||
placeholder=placeholder,
|
||||
display_name=display_name,
|
||||
**sanitize_field_config(field_config),
|
||||
)
|
||||
frontend_node.template.upsert_field(field_name, new_field)
|
||||
if isinstance(frontend_node.custom_fields, dict):
|
||||
frontend_node.custom_fields[field_name] = None
|
||||
|
||||
return frontend_node
|
||||
|
||||
|
||||
def sanitize_field_config(field_config: Dict):
|
||||
# If any of the already existing keys are in field_config, remove them
|
||||
for key in ["name", "field_type", "value", "required", "placeholder", "display_name", "advanced", "show"]:
|
||||
field_config.pop(key, None)
|
||||
return field_config
|
||||
|
||||
|
||||
# TODO: Move to correct place
|
||||
def add_code_field(frontend_node: CustomComponentFrontendNode, raw_code, field_config):
|
||||
code_field = TemplateField(
|
||||
dynamic=True,
|
||||
required=True,
|
||||
placeholder="",
|
||||
multiline=True,
|
||||
value=raw_code,
|
||||
password=False,
|
||||
name="code",
|
||||
advanced=field_config.pop("advanced", False),
|
||||
field_type="code",
|
||||
is_list=False,
|
||||
)
|
||||
frontend_node.template.add_field(code_field)
|
||||
|
||||
return frontend_node
|
||||
|
||||
|
||||
def extract_type_from_optional(field_type):
|
||||
"""
|
||||
Extract the type from a string formatted as "Optional[<type>]".
|
||||
|
||||
Parameters:
|
||||
field_type (str): The string from which to extract the type.
|
||||
|
||||
Returns:
|
||||
str: The extracted type, or an empty string if no type was found.
|
||||
"""
|
||||
match = re.search(r"\[(.*?)\]$", field_type)
|
||||
return match[1] if match else None
|
||||
|
||||
|
||||
def build_frontend_node(template_config):
|
||||
"""Build a frontend node for a custom component"""
|
||||
try:
|
||||
sanitized_template_config = sanitize_template_config(template_config)
|
||||
return CustomComponentFrontendNode(**sanitized_template_config)
|
||||
except Exception as exc:
|
||||
logger.error(f"Error while building base frontend node: {exc}")
|
||||
raise exc
|
||||
|
||||
|
||||
def sanitize_template_config(template_config):
|
||||
"""Sanitize the template config"""
|
||||
attributes = {
|
||||
"display_name",
|
||||
"description",
|
||||
"beta",
|
||||
"documentation",
|
||||
"output_types",
|
||||
}
|
||||
for key in template_config.copy():
|
||||
if key not in attributes:
|
||||
template_config.pop(key, None)
|
||||
|
||||
return template_config
|
||||
|
||||
|
||||
def build_field_config(
|
||||
custom_component: CustomComponent, user_id: Optional[Union[str, UUID]] = None, update_field=None
|
||||
):
|
||||
"""Build the field configuration for a custom component"""
|
||||
|
||||
try:
|
||||
if custom_component.code is None:
|
||||
return {}
|
||||
elif isinstance(custom_component.code, str):
|
||||
custom_class = eval_custom_component_code(custom_component.code)
|
||||
else:
|
||||
raise ValueError("Invalid code type")
|
||||
except Exception as exc:
|
||||
logger.error(f"Error while evaluating custom component code: {str(exc)}")
|
||||
raise HTTPException(
|
||||
status_code=400,
|
||||
detail={
|
||||
"error": ("Invalid type convertion. Please check your code and try again."),
|
||||
"traceback": traceback.format_exc(),
|
||||
},
|
||||
) from exc
|
||||
|
||||
try:
|
||||
build_config: Dict = custom_class(user_id=user_id).build_config()
|
||||
|
||||
for field_name, field in build_config.items():
|
||||
# Allow user to build TemplateField as well
|
||||
# as a dict with the same keys as TemplateField
|
||||
field_dict = get_field_dict(field)
|
||||
if update_field is not None and field_name != update_field:
|
||||
continue
|
||||
try:
|
||||
update_field_dict(field_dict)
|
||||
build_config[field_name] = field_dict
|
||||
except Exception as exc:
|
||||
logger.error(f"Error while getting build_config: {str(exc)}")
|
||||
|
||||
return build_config
|
||||
|
||||
except Exception as exc:
|
||||
logger.error(f"Error while building field config: {str(exc)}")
|
||||
raise HTTPException(
|
||||
status_code=400,
|
||||
detail={
|
||||
"error": ("Invalid type convertion. Please check your code and try again."),
|
||||
"traceback": traceback.format_exc(),
|
||||
},
|
||||
) from exc
|
||||
|
||||
|
||||
def get_field_dict(field):
|
||||
"""Get the field dictionary from a TemplateField or a dict"""
|
||||
if isinstance(field, TemplateField):
|
||||
return field.model_dump(by_alias=True, exclude_none=True)
|
||||
return field
|
||||
|
||||
|
||||
def update_field_dict(field_dict):
|
||||
"""Update the field dictionary by calling options() or value() if they are callable"""
|
||||
if "options" in field_dict and callable(field_dict["options"]):
|
||||
field_dict["options"] = field_dict["options"]()
|
||||
# Also update the "refresh" key
|
||||
field_dict["refresh"] = True
|
||||
|
||||
if "value" in field_dict and callable(field_dict["value"]):
|
||||
field_dict["value"] = field_dict["value"](field_dict.get("options", []))
|
||||
field_dict["refresh"] = True
|
||||
|
||||
# Let's check if "range_spec" is a RangeSpec object
|
||||
if "rangeSpec" in field_dict and isinstance(field_dict["rangeSpec"], RangeSpec):
|
||||
field_dict["rangeSpec"] = field_dict["rangeSpec"].model_dump()
|
||||
|
||||
|
||||
def add_extra_fields(frontend_node, field_config, function_args):
|
||||
"""Add extra fields to the frontend node"""
|
||||
if not function_args:
|
||||
return
|
||||
|
||||
# sort function_args which is a list of dicts
|
||||
function_args.sort(key=lambda x: x["name"])
|
||||
|
||||
for extra_field in function_args:
|
||||
if "name" not in extra_field or extra_field["name"] == "self":
|
||||
continue
|
||||
|
||||
field_name, field_type, field_value, field_required = get_field_properties(extra_field)
|
||||
config = field_config.get(field_name, {})
|
||||
frontend_node = add_new_custom_field(
|
||||
frontend_node,
|
||||
field_name,
|
||||
field_type,
|
||||
field_value,
|
||||
field_required,
|
||||
config,
|
||||
)
|
||||
|
||||
|
||||
def get_field_properties(extra_field):
|
||||
"""Get the properties of an extra field"""
|
||||
field_name = extra_field["name"]
|
||||
field_type = extra_field.get("type", "str")
|
||||
field_value = extra_field.get("default", "")
|
||||
field_required = "optional" not in field_type.lower()
|
||||
|
||||
if not field_required:
|
||||
field_type = extract_type_from_optional(field_type)
|
||||
if field_value is not None:
|
||||
with contextlib.suppress(Exception):
|
||||
field_value = ast.literal_eval(field_value)
|
||||
return field_name, field_type, field_value, field_required
|
||||
|
||||
|
||||
def add_base_classes(frontend_node: CustomComponentFrontendNode, return_types: List[str]):
|
||||
"""Add base classes to the frontend node"""
|
||||
for return_type_instance in return_types:
|
||||
if return_type_instance is None:
|
||||
raise HTTPException(
|
||||
status_code=400,
|
||||
detail={
|
||||
"error": ("Invalid return type. Please check your code and try again."),
|
||||
"traceback": traceback.format_exc(),
|
||||
},
|
||||
)
|
||||
|
||||
base_classes = get_base_classes(return_type_instance)
|
||||
|
||||
for base_class in base_classes:
|
||||
frontend_node.add_base_class(base_class)
|
||||
|
||||
|
||||
def add_output_types(frontend_node: CustomComponentFrontendNode, return_types: List[str]):
|
||||
"""Add output types to the frontend node"""
|
||||
for return_type in return_types:
|
||||
if return_type is None:
|
||||
raise HTTPException(
|
||||
status_code=400,
|
||||
detail={
|
||||
"error": ("Invalid return type. Please check your code and try again."),
|
||||
"traceback": traceback.format_exc(),
|
||||
},
|
||||
)
|
||||
if hasattr(return_type, "__name__"):
|
||||
return_type = return_type.__name__
|
||||
elif hasattr(return_type, "__class__"):
|
||||
return_type = return_type.__class__.__name__
|
||||
else:
|
||||
return_type = str(return_type)
|
||||
|
||||
frontend_node.add_output_type(return_type)
|
||||
|
||||
|
||||
def build_custom_component_template(
|
||||
custom_component: CustomComponent,
|
||||
user_id: Optional[Union[str, UUID]] = None,
|
||||
update_field: Optional[str] = None,
|
||||
) -> Optional[Dict[str, Any]]:
|
||||
"""Build a custom component template for the langchain"""
|
||||
try:
|
||||
logger.debug("Building custom component template")
|
||||
frontend_node = build_frontend_node(custom_component.template_config)
|
||||
|
||||
logger.debug("Built base frontend node")
|
||||
|
||||
logger.debug("Updated attributes")
|
||||
field_config = build_field_config(custom_component, user_id=user_id, update_field=update_field)
|
||||
logger.debug("Built field config")
|
||||
entrypoint_args = custom_component.get_function_entrypoint_args
|
||||
|
||||
add_extra_fields(frontend_node, field_config, entrypoint_args)
|
||||
logger.debug("Added extra fields")
|
||||
frontend_node = add_code_field(frontend_node, custom_component.code, field_config.get("code", {}))
|
||||
logger.debug("Added code field")
|
||||
add_base_classes(frontend_node, custom_component.get_function_entrypoint_return_type)
|
||||
add_output_types(frontend_node, custom_component.get_function_entrypoint_return_type)
|
||||
logger.debug("Added base classes")
|
||||
return frontend_node.to_dict(add_name=False)
|
||||
except Exception as exc:
|
||||
if isinstance(exc, HTTPException):
|
||||
raise exc
|
||||
raise HTTPException(
|
||||
status_code=400,
|
||||
detail={
|
||||
"error": ("Invalid type convertion. Please check your code and try again."),
|
||||
"traceback": traceback.format_exc(),
|
||||
},
|
||||
) from exc
|
||||
|
||||
|
||||
def load_files_from_path(path: str):
|
||||
"""Load all files from a given path"""
|
||||
reader = DirectoryReader(path, False)
|
||||
|
||||
return reader.get_files()
|
||||
|
||||
|
||||
def build_and_validate_all_files(reader: DirectoryReader, file_list):
|
||||
"""Build and validate all files"""
|
||||
data = reader.build_component_menu_list(file_list)
|
||||
|
||||
valid_components = reader.filter_loaded_components(data=data, with_errors=False)
|
||||
invalid_components = reader.filter_loaded_components(data=data, with_errors=True)
|
||||
|
||||
return valid_components, invalid_components
|
||||
|
||||
|
||||
def build_valid_menu(valid_components):
|
||||
"""Build the valid menu."""
|
||||
valid_menu = {}
|
||||
logger.debug("------------------- VALID COMPONENTS -------------------")
|
||||
for menu_item in valid_components["menu"]:
|
||||
menu_name = menu_item["name"]
|
||||
valid_menu[menu_name] = build_menu_items(menu_item)
|
||||
return valid_menu
|
||||
|
||||
|
||||
def build_menu_items(menu_item):
|
||||
"""Build menu items for a given menu."""
|
||||
menu_items = {}
|
||||
for component in menu_item["components"]:
|
||||
try:
|
||||
component_name, component_template = build_component(component)
|
||||
menu_items[component_name] = component_template
|
||||
logger.debug(f"Added {component_name} to valid menu.")
|
||||
except Exception as exc:
|
||||
logger.error(f"Error loading Component: {component['output_types']}")
|
||||
logger.exception(f"Error while building custom component {component['output_types']}: {exc}")
|
||||
return menu_items
|
||||
|
||||
|
||||
def build_component(component):
|
||||
"""Build a single component."""
|
||||
logger.debug(f"Building component: {component.get('name'), component.get('output_types')}")
|
||||
component_name = determine_component_name(component)
|
||||
component_template = create_component_template(component)
|
||||
return component_name, component_template
|
||||
|
||||
|
||||
def determine_component_name(component):
|
||||
"""Determine the name of the component."""
|
||||
component_output_types = component["output_types"]
|
||||
if len(component_output_types) == 1:
|
||||
return component_output_types[0]
|
||||
else:
|
||||
file_name = component.get("file").split(".")[0]
|
||||
return "".join(word.capitalize() for word in file_name.split("_")) if "_" in file_name else file_name
|
||||
|
||||
|
||||
def create_component_template(component):
|
||||
"""Create a template for a component."""
|
||||
component_code = component["code"]
|
||||
component_output_types = component["output_types"]
|
||||
|
||||
component_extractor = CustomComponent(code=component_code)
|
||||
component_extractor.validate()
|
||||
|
||||
component_template = build_custom_component_template(component_extractor)
|
||||
component_template["output_types"] = component_output_types
|
||||
return component_template
|
||||
|
||||
|
||||
def build_invalid_menu(invalid_components):
|
||||
"""Build the invalid menu."""
|
||||
if not invalid_components.get("menu"):
|
||||
return {}
|
||||
|
||||
logger.debug("------------------- INVALID COMPONENTS -------------------")
|
||||
invalid_menu = {}
|
||||
for menu_item in invalid_components["menu"]:
|
||||
menu_name = menu_item["name"]
|
||||
invalid_menu[menu_name] = build_invalid_menu_items(menu_item)
|
||||
return invalid_menu
|
||||
|
||||
|
||||
def build_invalid_menu_items(menu_item):
|
||||
"""Build invalid menu items for a given menu."""
|
||||
menu_items = {}
|
||||
for component in menu_item["components"]:
|
||||
try:
|
||||
component_name, component_template = build_invalid_component(component)
|
||||
menu_items[component_name] = component_template
|
||||
logger.debug(f"Added {component_name} to invalid menu.")
|
||||
except Exception as exc:
|
||||
logger.exception(f"Error while creating custom component [{component_name}]: {str(exc)}")
|
||||
return menu_items
|
||||
|
||||
|
||||
def build_invalid_component(component):
|
||||
"""Build a single invalid component."""
|
||||
component_name = component["name"]
|
||||
component_template = create_invalid_component_template(component, component_name)
|
||||
log_invalid_component_details(component)
|
||||
return component_name, component_template
|
||||
|
||||
|
||||
def create_invalid_component_template(component, component_name):
|
||||
"""Create a template for an invalid component."""
|
||||
component_code = component["code"]
|
||||
component_template = (
|
||||
CustomComponentFrontendNode(
|
||||
description="ERROR - Check your Python Code",
|
||||
display_name=f"ERROR - {component_name}",
|
||||
)
|
||||
.to_dict()
|
||||
.get(type(CustomComponent()).__name__)
|
||||
)
|
||||
|
||||
component_template["error"] = component.get("error", None)
|
||||
component_template.get("template").get("code")["value"] = component_code
|
||||
return component_template
|
||||
|
||||
|
||||
def log_invalid_component_details(component):
|
||||
"""Log details of an invalid component."""
|
||||
logger.debug(component)
|
||||
logger.debug(f"Component Path: {component.get('path', None)}")
|
||||
logger.debug(f"Component Error: {component.get('error', None)}")
|
||||
|
||||
|
||||
def get_new_key(dictionary, original_key):
|
||||
counter = 1
|
||||
new_key = original_key + " (" + str(counter) + ")"
|
||||
while new_key in dictionary:
|
||||
counter += 1
|
||||
new_key = original_key + " (" + str(counter) + ")"
|
||||
return new_key
|
||||
|
||||
|
||||
def merge_nested_dicts_with_renaming(dict1, dict2):
|
||||
for key, value in dict2.items():
|
||||
if key in dict1 and isinstance(value, dict) and isinstance(dict1.get(key), dict):
|
||||
for sub_key, sub_value in value.items():
|
||||
# if sub_key in dict1[key]:
|
||||
# new_key = get_new_key(dict1[key], sub_key)
|
||||
# dict1[key][new_key] = sub_value
|
||||
# else:
|
||||
dict1[key][sub_key] = sub_value
|
||||
else:
|
||||
dict1[key] = value
|
||||
return dict1
|
||||
|
||||
|
||||
def build_custom_component_list_from_path(path: str):
|
||||
"""Build a list of custom components for the langchain from a given path"""
|
||||
file_list = load_files_from_path(path)
|
||||
reader = DirectoryReader(path, False)
|
||||
|
||||
valid_components, invalid_components = build_and_validate_all_files(reader, file_list)
|
||||
|
||||
valid_menu = build_valid_menu(valid_components)
|
||||
invalid_menu = build_invalid_menu(invalid_components)
|
||||
|
||||
return merge_nested_dicts_with_renaming(valid_menu, invalid_menu)
|
||||
|
||||
|
||||
def get_all_types_dict(settings_service):
|
||||
"""Get all types dictionary combining native and custom components."""
|
||||
native_components = build_langchain_types_dict()
|
||||
|
|
@ -562,41 +72,3 @@ def get_all_types_dict(settings_service):
|
|||
return merge_nested_dicts_with_renaming(native_components, custom_components_from_file)
|
||||
|
||||
|
||||
def build_custom_components(settings_service):
|
||||
"""Build custom components from the specified paths."""
|
||||
if not settings_service.settings.COMPONENTS_PATH:
|
||||
return {}
|
||||
|
||||
logger.info(f"Building custom components from {settings_service.settings.COMPONENTS_PATH}")
|
||||
custom_components_from_file = {}
|
||||
processed_paths = set()
|
||||
for path in settings_service.settings.COMPONENTS_PATH:
|
||||
path_str = str(path)
|
||||
if path_str in processed_paths:
|
||||
continue
|
||||
|
||||
custom_component_dict = build_custom_component_list_from_path(path_str)
|
||||
if custom_component_dict:
|
||||
category = next(iter(custom_component_dict))
|
||||
logger.info(f"Loading {len(custom_component_dict[category])} component(s) from category {category}")
|
||||
custom_components_from_file = merge_nested_dicts_with_renaming(
|
||||
custom_components_from_file, custom_component_dict
|
||||
)
|
||||
processed_paths.add(path_str)
|
||||
|
||||
return custom_components_from_file
|
||||
|
||||
|
||||
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
|
||||
|
||||
|
||||
def create_and_validate_component(code: str) -> CustomComponent:
|
||||
component = CustomComponent(code=code)
|
||||
component.validate()
|
||||
return component
|
||||
|
|
|
|||
|
|
@ -4,11 +4,13 @@ from uuid import uuid4
|
|||
|
||||
import pytest
|
||||
from fastapi import HTTPException
|
||||
|
||||
from langflow.interface.custom.base import CustomComponent
|
||||
from langflow.interface.custom.code_parser import CodeParser, CodeSyntaxError
|
||||
from langflow.interface.custom.component import Component, ComponentCodeNullError
|
||||
from langflow.interface.types import build_custom_component_template, create_and_validate_component
|
||||
from langflow.interface.custom.code_parser.code_parser import (CodeParser,
|
||||
CodeSyntaxError)
|
||||
from langflow.interface.custom.custom_component.component import (
|
||||
Component, ComponentCodeNullError)
|
||||
from langflow.interface.custom.utils import (build_custom_component_template,
|
||||
create_and_validate_component)
|
||||
from langflow.services.database.models.flow import Flow, FlowCreate
|
||||
|
||||
code_default = """
|
||||
|
|
@ -49,7 +51,7 @@ def test_code_parser_get_tree():
|
|||
Test the __get_tree method of the CodeParser class.
|
||||
"""
|
||||
parser = CodeParser(code_default)
|
||||
tree = parser._CodeParser__get_tree()
|
||||
tree = parser.get_tree()
|
||||
assert isinstance(tree, ast.AST)
|
||||
|
||||
|
||||
|
|
@ -62,7 +64,7 @@ def test_code_parser_syntax_error():
|
|||
|
||||
parser = CodeParser(code_syntax_error)
|
||||
with pytest.raises(CodeSyntaxError):
|
||||
parser._CodeParser__get_tree()
|
||||
parser.get_tree()
|
||||
|
||||
|
||||
def test_component_init():
|
||||
|
|
@ -139,7 +141,7 @@ def test_code_parser_parse_imports_import():
|
|||
class with an import statement.
|
||||
"""
|
||||
parser = CodeParser(code_default)
|
||||
tree = parser._CodeParser__get_tree()
|
||||
tree = parser.get_tree()
|
||||
for node in ast.walk(tree):
|
||||
if isinstance(node, ast.Import):
|
||||
parser.parse_imports(node)
|
||||
|
|
@ -152,7 +154,7 @@ def test_code_parser_parse_imports_importfrom():
|
|||
class with an import from statement.
|
||||
"""
|
||||
parser = CodeParser("from os import path")
|
||||
tree = parser._CodeParser__get_tree()
|
||||
tree = parser.get_tree()
|
||||
for node in ast.walk(tree):
|
||||
if isinstance(node, ast.ImportFrom):
|
||||
parser.parse_imports(node)
|
||||
|
|
@ -164,7 +166,7 @@ def test_code_parser_parse_functions():
|
|||
Test the parse_functions method of the CodeParser class.
|
||||
"""
|
||||
parser = CodeParser("def test(): pass")
|
||||
tree = parser._CodeParser__get_tree()
|
||||
tree = parser.get_tree()
|
||||
for node in ast.walk(tree):
|
||||
if isinstance(node, ast.FunctionDef):
|
||||
parser.parse_functions(node)
|
||||
|
|
@ -177,7 +179,7 @@ def test_code_parser_parse_classes():
|
|||
Test the parse_classes method of the CodeParser class.
|
||||
"""
|
||||
parser = CodeParser("class Test: pass")
|
||||
tree = parser._CodeParser__get_tree()
|
||||
tree = parser.get_tree()
|
||||
for node in ast.walk(tree):
|
||||
if isinstance(node, ast.ClassDef):
|
||||
parser.parse_classes(node)
|
||||
|
|
@ -190,7 +192,7 @@ def test_code_parser_parse_global_vars():
|
|||
Test the parse_global_vars method of the CodeParser class.
|
||||
"""
|
||||
parser = CodeParser("x = 1")
|
||||
tree = parser._CodeParser__get_tree()
|
||||
tree = parser.get_tree()
|
||||
for node in ast.walk(tree):
|
||||
if isinstance(node, ast.Assign):
|
||||
parser.parse_global_vars(node)
|
||||
|
|
@ -311,7 +313,7 @@ def test_code_parser_parse_ann_assign():
|
|||
stmt = ast.AnnAssign(
|
||||
target=ast.Name(id="x", ctx=ast.Store()),
|
||||
annotation=ast.Name(id="int", ctx=ast.Load()),
|
||||
value=ast.Num(n=1),
|
||||
value=ast.Constant(n=1),
|
||||
simple=1,
|
||||
)
|
||||
result = parser.parse_ann_assign(stmt)
|
||||
|
|
|
|||
Loading…
Reference in a new issue