diff --git a/src/backend/base/langflow/components/processing/__init__.py b/src/backend/base/langflow/components/processing/__init__.py index d9e7faf1d..de249865e 100644 --- a/src/backend/base/langflow/components/processing/__init__.py +++ b/src/backend/base/langflow/components/processing/__init__.py @@ -4,6 +4,7 @@ from .create_data import CreateDataComponent from .extract_key import ExtractDataKeyComponent from .filter_data_values import DataFilterComponent from .json_cleaner import JSONCleaner +from .lambda_filter import LambdaFilterComponent from .llm_router import LLMRouterComponent from .merge_data import MergeDataComponent from .message_to_data import MessageToDataComponent @@ -23,6 +24,7 @@ __all__ = [ "ExtractDataKeyComponent", "JSONCleaner", "LLMRouterComponent", + "LambdaFilterComponent", "MergeDataComponent", "MessageToDataComponent", "ParseDataComponent", diff --git a/src/backend/base/langflow/components/processing/lambda_filter.py b/src/backend/base/langflow/components/processing/lambda_filter.py new file mode 100644 index 000000000..cf3a6e848 --- /dev/null +++ b/src/backend/base/langflow/components/processing/lambda_filter.py @@ -0,0 +1,165 @@ +from __future__ import annotations + +import json +import re +from typing import TYPE_CHECKING, Any + +from langflow.custom import Component +from langflow.io import DataInput, HandleInput, IntInput, MultilineInput, Output +from langflow.schema import Data +from langflow.schema.dataframe import DataFrame +from langflow.utils.data_structure import get_data_structure + +if TYPE_CHECKING: + from collections.abc import Callable + + +class LambdaFilterComponent(Component): + display_name = "Lambda Filter" + description = "Uses an LLM to generate a lambda function for filtering or transforming structured data." + icon = "filter" + name = "LambdaFilter" + beta = True + + inputs = [ + DataInput( + name="data", + display_name="Data", + info="The structured data to filter or transform using a lambda function.", + is_list=True, + required=True, + ), + HandleInput( + name="llm", + display_name="Language Model", + info="Connect the 'Language Model' output from your LLM component here.", + input_types=["LanguageModel"], + required=True, + ), + MultilineInput( + name="filter_instruction", + display_name="Instructions", + info=( + "Natural language instructions for how to filter or transform the data using a lambda function. " + "Example: Filter the data to only include items where the 'status' is 'active'." + ), + value="Filter the data to...", + required=True, + ), + IntInput( + name="sample_size", + display_name="Sample Size", + info="For large datasets, number of items to sample from head/tail.", + value=1000, + advanced=True, + ), + IntInput( + name="max_size", + display_name="Max Size", + info="Number of characters for the data to be considered large.", + value=30000, + advanced=True, + ), + ] + + outputs = [ + Output( + display_name="Filtered Data", + name="filtered_data", + method="filter_data", + ), + Output( + display_name="DataFrame", + name="dataframe", + method="as_dataframe", + ), + ] + + def get_data_structure(self, data): + """Extract the structure of a dictionary, replacing values with their types.""" + return {k: get_data_structure(v) for k, v in data.items()} + + def _validate_lambda(self, lambda_text: str) -> bool: + """Validate the provided lambda function text.""" + # Return False if the lambda function does not start with 'lambda' or does not contain a colon + return lambda_text.strip().startswith("lambda") and ":" in lambda_text + + async def filter_data(self) -> list[Data]: + self.log(str(self.data)) + data = self.data[0].data if isinstance(self.data, list) else self.data.data + + dump = json.dumps(data) + self.log(str(data)) + + llm = self.llm + instruction = self.filter_instruction + sample_size = self.sample_size + + # Get data structure and samples + data_structure = self.get_data_structure(data) + dump_structure = json.dumps(data_structure) + self.log(dump_structure) + + # For large datasets, sample from head and tail + if len(dump) > self.max_size: + data_sample = ( + f"Data is too long to display... \n\n First lines (head): {dump[:sample_size]} \n\n" + f" Last lines (tail): {dump[-sample_size:]})" + ) + else: + data_sample = dump + + self.log(data_sample) + + prompt = f"""Given this data structure and examples, create a Python lambda function that + implements the following instruction: + + Data Structure: + {dump_structure} + + Example Items: + {data_sample} + + Instruction: {instruction} + + Return ONLY the lambda function and nothing else. No need for ```python or whatever. + Just a string starting with lambda. + """ + + response = await llm.ainvoke(prompt) + response_text = response.content if hasattr(response, "content") else str(response) + self.log(response_text) + + # Extract lambda using regex + lambda_match = re.search(r"lambda\s+\w+\s*:.*?(?=\n|$)", response_text) + if not lambda_match: + msg = f"Could not find lambda in response: {response_text}" + raise ValueError(msg) + + lambda_text = lambda_match.group().strip() + self.log(lambda_text) + + # Validation is commented out as requested + if not self._validate_lambda(lambda_text): + msg = f"Invalid lambda format: {lambda_text}" + raise ValueError(msg) + + # Create and apply the function + fn: Callable[[Any], Any] = eval(lambda_text) # noqa: S307 + + # Apply the lambda function to the data + processed_data = fn(data) + + # If it's a dict, wrap it in a Data object + if isinstance(processed_data, dict): + return [Data(**processed_data)] + # If it's a list, convert each item to a Data object + if isinstance(processed_data, list): + return [Data(**item) if isinstance(item, dict) else Data(text=str(item)) for item in processed_data] + # If it's anything else, convert to string and wrap in a Data object + return [Data(text=str(processed_data))] + + async def as_dataframe(self) -> DataFrame: + """Return filtered data as a DataFrame.""" + filtered_data = await self.filter_data() + return DataFrame(filtered_data) diff --git a/src/backend/base/langflow/utils/data_structure.py b/src/backend/base/langflow/utils/data_structure.py new file mode 100644 index 000000000..5ba5023b9 --- /dev/null +++ b/src/backend/base/langflow/utils/data_structure.py @@ -0,0 +1,212 @@ +import json +from collections import Counter +from typing import Any + +from langflow.schema import Data + + +def infer_list_type(items: list, max_samples: int = 5) -> str: + """Infer the type of a list by sampling its items. + + Handles mixed types and provides more detailed type information. + """ + if not items: + return "list(unknown)" + + # Sample items (use all if less than max_samples) + samples = items[:max_samples] + types = [get_type_str(item) for item in samples] + + # Count type occurrences + type_counter = Counter(types) + + if len(type_counter) == 1: + # Single type + return f"list({types[0]})" + # Mixed types - show all found types + type_str = "|".join(sorted(type_counter.keys())) + return f"list({type_str})" + + +def get_type_str(value: Any) -> str: + """Get a detailed string representation of the type of a value. + + Handles special cases and provides more specific type information. + """ + if value is None: + return "null" + if isinstance(value, bool): + return "bool" + if isinstance(value, int): + return "int" + if isinstance(value, float): + return "float" + if isinstance(value, str): + # Check if string is actually a date/datetime + if any(date_pattern in value.lower() for date_pattern in ["date", "time", "yyyy", "mm/dd", "dd/mm", "yyyy-mm"]): + return "str(possible_date)" + # Check if it's a JSON string + try: + json.loads(value) + return "str(json)" + except (json.JSONDecodeError, TypeError): + pass + else: + return "str" + if isinstance(value, list | tuple | set): + return infer_list_type(list(value)) + if isinstance(value, dict): + return "dict" + # Handle custom objects + return type(value).__name__ + + +def analyze_value( + value: Any, + max_depth: int = 10, + current_depth: int = 0, + path: str = "", + *, + size_hints: bool = True, + include_samples: bool = True, +) -> str | dict: + """Analyze a value and return its structure with additional metadata. + + Args: + value: The value to analyze + max_depth: Maximum recursion depth + current_depth: Current recursion depth + path: Current path in the structure + size_hints: Whether to include size information for collections + include_samples: Whether to include sample structure for lists + """ + if current_depth >= max_depth: + return f"max_depth_reached(depth={max_depth})" + + try: + if isinstance(value, list | tuple | set): + length = len(value) + if length == 0: + return "list(unknown)" + + type_info = infer_list_type(list(value)) + size_info = f"[size={length}]" if size_hints else "" + + # For lists of complex objects, include a sample of the structure + if ( + include_samples + and length > 0 + and isinstance(value, list | tuple) + and isinstance(value[0], dict | list) + and current_depth < max_depth - 1 + ): + sample = analyze_value( + value[0], + max_depth, + current_depth + 1, + f"{path}[0]", + size_hints=size_hints, + include_samples=include_samples, + ) + return f"{type_info}{size_info}, sample: {json.dumps(sample)}" + + return f"{type_info}{size_info}" + + if isinstance(value, dict): + result = {} + for k, v in value.items(): + new_path = f"{path}.{k}" if path else k + try: + result[k] = analyze_value( + v, + max_depth, + current_depth + 1, + new_path, + size_hints=size_hints, + include_samples=include_samples, + ) + except Exception as e: # noqa: BLE001 + result[k] = f"error({e!s})" + return result + + return get_type_str(value) + + except Exception as e: # noqa: BLE001 + return f"error({e!s})" + + +def get_data_structure( + data_obj: Data | dict, + max_depth: int = 10, + max_sample_size: int = 3, + *, + size_hints: bool = True, + include_sample_values: bool = False, + include_sample_structure: bool = True, +) -> dict: + """Convert a Data object or dictionary into a detailed schema representation. + + Args: + data_obj: The Data object or dictionary to analyze + max_depth: Maximum depth for nested structures + size_hints: Include size information for collections + include_sample_values: Whether to include sample values in the output + include_sample_structure: Whether to include sample structure for lists + max_sample_size: Maximum number of sample values to include + + Returns: + dict: A dictionary containing: + - structure: The structure of the data + - samples: (optional) Sample values from the data + + Example: + >>> data = { + ... "name": "John", + ... "scores": [1, 2, 3, 4, 5], + ... "details": { + ... "age": 30, + ... "cities": ["NY", "LA", "SF", "CHI"], + ... "metadata": { + ... "created": "2023-01-01", + ... "tags": ["user", "admin", 123] + ... } + ... } + ... } + >>> result = get_data_structure(data) + { + "structure": { + "name": "str", + "scores": "list(int)[size=5]", + "details": { + "age": "int", + "cities": "list(str)[size=4]", + "metadata": { + "created": "str(possible_date)", + "tags": "list(str|int)[size=3]" + } + } + } + } + """ + # Handle both Data objects and dictionaries + data = data_obj.data if isinstance(data_obj, Data) else data_obj + + result = { + "structure": analyze_value( + data, max_depth=max_depth, size_hints=size_hints, include_samples=include_sample_structure + ) + } + + if include_sample_values: + result["samples"] = get_sample_values(data, max_items=max_sample_size) + + return result + + +def get_sample_values(data: Any, max_items: int = 3) -> Any: + """Get sample values from a data structure, handling nested structures.""" + if isinstance(data, list | tuple | set): + return [get_sample_values(item) for item in list(data)[:max_items]] + if isinstance(data, dict): + return {k: get_sample_values(v, max_items) for k, v in data.items()} + return data diff --git a/src/backend/tests/unit/components/processing/test_lambda_filter.py b/src/backend/tests/unit/components/processing/test_lambda_filter.py new file mode 100644 index 000000000..cfdf595f3 --- /dev/null +++ b/src/backend/tests/unit/components/processing/test_lambda_filter.py @@ -0,0 +1,122 @@ +from unittest.mock import AsyncMock + +import pytest +from langflow.components.processing.lambda_filter import LambdaFilterComponent +from langflow.schema import Data + +from tests.base import ComponentTestBaseWithoutClient + + +class TestLambdaFilterComponent(ComponentTestBaseWithoutClient): + @pytest.fixture + def component_class(self): + return LambdaFilterComponent + + @pytest.fixture + def default_kwargs(self): + return { + "data": [Data(data={"items": [{"name": "test1", "value": 10}, {"name": "test2", "value": 20}]})], + "llm": AsyncMock(), + "filter_instruction": "Filter items with value greater than 15", + "sample_size": 1000, + "max_size": 30000, + } + + @pytest.fixture + def file_names_mapping(self): + return [] + + async def test_successful_lambda_generation(self, component_class, default_kwargs): + component = await self.component_setup(component_class, default_kwargs) + component.llm.ainvoke.return_value.content = "lambda x: [item for item in x['items'] if item['value'] > 15]" + + # Execute filter + result = await component.filter_data() + + # Assertions + assert isinstance(result, list) + assert len(result) == 1 + assert result[0].name == "test2" + assert result[0].value == 20 + + async def test_invalid_lambda_response(self, component_class, default_kwargs): + component = await self.component_setup(component_class, default_kwargs) + component.llm.ainvoke.return_value.content = "invalid lambda syntax" + + # Test exception handling + with pytest.raises(ValueError, match="Could not find lambda in response"): + await component.filter_data() + + async def test_lambda_with_large_dataset(self, component_class, default_kwargs): + large_data = {"items": [{"name": f"test{i}", "value": i} for i in range(2000)]} + default_kwargs["data"] = [Data(data=large_data)] + default_kwargs["filter_instruction"] = "Filter items with value greater than 1500" + component = await self.component_setup(component_class, default_kwargs) + component.llm.ainvoke.return_value.content = "lambda x: [item for item in x['items'] if item['value'] > 1500]" + + # Execute filter + result = await component.filter_data() + + # Assertions + assert isinstance(result, list) + assert len(result) == 499 # Items with value from 1501 to 1999 + assert all(item.value > 1500 for item in result) + + async def test_lambda_with_complex_data_structure(self, component_class, default_kwargs): + complex_data = { + "categories": { + "A": [{"id": 1, "score": 90}, {"id": 2, "score": 85}], + "B": [{"id": 3, "score": 95}, {"id": 4, "score": 88}], + } + } + default_kwargs["data"] = [Data(data=complex_data)] + default_kwargs["filter_instruction"] = "Filter items with score greater than 90" + component = await self.component_setup(component_class, default_kwargs) + component.llm.ainvoke.return_value.content = ( + "lambda x: [item for cat in x['categories'].values() for item in cat if item['score'] > 90]" + ) + + # Execute filter + result = await component.filter_data() + + # Assertions + assert isinstance(result, list) + assert len(result) == 1 + assert result[0].id == 3 + assert result[0].score == 95 + + def test_validate_lambda(self, component_class): + component = component_class() + + # Valid lambda + valid_lambda = "lambda x: x + 1" + assert component._validate_lambda(valid_lambda) is True + + # Invalid lambda: missing 'lambda' + invalid_lambda_1 = "x: x + 1" + assert component._validate_lambda(invalid_lambda_1) is False + + # Invalid lambda: missing ':' + invalid_lambda_2 = "lambda x x + 1" + assert component._validate_lambda(invalid_lambda_2) is False + + def test_get_data_structure(self, component_class): + component = component_class() + test_data = { + "string": "test", + "number": 42, + "list": [1, 2, 3], + "dict": {"key": "value"}, + "nested": {"a": [{"b": 1}]}, + } + + structure = component.get_data_structure(test_data) + + # Assertions + assert structure["string"]["structure"] == "str" + assert structure["number"]["structure"] == "int" + assert structure["list"]["structure"] == "list(int)[size=3]" + assert structure["dict"]["structure"]["key"] == "str" + assert "structure" in structure["nested"] + assert "a" in structure["nested"]["structure"] + assert "list" in structure["nested"]["structure"]["a"]