Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
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
30 changes: 30 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_whitespaces: bool = False,
replace_regexes: 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_whitespaces: 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 replace_regexes: 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_whitespaces = strip_whitespaces
self.replace_regexes = replace_regexes

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.replace_regexes:
text = self._replace_regexes(text, self.replace_regexes)
if self.remove_repeated_substrings:
text = self._remove_repeated_substrings(text)
if self.strip_whitespaces:
text = text.strip()

clean_doc = Document(
id=doc.id if self.keep_id else "",
Expand Down Expand Up @@ -204,6 +218,22 @@ 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_regexes(self, text: str, replace_regexes: dict[str, str]) -> str:
"""
Replace substrings that match the specified regex patterns with custom replacement strings.

:param text: Text to clean.
:param replace_regexes: 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.
pages = text.split("\f")
cleaned_pages = []
for page in pages:
for pattern, replacement in replace_regexes.items():
page = re.sub(pattern, replacement, page)
cleaned_pages.append(page)
return "\f".join(cleaned_pages)

def _remove_substrings(self, text: str, substrings: list[str]) -> str:
"""
Remove all specified substrings from the text.
Expand Down
Original file line number Diff line number Diff line change
@@ -0,0 +1,30 @@
---
enhancements:
- |
Add ``strip_whitespaces`` and ``replace_regexes`` parameters to DocumentCleaner component.

The ``strip_whitespaces`` parameter removes leading and trailing whitespace from document
content using Python's ``str.strip()``method. Unlike ``remove_extra_whitespaces``, this only
affects the beginning and end of the text, preserving internal whitespace which is useful
for maintaining markdown formatting.

The ``replace_regexes`` parameter accepts a dictionary mapping regex patterns to replacement
strings, allowing custom text transformations. For example, ``{r'\\n\\n+': '\\n'}`` replaces
multiple consecutive newlines with a single newline. This is applied after ``remove_regex``
and provides more flexibility than simple pattern removal.

Example usage:

.. code:: python

from haystack.components.preprocessors import DocumentCleaner
from haystack.dataclasses import Document

cleaner = DocumentCleaner(
strip_whitespaces=True,
replace_regexes={r'\n\n+': '\n'}
)

doc = Document(content=" \n\nHello World\n\n\n ")
result = cleaner.run(documents=[doc])
# Result: "Hello World\n"
88 changes: 88 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,91 @@ 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_whitespaces(self):
"""Test that strip_whitespaces removes only leading and trailing whitespace."""
cleaner = DocumentCleaner(remove_empty_lines=False, remove_extra_whitespaces=False, strip_whitespaces=True)
result = cleaner.run(documents=[Document(content=" \n\nHello World\n\n Some text here \n\n ")])
assert len(result["documents"]) == 1
# strip_whitespaces should only remove leading/trailing whitespace, preserving internal whitespace
assert result["documents"][0].content == "Hello World\n\n Some text here"

def test_strip_whitespaces_preserves_internal_formatting(self):
"""Test that strip_whitespaces preserves internal whitespace like markdown formatting."""
cleaner = DocumentCleaner(remove_empty_lines=False, remove_extra_whitespaces=False, strip_whitespaces=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_replace_regexes_single_pattern(self):
"""Test replace_regexes with a single pattern."""
cleaner = DocumentCleaner(
remove_empty_lines=False, remove_extra_whitespaces=False, replace_regexes={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_replace_regexes_multiple_patterns(self):
"""Test replace_regexes with multiple patterns."""
cleaner = DocumentCleaner(
remove_empty_lines=False, remove_extra_whitespaces=False, replace_regexes={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_replace_regexes_custom_replacement(self):
"""Test replace_regexes with custom replacement strings."""
cleaner = DocumentCleaner(
remove_empty_lines=False,
remove_extra_whitespaces=False,
replace_regexes={r"\[REDACTED\]": "***", r"(\d{4})-(\d{2})-(\d{2})": r"\2/\3/\1"},
)
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_whitespaces_and_replace_regexes_combined(self):
"""Test using both strip_whitespaces and replace_regexes together."""
cleaner = DocumentCleaner(
remove_empty_lines=False,
remove_extra_whitespaces=False,
strip_whitespaces=True,
replace_regexes={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_whitespaces=True, replace_regexes={r"\n+": "\n"})
assert cleaner.strip_whitespaces is True
assert cleaner.replace_regexes == {r"\n+": "\n"}

def test_replace_regexes_with_page_breaks(self):
"""Test replace_regexes with page breaks (form feed character)."""
cleaner = DocumentCleaner(
remove_empty_lines=False, remove_extra_whitespaces=False, replace_regexes={r"Page \d+": ""}
)
content = "Page 1 content.\fPage 2 content."
result = cleaner.run(documents=[Document(content=content)])
assert len(result["documents"]) == 1
assert result["documents"][0].content == " content.\f content."