feat: Add LambdaFilter for enhanced data filtering (#7095)
* feat: add LambdaFilterComponent for dynamic data filtering - Introduced LambdaFilterComponent to generate lambda functions for filtering or transforming structured data using LLMs. - Updated __init__.py to include the new component. - Added utility functions in data_structure.py for analyzing and inferring data types. - Implemented unit tests for LambdaFilterComponent to ensure functionality and error handling. * feat: enhance LambdaFilterComponent with new features and improvements - Updated filter_instruction input to provide clearer guidance and examples. - Added max_size input to specify character limits for large datasets. - Renamed output from "Processed Data" to "Filtered Data" for clarity. - Introduced new output "DataFrame" to return filtered data in DataFrame format. - Improved data handling in filter_data method to ensure proper conversion of processed data to Data objects. - Added as_dataframe method to return filtered data as a DataFrame. This update enhances usability and functionality of the LambdaFilterComponent. * [autofix.ci] apply automated fixes * fix: ruff errors * [autofix.ci] apply automated fixes * [autofix.ci] apply automated fixes (attempt 2/3) * [autofix.ci] apply automated fixes (attempt 3/3) * [autofix.ci] apply automated fixes * updated the test file and fixed formatting issues * [autofix.ci] apply automated fixes --------- Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com> Co-authored-by: italojohnny <italojohnnydosanjos@gmail.com> Co-authored-by: Edwin Jose <edwin.jose@datastax.com>
This commit is contained in:
parent
b5a93b4c55
commit
5f227a7adb
4 changed files with 501 additions and 0 deletions
|
|
@ -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",
|
||||
|
|
|
|||
165
src/backend/base/langflow/components/processing/lambda_filter.py
Normal file
165
src/backend/base/langflow/components/processing/lambda_filter.py
Normal file
|
|
@ -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)
|
||||
212
src/backend/base/langflow/utils/data_structure.py
Normal file
212
src/backend/base/langflow/utils/data_structure.py
Normal file
|
|
@ -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
|
||||
|
|
@ -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"]
|
||||
Loading…
Add table
Add a link
Reference in a new issue