windlass.providers.preprocessors.clean

clean

Cleaning, language detection and metadata enrichment — all dependency-free.

These are the preprocessors almost every pipeline wants. clean alone typically removes 5-15% of a PDF corpus's characters (headers, page numbers, control codes) without losing meaning, which is 5-15% off your embedding bill.

Example

from windlass.core.types import Document doc = Document(content="Hello world\n\n\n\nAgain") CleanPreprocessor(min_length=0).process([doc])[0].content 'Hello world\n\nAgain'

CleanPreprocessor

CleanPreprocessor(
    *,
    unicode: bool = True,
    whitespace: bool = True,
    strip_html_tags: bool = False,
    remove_urls: bool = False,
    remove_page_numbers: bool = True,
    normalize_bullets: bool = True,
    min_length: int = 20,
    max_length: int | None = None,
    **config: Any
)

Bases: Preprocessor

Normalises text and removes common noise.

Parameters:

Name	Type	Description	Default
unicode	bool	Apply NFKC normalisation and drop control characters. Fixes the ligatures and non-breaking spaces that PDF extraction produces.	True
whitespace	bool	Collapse runs of spaces and blank lines.	True
strip_html_tags	bool	Remove HTML tags left over from a mixed-format source.	False
remove_urls	bool	Delete bare URLs, which embed poorly and add no meaning.	False
remove_page_numbers	bool	Delete lines that are nothing but a page number.	True
normalize_bullets	bool	Rewrite Unicode bullets as -.	True
min_length	int	Drop documents shorter than this after cleaning. This is how blank pages and extraction artefacts get filtered out.	20
max_length	int | None	Truncate documents longer than this. None means no limit.	None
**config	Any	Forwarded to :class:~windlass.interfaces.preprocessor.Preprocessor.	{}

Example

from windlass.core.types import Document p = CleanPreprocessor(remove_urls=True, min_length=0) p.process([Document(content="see https://x.com now")])[0].content 'see now'

Source code in src\windlass\providers\preprocessors\clean.py

def __init__(
    self,
    *,
    unicode: bool = True,
    whitespace: bool = True,
    strip_html_tags: bool = False,
    remove_urls: bool = False,
    remove_page_numbers: bool = True,
    normalize_bullets: bool = True,
    min_length: int = 20,
    max_length: int | None = None,
    **config: Any,
) -> None:
    super().__init__(**config)
    self.unicode = unicode
    self.whitespace = whitespace
    self.strip_html_tags = strip_html_tags
    self.remove_urls = remove_urls
    self.remove_page_numbers = remove_page_numbers
    self.normalize_bullets = normalize_bullets
    self.min_length = min_length
    self.max_length = max_length

aprocess_one async

aprocess_one(document: Document) -> list[Document]

Clean one document.

Parameters:

Name	Type	Description	Default
document	Document	The document to clean.	required

Returns:

Name	Type	Description
	list[Document]	A single-element list, or [] when the cleaned text is shorter
than	list[Document]	attr:min_length.

Source code in src\windlass\providers\preprocessors\clean.py

async def aprocess_one(self, document: Document) -> list[Document]:
    """Clean one document.

    Args:
        document: The document to clean.

    Returns:
        A single-element list, or ``[]`` when the cleaned text is shorter
        than :attr:`min_length`.
    """
    text = document.content
    if self.strip_html_tags and "<" in text:
        text = strip_html(text)
    if self.unicode:
        text = normalize_unicode(text)
    if self.remove_urls:
        text = _URL_RE.sub(" ", text)
    if self.remove_page_numbers:
        text = "\n".join(line for line in text.split("\n") if not _PAGE_NUMBER_RE.match(line))
    if self.normalize_bullets:
        text = _BULLET_RE.sub("- ", text)
    if self.whitespace:
        text = normalize_whitespace(text)
    if self.max_length and len(text) > self.max_length:
        text = text[: self.max_length]

    if len(text) < self.min_length:
        self._log.debug(
            "Dropping %s: %d characters after cleaning (minimum %d).",
            document.source or document.id,
            len(text),
            self.min_length,
        )
        return []

    return [
        document.model_copy(
            update={
                "content": text,
                "metadata": {
                    **document.metadata,
                    "cleaned": True,
                    "original_length": len(document.content),
                },
            }
        )
    ]

LanguagePreprocessor

LanguagePreprocessor(
    *,
    allowed: Sequence[str] | None = None,
    default: str = "en",
    field: str = "language",
    **config: Any
)

Bases: Preprocessor

Tags each document with a detected language, and can filter on it.

