Migration to New Custom Components in Langflow

YamonBot commented 3 days ago

We have previously written code that parses the function signatures from source code entered as a Code type in real-time and adds the relevant details as fields in the build configuration.

Now, it seems that Langflow has introduced a new concept of custom components (from langflow.custom import Component).

How can we use the existing design to add the signatures of specific methods as new fields in the UI?

from langflow import CustomComponent
from typing import Any, Dict, List, Union
import ast
from langchain_core.tools import Tool
from langchain.tools import StructuredTool
from langflow.schema.dotdict import dotdict
from langchain.pydantic_v1 import BaseModel, Field, create_model
from loguru import logger
from langflow.custom.code_parser import CodeParser  # Already defined methods included

class PythonMethodTool(CustomComponent):
    display_name = "Python Method Tool"
    description = "Creates a StructuredTool from a given Python function code."
    documentation = "https://python.langchain.com/docs/modules/tools/custom_tools/#structuredtool-dataclass"
    icon = "Python"
    field_order = ["name", "description", "tool_code", "return_direct", "tool_function"]

    def build_config(self) -> Dict[str, Any]:
        return {
            "tool_code": {
                "display_name": "Tool Code",
                "info": "Enter the dataclass code.",
                "placeholder": "def my_function(args):\n    pass",
            },
            "name": {
                "display_name": "Tool Name",
                "info": "Enter the name of the tool.",
            },
            "description": {
                "display_name": "Description",
                "info": "Provide a brief description of what the tool does.",
            },
            "return_direct": {
                "display_name": "Return Directly",
                "info": "Should the tool return the function output directly?",
            },
            "tool_function": {
                "display_name": "Tool Function",
                "info": "Select the function for additional expressions.",
                "options": [],
                "real_time_refresh": True,
                "refresh_button": True,
            },
        }

    def parse_source_name(self, code: str) -> List[str]:
        try:
            parser = CodeParser(code)
            return [node.name for node in parser.get_tree().body if isinstance(node, ast.FunctionDef)]
        except SyntaxError as e:
            logger.error(f"Syntax error in code: {e}")
            return []

    def get_function_args(self, code: str, func_name: str) -> Dict[str, str]:
        try:
            parser = CodeParser(code)
            func_def = next(node for node in parser.get_tree().body if isinstance(node, ast.FunctionDef) and node.name == func_name)
            args_info = {}
            for arg in func_def.args.args:
                arg_name = arg.arg
                param_type_str = 'Any'
                if arg.annotation:
                    if isinstance(arg.annotation, ast.Subscript) and isinstance(arg.annotation.value, ast.Name) and arg.annotation.value.id == 'Optional':
                        param_type_str = f"Optional[{arg.annotation.slice.id}]"
                    else:
                        param_type_str = arg.annotation.id if isinstance(arg.annotation, ast.Name) else 'Any'
                args_info[arg_name] = param_type_str
            return args_info
        except Exception as e:
            logger.error(f"Error getting function args: {e}")
            return {}

    def update_build_config(self, build_config: dotdict, field_value: Any, field_name: Union[str, None] = None) -> dotdict:
        try:
            if field_name in ["tool_code", "tool_function"]:
                build_config["tool_function"]["options"] = self.parse_source_name(build_config["tool_code"]["value"])
                if field_name == "tool_function" and build_config["tool_function"]["value"]:
                    inputs = self.get_function_args(build_config["tool_code"]["value"], build_config["tool_function"]["value"])
                    build_config.update({name: {"display_name": name, "name": name, "field_type": "str", "required": True, "info": f"Write a description of the parameter {name}"} for name in inputs})
        except Exception as e:
            logger.error(f"Error updating build config: {e}")
            self.status = f"Failed to update build config: {str(e)}"
        return build_config

    def create_args_schema(self, tool_code: str, function_name: str) -> Dict[str, BaseModel]:
        inputs = self.get_function_args(tool_code, function_name)
        fields = {name: (eval(_type), Field(description=name)) for name, _type in inputs.items()}
        model = create_model('ToolInput', **fields, __base__=BaseModel)
        return {'ToolInput': model}

    async def build(self, tool_code: Code, name: str, description: str, tool_function: str, return_direct: bool) -> Tool:
        parser = CodeParser(tool_code)
        func_node = next(node for node in parser.get_tree().body if isinstance(node, ast.FunctionDef) and node.name == tool_function)
        local_namespace = {}
        exec(compile(ast.Module(body=[func_node]), filename="<ast>", mode="exec"), globals(), local_namespace)
        func = local_namespace[tool_function]

        schema_dict = self.create_args_schema(tool_code, tool_function)
        tool = StructuredTool.from_function(
            func=func,
            args_schema=schema_dict['ToolInput'],
            name=name,
            description=description,
            return_direct=return_direct
        )
        return tool

