Skip to content
Merged
Show file tree
Hide file tree
Changes from 1 commit
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
26 changes: 26 additions & 0 deletions haystack/components/preprocessors/document_cleaner.py
Original file line number Diff line number Diff line change
Expand Up @@ -48,6 +48,8 @@ def __init__( # pylint: disable=too-many-positional-arguments
remove_regex: str | None = None,
unicode_normalization: Literal["NFC", "NFKC", "NFD", "NFKD"] | None = None,
ascii_only: bool = False,
strip_whitespace: bool = False,
regex_replace: dict[str, str] | None = None,
):
"""
Initialize DocumentCleaner.
Expand All @@ -66,6 +68,12 @@ def __init__( # pylint: disable=too-many-positional-arguments
Will remove accents from characters and replace them with ASCII characters.
Other non-ASCII characters will be removed.
Note: This will run before any pattern matching or removal.
:param strip_whitespace: If `True`, removes leading and trailing whitespace from the document content
using Python's `str.strip()`. Unlike `remove_extra_whitespaces`, this only affects the beginning
and end of the text, preserving internal whitespace (useful for markdown formatting).
:param regex_replace: A dictionary mapping regex patterns to their replacement strings.
For example, `{r'\\n\\n+': '\\n'}` replaces multiple consecutive newlines with a single newline.
This is applied after `remove_regex` and allows custom replacements instead of just removal.
"""

self._validate_params(unicode_normalization=unicode_normalization)
Expand All @@ -78,6 +86,8 @@ def __init__( # pylint: disable=too-many-positional-arguments
self.keep_id = keep_id
self.unicode_normalization = unicode_normalization
self.ascii_only = ascii_only
self.strip_whitespace = strip_whitespace
self.regex_replace = regex_replace

def _validate_params(self, unicode_normalization: str | None):
"""
Expand Down Expand Up @@ -128,8 +138,12 @@ def run(self, documents: list[Document]):
text = self._remove_substrings(text, self.remove_substrings)
if self.remove_regex:
text = self._remove_regex(text, self.remove_regex)
if self.regex_replace:
text = self._replace_regex(text, self.regex_replace)
if self.remove_repeated_substrings:
text = self._remove_repeated_substrings(text)
if self.strip_whitespace:
text = text.strip()

clean_doc = Document(
id=doc.id if self.keep_id else "",
Expand Down Expand Up @@ -204,6 +218,18 @@ def _remove_regex(self, text: str, regex: str) -> str:
cleaned_text = [re.sub(regex, "", text).strip() for text in texts]
return "\f".join(cleaned_text)

def _replace_regex(self, text: str, regex_replace: dict[str, str]) -> str:
"""
Replace substrings that match the specified regex patterns with custom replacement strings.

:param text: Text to clean.
:param regex_replace: A dictionary mapping regex patterns to their replacement strings.
:returns: The text with the regex matches replaced by the specified strings.
"""
Comment thread
julian-risch marked this conversation as resolved.
for pattern, replacement in regex_replace.items():
text = re.sub(pattern, replacement, text)
return text

def _remove_substrings(self, text: str, substrings: list[str]) -> str:
"""
Remove all specified substrings from the text.
Expand Down
92 changes: 92 additions & 0 deletions test/components/preprocessors/test_document_cleaner.py
Original file line number Diff line number Diff line change
Expand Up @@ -230,3 +230,95 @@ def test_other_document_fields_are_not_lost(self):
assert res["documents"][0].score == document.score
assert res["documents"][0].embedding == document.embedding
assert res["documents"][0].sparse_embedding == document.sparse_embedding

def test_strip_whitespace(self):
"""Test that strip_whitespace removes only leading and trailing whitespace."""
cleaner = DocumentCleaner(
remove_empty_lines=False, remove_extra_whitespaces=False, strip_whitespace=True
)
result = cleaner.run(
documents=[Document(content=" \n\nHello World\n\n Some text here \n\n ")]
)
assert len(result["documents"]) == 1
# strip_whitespace should only remove leading/trailing whitespace, preserving internal whitespace
assert result["documents"][0].content == "Hello World\n\n Some text here"

def test_strip_whitespace_preserves_internal_formatting(self):
"""Test that strip_whitespace preserves internal whitespace like markdown formatting."""
cleaner = DocumentCleaner(
remove_empty_lines=False, remove_extra_whitespaces=False, strip_whitespace=True
)
markdown_content = """

# Header

This is a paragraph.

- Item 1
- Item 2

"""
result = cleaner.run(documents=[Document(content=markdown_content)])
assert len(result["documents"]) == 1
expected = """# Header

This is a paragraph.

- Item 1
- Item 2"""
assert result["documents"][0].content == expected

def test_regex_replace_single_pattern(self):
"""Test regex_replace with a single pattern."""
cleaner = DocumentCleaner(
remove_empty_lines=False,
remove_extra_whitespaces=False,
regex_replace={r"\n\n+": "\n"},
)
result = cleaner.run(documents=[Document(content="Line 1\n\n\n\nLine 2\n\nLine 3")])
assert len(result["documents"]) == 1
assert result["documents"][0].content == "Line 1\nLine 2\nLine 3"

def test_regex_replace_multiple_patterns(self):
"""Test regex_replace with multiple patterns."""
cleaner = DocumentCleaner(
remove_empty_lines=False,
remove_extra_whitespaces=False,
regex_replace={r"\n\n+": "\n", r"\s{2,}": " "},
)
result = cleaner.run(documents=[Document(content="Hello World\n\n\nGoodbye")])
assert len(result["documents"]) == 1
assert result["documents"][0].content == "Hello World\nGoodbye"

def test_regex_replace_custom_replacement(self):
"""Test regex_replace with custom replacement strings."""
cleaner = DocumentCleaner(
remove_empty_lines=False,
remove_extra_whitespaces=False,
regex_replace={r"\[REDACTED\]": "***", r"(\d{4})-(\d{2})-(\d{2})": r"\2/\3/\1"},

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Just a note for our team: This is quite powerful and I think we could include such an example in our documentation. That would make it easier for users to understand what the parameter can be used for.

)
result = cleaner.run(
documents=[Document(content="Name: [REDACTED], Date: 2024-01-15")]
)
assert len(result["documents"]) == 1
assert result["documents"][0].content == "Name: ***, Date: 01/15/2024"

def test_strip_whitespace_and_regex_replace_combined(self):
"""Test using both strip_whitespace and regex_replace together."""
cleaner = DocumentCleaner(
remove_empty_lines=False,
remove_extra_whitespaces=False,
strip_whitespace=True,
regex_replace={r"\n\n+": "\n"},
)
result = cleaner.run(
documents=[Document(content="\n\n Hello\n\n\nWorld \n\n")]
)
assert len(result["documents"]) == 1
assert result["documents"][0].content == "Hello\nWorld"

def test_init_with_new_params(self):
"""Test that new parameters are properly initialized."""
cleaner = DocumentCleaner(strip_whitespace=True, regex_replace={r"\n+": "\n"})
assert cleaner.strip_whitespace is True
assert cleaner.regex_replace == {r"\n+": "\n"}
Loading