Mixed-language corpora quietly degrade retrieval: a monolingual embedding model maps text it does not understand to roughly meaningless vectors. Tag first, then either filter or route to a multilingual model.

Parameters:

Name	Type	Description	Default
allowed	Sequence[str] | None	Language codes to keep. None tags without filtering.	None
default	str	Language assumed when detection is inconclusive.	'en'
field	str	Metadata key to write the detected code to.	'language'
**config	Any	Forwarded to :class:~windlass.interfaces.preprocessor.Preprocessor.	{}

Note

Uses langdetect when installed, and a stop-word plus script heuristic otherwise. The heuristic is good enough for tagging and routing, not for anything adversarial.

Example

from windlass.core.types import Document p = LanguagePreprocessor() p.process([Document(content="the quick brown fox is here")])[0].metadata["language"] 'en'

Source code in src\windlass\providers\preprocessors\clean.py

def __init__(
    self,
    *,
    allowed: Sequence[str] | None = None,
    default: str = "en",
    field: str = "language",
    **config: Any,
) -> None:
    super().__init__(**config)
    self.allowed = {a.lower() for a in allowed} if allowed else None
    self.default = default
    self.field = field

aprocess_one async

aprocess_one(document: Document) -> list[Document]

Detect the language and tag (or drop) the document.

Parameters:

Name	Type	Description	Default
document	Document	The document to inspect.	required

Returns:

Type	Description
list[Document]	The tagged document, or [] when its language is not allowed.

Source code in src\windlass\providers\preprocessors\clean.py

async def aprocess_one(self, document: Document) -> list[Document]:
    """Detect the language and tag (or drop) the document.

    Args:
        document: The document to inspect.

    Returns:
        The tagged document, or ``[]`` when its language is not allowed.
    """
    language = detect_language(document.content, default=self.default)
    if self.allowed is not None and language not in self.allowed:
        self._log.debug(
            "Dropping %s: language %r not in %s",
            document.source or document.id,
            language,
            sorted(self.allowed),
        )
        return []
    return [
        document.model_copy(update={"metadata": {**document.metadata, self.field: language}})
    ]

MetadataPreprocessor

MetadataPreprocessor(
    *,
    extra: dict[str, Any] | None = None,
    keywords: int = 8,
    estimate_tokens: bool = True,
    **config: Any
)

Bases: Preprocessor

Enriches metadata with cheap, useful signals.

Everything here is computed locally with no model call: character and word counts, an estimated token count, a reading-time estimate, and the top keywords by frequency. Those fields make retrieval filters (token_count < 2000) and result display far more useful.

Parameters:

Name	Type	Description	Default
extra	dict[str, Any] | None	Static key/value pairs merged into every document — a tenant id, a corpus name, an ingestion timestamp.	None
keywords	int	How many top keywords to extract. 0 disables it.	8
estimate_tokens	bool	Compute a token count. Costs a tokenisation pass.	True
**config	Any	Forwarded to :class:~windlass.interfaces.preprocessor.Preprocessor.	{}

Example

from windlass.core.types import Document meta = MetadataPreprocessor().process( ... [Document(content="alpha beta alpha gamma")] ... )[0].metadata meta["word_count"] 4

Source code in src\windlass\providers\preprocessors\clean.py

def __init__(
    self,
    *,
    extra: dict[str, Any] | None = None,
    keywords: int = 8,
    estimate_tokens: bool = True,
    **config: Any,
) -> None:
    super().__init__(**config)
    self.extra = dict(extra or {})
    self.keywords = keywords
    self.estimate_tokens = estimate_tokens

aprocess_one async

aprocess_one(document: Document) -> list[Document]

Compute and attach metadata.

Parameters:

Name	Type	Description	Default
document	Document	The document to enrich.	required

Returns:

Type	Description
list[Document]	A single-element list with the enriched document.

Source code in src\windlass\providers\preprocessors\clean.py

async def aprocess_one(self, document: Document) -> list[Document]:
    """Compute and attach metadata.

    Args:
        document: The document to enrich.

    Returns:
        A single-element list with the enriched document.
    """
    text = document.content
    words = text.split()
    metadata: dict[str, Any] = {
        **document.metadata,
        **self.extra,
        "char_count": len(text),
        "word_count": len(words),
        "reading_time_minutes": round(len(words) / 220, 1),
    }
    if self.estimate_tokens:
        metadata["token_count"] = count_tokens(text)
    if self.keywords:
        metadata["keywords"] = _top_keywords(text, self.keywords)
    return [document.model_copy(update={"metadata": metadata})]