dosubot[bot] commented 3 days ago

To integrate the function signatures as new fields in the UI using the new CustomComponent class in Langflow, you can follow these steps:

Define the Component: Use the CustomComponent class to define your component, specifying the necessary inputs and outputs.
Build Configuration: Implement the build_config method to define the initial configuration of the component, including the fields for the tool code, name, description, return directly, and tool function.
Update Configuration: Implement the update_build_config method to dynamically update the configuration based on the provided tool code and selected function. This method will parse the function signatures and add them as new fields.
Parse Function Signatures: Use the CodeParser class to parse the function signatures from the provided code and extract the arguments.
Create Argument Schema: Implement a method to create a Pydantic model for the function arguments, which will be used to validate the inputs.

Here is the updated code with these steps:

from langflow import CustomComponent
from typing import Any, Dict, List, Union
import ast
from langchain_core.tools import Tool
from langchain.tools import StructuredTool
from langflow.schema.dotdict import dotdict
from langchain.pydantic_v1 import BaseModel, Field, create_model
from loguru import logger
from langflow.custom.code_parser import CodeParser  # Already defined methods included

class PythonMethodTool(CustomComponent):
    display_name = "Python Method Tool"
    description = "Creates a StructuredTool from a given Python function code."
    documentation = "https://python.langchain.com/docs/modules/tools/custom_tools/#structuredtool-dataclass"
    icon = "Python"
    field_order = ["name", "description", "tool_code", "return_direct", "tool_function"]

    def build_config(self) -> Dict[str, Any]:
        return {
            "tool_code": {
                "display_name": "Tool Code",
                "info": "Enter the dataclass code.",
                "placeholder": "def my_function(args):\n    pass",
            },
            "name": {
                "display_name": "Tool Name",
                "info": "Enter the name of the tool.",
            },
            "description": {
                "display_name": "Description",
                "info": "Provide a brief description of what the tool does.",
            },
            "return_direct": {
                "display_name": "Return Directly",
                "info": "Should the tool return the function output directly?",
            },
            "tool_function": {
                "display_name": "Tool Function",
                "info": "Select the function for additional expressions.",
                "options": [],
                "real_time_refresh": True,
                "refresh_button": True,
            },
        }

    def parse_source_name(self, code: str) -> List[str]:
        try:
            parser = CodeParser(code)
            return [node.name for node in parser.get_tree().body if isinstance(node, ast.FunctionDef)]
        except SyntaxError as e:
            logger.error(f"Syntax error in code: {e}")
            return []

    def get_function_args(self, code: str, func_name: str) -> Dict[str, str]:
        try:
            parser = CodeParser(code)
            func_def = next(node for node in parser.get_tree().body if isinstance(node, ast.FunctionDef) and node.name == func_name)
            args_info = {}
            for arg in func_def.args.args:
                arg_name = arg.arg
                param_type_str = 'Any'
                if arg.annotation:
                    if isinstance(arg.annotation, ast.Subscript) and isinstance(arg.annotation.value, ast.Name) and arg.annotation.value.id == 'Optional':
                        param_type_str = f"Optional[{arg.annotation.slice.id}]"
                    else:
                        param_type_str = arg.annotation.id if isinstance(arg.annotation, ast.Name) else 'Any'
                args_info[arg_name] = param_type_str
            return args_info
        except Exception as e:
            logger.error(f"Error getting function args: {e}")
            return {}

    def update_build_config(self, build_config: dotdict, field_value: Any, field_name: Union[str, None] = None) -> dotdict:
        try:
            if field_name in ["tool_code", "tool_function"]:
                build_config["tool_function"]["options"] = self.parse_source_name(build_config["tool_code"]["value"])
                if field_name == "tool_function" and build_config["tool_function"]["value"]:
                    inputs = self.get_function_args(build_config["tool_code"]["value"], build_config["tool_function"]["value"])
                    build_config.update({name: {"display_name": name, "name": name, "field_type": "str", "required": True, "info": f"Write a description of the parameter {name}"} for name in inputs})
        except Exception as e:
            logger.error(f"Error updating build config: {e}")
            self.status = f"Failed to update build config: {str(e)}"
        return build_config

    def create_args_schema(self, tool_code: str, function_name: str) -> Dict[str, BaseModel]:
        inputs = self.get_function_args(tool_code, function_name)
        fields = {name: (eval(_type), Field(description=name)) for name, _type in inputs.items()}
        model = create_model('ToolInput', **fields, __base__=BaseModel)
        return {'ToolInput': model}

    async def build(self, tool_code: str, name: str, description: str, tool_function: str, return_direct: bool) -> Tool:
        parser = CodeParser(tool_code)
        func_node = next(node for node in parser.get_tree().body if isinstance(node, ast.FunctionDef) and node.name == tool_function)
        local_namespace = {}
        exec(compile(ast.Module(body=[func_node]), filename="<ast>", mode="exec"), globals(), local_namespace)
        func = local_namespace[tool_function]

        schema_dict = self.create_args_schema(tool_code, tool_function)
        tool = StructuredTool.from_function(
            func=func,
            args_schema=schema_dict['ToolInput'],
            name=name,
            description=description,
            return_direct=return_direct
        )
        return tool

