refactor: Update langflow components and constants

This commit is contained in:
Rodrigo 2024-06-13 16:59:37 -03:00
commit c89f87baea
10 changed files with 158 additions and 129 deletions

View file

@ -5,6 +5,8 @@ from langflow.inputs import StrInput
from langflow.schema import Data from langflow.schema import Data
from langflow.template import Output from langflow.template import Output
import re
class URLComponent(Component): class URLComponent(Component):
display_name = "URL" display_name = "URL"
@ -25,8 +27,39 @@ class URLComponent(Component):
Output(display_name="Data", name="data", method="fetch_content"), Output(display_name="Data", name="data", method="fetch_content"),
] ]
def ensure_url(self, string: str) -> str:
"""
Ensures the given string is a URL by adding 'http://' if it doesn't start with 'http://' or 'https://'.
Raises an error if the string is not a valid URL.
Parameters:
string (str): The string to be checked and possibly modified.
Returns:
str: The modified string that is ensured to be a URL.
Raises:
ValueError: If the string is not a valid URL.
"""
if not string.startswith(("http://", "https://")):
string = "http://" + string
# Basic URL validation regex
url_regex = re.compile(
r"^(http://|https://)?" # http:// or https://
r"(([a-zA-Z0-9\.-]+)" # domain
r"(\.[a-zA-Z]{2,}))" # top-level domain
r"(:[0-9]{1,5})?" # optional port
r"(\/.*)?$" # optional path
)
if not re.match(url_regex, string):
raise ValueError(f"Invalid URL: {string}")
return string
def fetch_content(self) -> Data: def fetch_content(self) -> Data:
urls = [url.strip() for url in self.urls if url.strip()] urls = [self.ensure_url(url.strip()) for url in self.urls if url.strip()]
loader = WebBaseLoader(web_paths=urls) loader = WebBaseLoader(web_paths=urls)
docs = loader.load() docs = loader.load()
data = [Data(content=doc.page_content, **doc.metadata) for doc in docs] data = [Data(content=doc.page_content, **doc.metadata) for doc in docs]

View file

@ -1,69 +1,71 @@
from typing import Union
from langflow.custom import Component from langflow.custom import Component
from langflow.field_typing import Text from langflow.field_typing import Text
from langflow.schema import Data from langflow.inputs import BoolInput, DropdownInput, StrInput
from langflow.template import Input, Output from langflow.template import Output
class TextOperatorComponent(Component): class TextOperatorComponent(Component):
display_name = "Text Operator" display_name = "Text Operator"
description = "Compares two text inputs based on a specified condition such as equality or inequality, with optional case sensitivity." description = "Compares two text inputs based on a specified condition such as equality or inequality, with optional case sensitivity."
icon = "equal"
inputs = [ inputs = [
Input(name="input_text", type=str, display_name="Input Text", info="The primary text input for the operation."), StrInput(
Input(name="match_text", type=str, display_name="Match Text", info="The text input to compare against."), name="input_text",
Input( display_name="Input Text",
name="operator", info="The primary text input for the operation.",
type=str,
display_name="Operator",
info="The operator to apply for comparing the texts.",
options=["equals", "not equals", "contains", "starts with", "ends with", "exists"],
), ),
Input( StrInput(
name="match_text",
display_name="Match Text",
info="The text input to compare against.",
),
DropdownInput(
name="operator",
display_name="Operator",
options=["equals", "not equals", "contains", "starts with", "ends with"],
info="The operator to apply for comparing the texts.",
),
BoolInput(
name="case_sensitive", name="case_sensitive",
type=bool,
display_name="Case Sensitive", display_name="Case Sensitive",
info="If true, the comparison will be case sensitive.", info="If true, the comparison will be case sensitive.",
default=False, value=False,
advanced=True,
), ),
Input( StrInput(
name="true_output", name="true_output",
type=Union[str, Data],
display_name="True Output", display_name="True Output",
info="The output to return or display when the comparison is true.", info="The output to return or display when the comparison is true.",
input_types=["Text", "Data"], advanced=True,
), ),
Input( StrInput(
name="false_output", name="false_output",
type=Union[str, Data],
display_name="False Output", display_name="False Output",
info="The output to return or display when the comparison is false.", info="The output to return or display when the comparison is false.",
input_types=["Text", "Data"], advanced=True,
), ),
] ]
outputs = [ outputs = [
Output(display_name="True Result", name="true_result", method="result_response"), Output(display_name="True Result", name="true_result", method="true_response"),
Output(display_name="False Result", name="false_result", method="result_response"), Output(display_name="False Result", name="false_result", method="false_response"),
] ]
def true_response(self) -> Union[Text, Data]: def true_response(self) -> Text:
self.stop("False Result") self.stop("false_result")
return self.true_output if self.true_output else self.input_text return self.true_output or self.input_text
def false_response(self) -> Union[Text, Data]: def false_response(self) -> Text:
self.stop("True Result") self.stop("true_result")
return self.false_output if self.false_output else self.input_text return self.false_output or self.input_text
def result_response(self) -> Union[Text, Data]: def run(self) -> Text:
input_text = self.input_text input_text = self.input_text
match_text = self.match_text match_text = self.match_text
operator = self.operator operator = self.operator
case_sensitive = self.case_sensitive case_sensitive = self.case_sensitive
if not input_text or not match_text:
raise ValueError("Both 'input_text' and 'match_text' must be provided and non-empty.")
if not case_sensitive: if not case_sensitive:
input_text = input_text.lower() input_text = input_text.lower()
match_text = match_text.lower() match_text = match_text.lower()

View file

@ -1,29 +1,38 @@
from langflow.custom import CustomComponent from langflow.custom import Component
from langflow.field_typing import Text from langflow.field_typing import Text
from langflow.inputs import StrInput
from langflow.template import Output
class CombineTextComponent(CustomComponent): class CombineTextComponent(Component):
display_name = "Combine Text" display_name = "Combine Text"
description = "Concatenate two text sources into a single text chunk using a specified delimiter." description = "Concatenate two text sources into a single text chunk using a specified delimiter."
icon = "merge" icon = "merge"
def build_config(self): inputs = [
return { StrInput(
"text1": { name="text1",
"display_name": "First Text", display_name="First Text",
"info": "The first text input to concatenate.", info="The first text input to concatenate.",
}, ),
"text2": { StrInput(
"display_name": "Second Text", name="text2",
"info": "The second text input to concatenate.", display_name="Second Text",
}, info="The second text input to concatenate.",
"delimiter": { ),
"display_name": "Delimiter", StrInput(
"info": "A string used to separate the two text inputs. Defaults to a whitespace.", name="delimiter",
}, display_name="Delimiter",
} info="A string used to separate the two text inputs. Defaults to a whitespace.",
default=" ",
),
]
def build(self, text1: str, text2: str, delimiter: str = " ") -> Text: outputs = [
combined = delimiter.join([text1, text2]) Output(display_name="Combined Text", name="combined_text", method="combine_texts"),
]
def combine_texts(self) -> Text:
combined = self.delimiter.join([self.text1, self.text2])
self.status = combined self.status = combined
return combined return combined

View file

@ -1,25 +0,0 @@
from langflow.custom import CustomComponent
from langflow.field_typing import Text
class CombineTextsUnsortedComponent(CustomComponent):
display_name = "Combine Texts (Unsorted)"
description = "Concatenate text sources into a single text chunk using a specified delimiter."
icon = "merge"
def build_config(self):
return {
"texts": {
"display_name": "Texts",
"info": "The first text input to concatenate.",
},
"delimiter": {
"display_name": "Delimiter",
"info": "A string used to separate the two text inputs. Defaults to a whitespace.",
},
}
def build(self, texts: list[str], delimiter: str = " ") -> Text:
combined = delimiter.join(texts)
self.status = combined
return combined

View file

@ -1,36 +0,0 @@
from langflow.custom import CustomComponent
from langflow.field_typing import Text
from langflow.helpers.data import data_to_text
from langflow.schema import Data
class DataToTextComponent(CustomComponent):
display_name = "Data To Text"
description = "Convert Data into plain text following a specified template."
def build_config(self):
return {
"data": {
"display_name": "Data",
"info": "The data to convert to text.",
},
"template": {
"display_name": "Template",
"info": "The template to use for formatting the data. It can contain the keys {text}, {data} or any other key in the Data.",
"multiline": True,
},
}
def build(
self,
data: list[Data],
template: str = "Text: {text}\nData: {data}",
) -> Text:
if not data:
return ""
if isinstance(data, Data):
data = [data]
result_string = data_to_text(template, data)
self.status = result_string
return result_string

View file

@ -6,9 +6,10 @@ from langflow.custom import CustomComponent
from langflow.schema import Data from langflow.schema import Data
class DocumentToDataComponent(CustomComponent): class DocumentsToDataComponent(CustomComponent):
display_name = "Documents To Data" display_name = "Documents ⇢ Data"
description = "Convert LangChain Documents into Data." description = "Convert LangChain Documents into Data."
icon = "LangChain"
field_config = { field_config = {
"documents": {"display_name": "Documents"}, "documents": {"display_name": "Documents"},

View file

@ -1,20 +1,28 @@
from typing import List from typing import List
from langflow.custom import Component from langflow.custom import Component
from langflow.inputs import StrInput from langflow.inputs import StrInput, HandleInput
from langflow.schema import Data from langflow.schema import Data
from langflow.template import Input, Output from langflow.template import Output
class FilterDataComponent(Component): class FilterDataComponent(Component):
display_name = "Filter Message" display_name = "Filter Data"
description = "Filters a Message object based on a list of strings." description = "Filters a Data object based on a list of keys."
icon = "filter" icon = "filter"
inputs = [ inputs = [
Input(name="message", display_name="Message", info="Message object to filter.", input_types=["Message"]), HandleInput(
name="data",
display_name="Data",
info="Data object to filter.",
input_types=["Message", "Data"],
),
StrInput( StrInput(
name="filter_criteria", display_name="Filter Criteria", info="List of strings to filter by.", is_list=True name="filter_criteria",
display_name="Filter Criteria",
info="List of keys to filter by.",
is_list=True,
), ),
] ]
@ -24,10 +32,12 @@ class FilterDataComponent(Component):
def filter_data(self) -> Data: def filter_data(self) -> Data:
filter_criteria: List[str] = self.filter_criteria filter_criteria: List[str] = self.filter_criteria
data = self.data.data if isinstance(self.data, Data) else {}
# Filter the data # Filter the data
filtered = {key: value for key, value in self.message.data.items() if key == filter_criteria} filtered = {key: value for key, value in data.items() if key in filter_criteria}
# Create a new Data object with the filtered data # Create a new Data object with the filtered data
self.status = filtered filtered_data = Data(data=filtered)
return filtered self.status = filtered_data
return filtered_data

View file

@ -0,0 +1,34 @@
from langflow.custom import Component
from langflow.helpers.data import data_to_text
from langflow.field_typing import Text
from langflow.inputs import MultilineInput, HandleInput
from langflow.template import Output
class ParseDataComponent(Component):
display_name = "Parse Data"
description = "Convert Data into plain text following a specified template."
icon = "braces"
inputs = [
HandleInput(
name="data", display_name="Data", info="The data to convert to text.", input_types=["Message", "Data"]
),
MultilineInput(
name="template",
display_name="Template",
info="The template to use for formatting the data. It can contain the keys {text}, {data} or any other key in the Data.",
),
]
outputs = [
Output(display_name="Text", name="text", method="parse_data_to_text"),
]
def parse_data_to_text(self) -> Text:
data = self.data if isinstance(self.data, list) else [self.data]
template = self.template or "Text: {text}"
result_string = data_to_text(template, data)
self.status = result_string
return result_string

View file

@ -1,15 +1,15 @@
from .CreateData import CreateDataComponent from .CreateData import CreateDataComponent
from .CustomComponent import Component from .CustomComponent import Component
from .DataToText import DataToTextComponent from .ParseData import ParseDataComponent
from .DocumentToData import DocumentToDataComponent from .DocumentToData import DocumentsToDataComponent
from .IDGenerator import UUIDGeneratorComponent from .IDGenerator import UUIDGeneratorComponent
from .UpdateData import UpdateDataComponent from .UpdateData import UpdateDataComponent
__all__ = [ __all__ = [
"Component", "Component",
"UpdateDataComponent", "UpdateDataComponent",
"DocumentToDataComponent", "DocumentsToDataComponent",
"UUIDGeneratorComponent", "UUIDGeneratorComponent",
"DataToTextComponent", "ParseDataComponent",
"CreateDataComponent", "CreateDataComponent",
] ]

View file

@ -35,6 +35,7 @@ class PromptInput(BaseInputMixin, ListableInputMixin):
class StrInput(BaseInputMixin, ListableInputMixin, DatabaseLoadMixin): # noqa: F821 class StrInput(BaseInputMixin, ListableInputMixin, DatabaseLoadMixin): # noqa: F821
field_type: Optional[SerializableFieldTypes] = FieldTypes.TEXT field_type: Optional[SerializableFieldTypes] = FieldTypes.TEXT
load_from_db: StrictBoolean = False load_from_db: StrictBoolean = False
input_types: list[str] = ["Text"]
"""Defines if the field will allow the user to open a text editor. Default is False.""" """Defines if the field will allow the user to open a text editor. Default is False."""