windlass.interfaces.preprocessor

preprocessor

The preprocessor interface.

Preprocessors sit between loading and chunking. They clean, enrich, filter or redact documents, and they compose: a pipeline runs them in order, each seeing what the previous one produced.

A preprocessor may return zero documents (dropping the input — that is how deduplication and language filtering work), one (the common case), or many (splitting a spreadsheet into per-sheet documents).

Implementers override one coroutine, :meth:Preprocessor.aprocess_one.

Example

from windlass.providers.preprocessors.clean import CleanPreprocessor from windlass.core.types import Document clean = CleanPreprocessor(min_length=0) docs = clean.process([Document(content="a \t b")]) docs[0].content 'a b'

Preprocessor

Preprocessor(*, name: str | None = None, **config: Any)

Bases: Component

Abstract document preprocessor.

Parameters:

Name	Type	Description	Default
name	str | None	Component name for traces.	None
**config	Any	Preprocessor-specific options.	{}

Example

Implementing a preprocessor takes one method::

class Shout(Preprocessor):
    provider_name = "shout"

    async def aprocess_one(self, document):
        return [document.model_copy(
            update={"content": document.content.upper()}
        )]

Source code in src\windlass\interfaces\base.py

def __init__(self, *, name: str | None = None, **config: Any) -> None:
    self.name = name or getattr(type(self), "provider_name", None) or type(self).__name__
    self.config: dict[str, Any] = dict(config)
    self._log = get_logger(f"{self.kind}.{self.name}")

aprocess_one abstractmethod async

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

Transform a single document.

Parameters:

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

Returns:

Type	Description
list[Document]	Zero, one or many documents. Return [] to drop the input.

Raises:

Type	Description
IngestionError	When the document cannot be processed and dropping it silently would hide a real problem.

Source code in src\windlass\interfaces\preprocessor.py

@abc.abstractmethod
async def aprocess_one(self, document: Document) -> list[Document]:
    """Transform a single document.

    Args:
        document: The document to process.

    Returns:
        Zero, one or many documents. Return ``[]`` to drop the input.

    Raises:
        IngestionError: When the document cannot be processed and dropping
            it silently would hide a real problem.
    """

aprocess async

aprocess(
    documents: Sequence[Document], *, concurrency: int | None = None
) -> list[Document]

Process a batch of documents concurrently.

Parameters:

Name	Type	Description	Default
documents	Sequence[Document]	The documents to process.	required
concurrency	int | None	Maximum simultaneous invocations. Defaults to the global max_concurrency setting.	None

Returns:

Type	Description
list[Document]	The flattened output, preserving input order.

Performance

Order is preserved even though execution is concurrent, so a preprocessor that calls an LLM (metadata extraction, contextual enrichment) still yields a deterministic corpus.

Source code in src\windlass\interfaces\preprocessor.py

async def aprocess(
    self, documents: Sequence[Document], *, concurrency: int | None = None
) -> list[Document]:
    """Process a batch of documents concurrently.

    Args:
        documents: The documents to process.
        concurrency: Maximum simultaneous invocations. Defaults to the
            global ``max_concurrency`` setting.

    Returns:
        The flattened output, preserving input order.

    Performance:
        Order is preserved even though execution is concurrent, so a
        preprocessor that calls an LLM (metadata extraction, contextual
        enrichment) still yields a deterministic corpus.
    """
    if not documents:
        return []
    limit = concurrency or settings().max_concurrency
    batches = await gather_bounded([self.aprocess_one(doc) for doc in documents], limit=limit)
    return [doc for batch in batches for doc in batch]

process

process(documents: Sequence[Document] | Document) -> list[Document]

Blocking :meth:aprocess.

Parameters:

Name	Type	Description	Default
documents	Sequence[Document] | Document	One document or a sequence of them.	required

Returns:

Type	Description
list[Document]	The processed documents.

Source code in src\windlass\interfaces\preprocessor.py

def process(self, documents: Sequence[Document] | Document) -> list[Document]:
    """Blocking :meth:`aprocess`.

    Args:
        documents: One document or a sequence of them.

    Returns:
        The processed documents.
    """
    items = [documents] if isinstance(documents, Document) else list(documents)
    return run_sync(self.aprocess(items))

PreprocessorChain

PreprocessorChain(steps: Sequence[Preprocessor] = (), *, name: str | None = None)

Bases: Preprocessor

Runs several preprocessors in sequence.

Each step sees the full output of the previous one, which matters for corpus-level steps like deduplication that need to compare documents against each other rather than in isolation.

Parameters:

Name	Type	Description	Default
steps	Sequence[Preprocessor]	The preprocessors to run, in order.	()
name	str | None	Component name for traces.	None

Attributes:

Name	Type	Description
steps	list[Preprocessor]	The configured steps.

Example

from windlass.providers.preprocessors.clean import CleanPreprocessor chain = PreprocessorChain([CleanPreprocessor(min_length=0)]) from windlass.core.types import Document chain.process([Document(content=" hi ")])[0].content 'hi'

Source code in src\windlass\interfaces\preprocessor.py

def __init__(self, steps: Sequence[Preprocessor] = (), *, name: str | None = None) -> None:
    super().__init__(name=name or "chain")
    self.steps: list[Preprocessor] = list(steps)

add

add(step: Preprocessor) -> PreprocessorChain

Append a step and return self for chaining.

Source code in src\windlass\interfaces\preprocessor.py

def add(self, step: Preprocessor) -> PreprocessorChain:
    """Append a step and return ``self`` for chaining."""
    self.steps.append(step)
    return self

aprocess_one async

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

Run every step against a single document.

Source code in src\windlass\interfaces\preprocessor.py

async def aprocess_one(self, document: Document) -> list[Document]:
    """Run every step against a single document."""
    return await self.aprocess([document])

aprocess async

aprocess(
    documents: Sequence[Document], *, concurrency: int | None = None
) -> list[Document]

Run every step in order over the whole batch.

Parameters:

Name	Type	Description	Default
documents	Sequence[Document]	The documents to process.	required
concurrency	int | None	Forwarded to each step.	None

Returns:

Type	Description
list[Document]	The output of the final step. Short-circuits to [] as soon as a
list[Document]	step drops everything.

Source code in src\windlass\interfaces\preprocessor.py

async def aprocess(
    self, documents: Sequence[Document], *, concurrency: int | None = None
) -> list[Document]:
    """Run every step in order over the whole batch.

    Args:
        documents: The documents to process.
        concurrency: Forwarded to each step.

    Returns:
        The output of the final step. Short-circuits to ``[]`` as soon as a
        step drops everything.
    """
    current = list(documents)
    for step in self.steps:
        if not current:
            break
        current = await step.aprocess(current, concurrency=concurrency)
    return current

describe

describe() -> dict[str, Any]

Return a summary including each step.

Source code in src\windlass\interfaces\preprocessor.py

def describe(self) -> dict[str, Any]:
    """Return a summary including each step."""
    return {**super().describe(), "steps": [s.describe() for s in self.steps]}