This code leverages the CustomComponent class to dynamically update the UI with function signatures as new fields, ensuring real-time updates and dynamic field generation ^[1].

To continue talking to Dosu, mention @dosu.

YamonBot commented 3 days ago

@dosu

I would like an explanation of the structure that inherits from the from langflow.custom import Component code.

dosubot[bot] commented 3 days ago

To create custom components in Langflow, you need to define a class that inherits from langflow.custom.Component and implement specific methods and attributes. Here is a step-by-step guide:

Inherit from Component: Create a class that inherits from langflow.custom.Component.
Define Metadata: Set attributes like display_name, description, icon, and documentation to provide metadata for your component.
Define Inputs and Outputs: Use the inputs and outputs lists to specify the inputs and outputs for the component. Inputs can be of various types such as MessageTextInput, IntInput, BoolInput, DropdownInput, etc.
Implement Processing Logic: Define methods for processing data within the component. The main method should process the inputs and return the results.

Here is an example:

from langflow.custom import Component
from langflow.io import MessageTextInput, Output
from langflow.schema import Data

class MyCustomComponent(Component):
    display_name = "My Custom Component"
    description = "A template for creating custom components."
    documentation = "http://docs.langflow.org/components/custom"
    icon = "custom_components"

    inputs = [
        MessageTextInput(
            name="input_value",
            display_name="Input Value",
            value="Hello, World!",
        ),
    ]

    outputs = [
        Output(display_name="Output", name="output", method="build_output"),
    ]

    def build_output(self) -> Data:
        data = Data(value=self.input_value)
        self.status = data
        return data

Key Points:

Class Definition: Inherits from the Component class.
Component Metadata: Defines display_name, description, icon, and documentation.
Inputs and Outputs: Specifies the inputs and outputs for the component.
Processing Logic: Implements the logic for processing data within the component.

By following these steps and examples, you can create custom components in Langflow tailored to your specific needs. The modular structure of Custom Components allows for flexible and reusable components that can be easily integrated into your workflows ^[1]^[2]^[3].

To continue talking to Dosu, mention @dosu.

nicoloboschi commented 22 hours ago

Hey @YamonBot

The new component system is much more usable and easy.

My suggestion would be to look at an existing component: https://github.com/langflow-ai/langflow/blob/main/src/backend/base/langflow/components/vectorstores/AstraDB.py

All the paramters now must be in the inputs. There is a class for each input type.

The build(..) method must be renamed to something else, like build_component(self,...) and referenced in the outputs section.

langflow-ai / langflow

Migration to New Custom Components in Langflow #2335

Key Points: