# How to Use Conditional Task Execution with the Condition Parameter in Sieves

> Master conditional task execution in Sieves. Learn how to use the condition parameter to skip documents per task and control your workflow efficiently.

- Repository: [Mantis/sieves](https://github.com/mantisai/sieves)
- Tags: how-to-guide
- Published: 2026-03-06

---

**You can skip documents on a per-task basis by passing a callable to the `condition` parameter of any `Task` in Sieves, which evaluates each document and stores `None` for skipped items while preserving the original document order.**

Sieves is a modular document processing framework that supports **conditional task execution** through a flexible `condition` parameter available on all tasks. This feature allows you to filter documents dynamically based on custom logic, ensuring that expensive operations like model inference only run on relevant documents. By leveraging the `condition` parameter in the `mantisai/sieves` repository, you can build efficient pipelines that skip unnecessary processing without breaking the document stream.

## How Conditional Task Execution Works in Sieves

### The Task Base Class and Condition Storage

The conditional logic is implemented in the base `Task` class defined in [`sieves/tasks/core.py`](https://github.com/mantisai/sieves/blob/main/sieves/tasks/core.py). When initializing any task, the `condition` parameter is stored as `self._condition`:

```python

# sieves/tasks/core.py

class Task(abc.ABC):
    def __init__(self, task_id, include_meta, batch_size, condition=None):
        # ...

        self._condition = condition                         # ← stored here

```

This design ensures that all subclasses—including built-in tasks like `Classification` and `Chunking`—inherit conditional execution capabilities without additional implementation.

### The Execution Flow and Document Filtering

When a pipeline invokes a task, the `__call__` method (lines 68-88 in [`sieves/tasks/core.py`](https://github.com/mantisai/sieves/blob/main/sieves/tasks/core.py)) handles conditional filtering automatically. The method evaluates the condition for each document in the batch to determine which items to process:

```python

# sieves/tasks/core.py (lines 68-88)

while docs_batch := [doc for doc in itertools.islice(docs, batch_size)]:
    passing_indices = {
        idx for idx, doc in enumerate(docs_batch)
        if self._condition is None or self._condition(doc)
    }
    processed = self._call(d for i, d in enumerate(docs_batch) if i in passing_indices)
    for idx, doc in enumerate(docs_batch):
        if idx in passing_indices:
            yield next(processed)          # run the task

        else:
            doc.results[self.id] = None    # skipped → store None

            yield doc

```

Documents that fail the condition check receive `None` in their `results` dictionary under the task ID, while passing documents are processed normally. The original document order is preserved throughout the pipeline.

The pipeline itself simply forwards documents to each task. Because each task handles its own conditional logic in [`sieves/pipeline/core.py`](https://github.com/mantisai/sieves/blob/main/sieves/pipeline/core.py), no extra pipeline configuration is required:

```python

# sieves/pipeline/core.py

for i, task in enumerate(self._tasks):
    processed_docs = task(processed_docs)   # ← each task decides what to do

```

## When to Use Conditional Task Execution

Conditional task execution is particularly valuable when processing heterogeneous document collections. Common scenarios include:

| Situation | Example Condition |
|-----------|-------------------|
| **Skip short texts** | `lambda d: len(d.text or "") > 20` |
| **Process only PDFs** | `lambda d: d.uri and d.uri.endswith('.pdf')` |
| **Rate limiting** | `lambda d, i=itertools.count(): next(i) < N` |
| **External data validation** | Any callable that queries external APIs or databases to determine eligibility |

## Implementation Examples

### Custom Task with a Condition

The following example demonstrates a custom task that only processes documents longer than 20 characters. This pattern is tested in [`sieves/tests/tasks/test_conditional_execution.py`](https://github.com/mantisai/sieves/blob/main/sieves/tests/tasks/test_conditional_execution.py).

```python
from sieves.tasks.core import Task
from sieves.data import Doc

class DummyTask(Task):
    def __init__(self, condition=None):
        super().__init__(task_id="DummyTask", include_meta=False,
                         batch_size=-1, condition=condition)

    def _call(self, docs):
        for doc in docs:
            doc.results[self.id] = {"processed": True}
            yield doc

```

To use this task with conditional execution:

```python
from sieves.pipeline import Pipeline
from sieves import Doc

docs = [Doc(text="short"), Doc(text="this is a much longer document")]
task = DummyTask(condition=lambda d: len(d.text or "") > 20)

pipe = Pipeline([task])
for d in pipe(docs):
    print(d.results["DummyTask"])

# Output: None for the short doc, {"processed": True} for the long one

```

### Built-in Classification with Conditional Logic

Built-in predictive tasks like `Classification` inherit the `condition` parameter from the base `Task` class. This example from [`sieves/tests/docs/test_pipeline_docs.py`](https://github.com/mantisai/sieves/blob/main/sieves/tests/docs/test_pipeline_docs.py) shows how to conditionally classify only long documents.

```python
from sieves import Pipeline, tasks, Doc
from sieves.models import SmallTransformer  # any ModelWrapper implementation

model = SmallTransformer()
docs = [
    Doc(text="short"),
    Doc(text="this is a much longer document that will be processed"),
    Doc(text="med")
]

def is_long(doc: Doc) -> bool:
    return len(doc.text or "") > 20

# Pass condition directly to the task constructor

task = tasks.Classification(labels=["science", "politics"],
                            model=model,
                            condition=is_long)

pipe = Pipeline([task])
for d in pipe(docs):
    print(d.results[task.id])   # None for short/med docs, prediction dict for the long one

```

### Multiple Tasks with Independent Conditions

You can assign different conditions to different tasks within the same pipeline. Each task evaluates its condition independently, allowing complex processing workflows.

```python
from sieves import Pipeline, tasks, Doc
from sieves.models import SmallTransformer
from sieves.chunkers import example_chunker

model = SmallTransformer()
docs = [
    Doc(text="short"),
    Doc(text="this is a much longer document"),
    Doc(text="medium text here")
]

# Task 1: Only chunk documents longer than 10 characters

task1 = tasks.Chunking(example_chunker,
                       condition=lambda d: len(d.text or "") > 10)

# Task 2: Only classify documents longer than 20 characters

task2 = tasks.Classification(labels=["science", "politics"],
                             model=model,
                             condition=lambda d: len(d.text or "") > 20)

pipe = Pipeline([task1, task2])
for d in pipe(docs):
    print(d.chunks, d.results.get(task2.id))

```

## Key Implementation Files

The conditional execution system is implemented across these core files in the `mantisai/sieves` repository:

| File | Purpose | Link |
|------|---------|------|
| [`sieves/tasks/core.py`](https://github.com/mantisai/sieves/blob/main/sieves/tasks/core.py) | Defines the base `Task` class with `condition` parameter and the `__call__` method that filters documents per batch. | [View source](https://github.com/mantisai/sieves/blob/main/sieves/tasks/core.py) |
| [`sieves/pipeline/core.py`](https://github.com/mantisai/sieves/blob/main/sieves/pipeline/core.py) | Orchestrates task execution; forwards documents to each task which handles its own conditional logic independently. | [View source](https://github.com/mantisai/sieves/blob/main/sieves/pipeline/core.py) |
| [`sieves/tests/tasks/test_conditional_execution.py`](https://github.com/mantisai/sieves/blob/main/sieves/tests/tasks/test_conditional_execution.py) | Unit tests demonstrating custom task behavior with conditional execution. | [View source](https://github.com/mantisai/sieves/blob/main/sieves/tests/tasks/test_conditional_execution.py) |
| [`sieves/tests/docs/test_pipeline_docs.py`](https://github.com/mantisai/sieves/blob/main/sieves/tests/docs/test_pipeline_docs.py) | Integration examples showing built-in tasks with conditions in pipeline contexts. | [View source](https://github.com/mantisai/sieves/blob/main/sieves/tests/docs/test_pipeline_docs.py) |
| [`sieves/tasks/predictive/classification/core.py`](https://github.com/mantisai/sieves/blob/main/sieves/tasks/predictive/classification/core.py) | Example of a predictive task inheriting conditional capabilities from the base `Task` class. | [View source](https://github.com/mantisai/sieves/blob/main/sieves/tasks/predictive/classification/core.py) |

## Summary

- The `condition` parameter in Sieves accepts any callable that takes a `Doc` object and returns a boolean, enabling **conditional task execution** on a per-document basis.
- When a document fails the condition check, the task stores `None` in `doc.results[task.id]` and yields the document unchanged, preserving the original stream order.
- All tasks—including custom subclasses and built-in predictive tasks like `Classification`—inherit this functionality from the base `Task` class in [`sieves/tasks/core.py`](https://github.com/mantisai/sieves/blob/main/sieves/tasks/core.py).
- Conditions are evaluated independently for each task in a pipeline, allowing different filtering logic for chunking, classification, or custom processing steps.
- Skipped documents remain in the pipeline for subsequent tasks, making the system cache-friendly and composable.

## Frequently Asked Questions

### What happens to documents that don't meet the condition?

Documents that fail the condition check are not processed by the task's `_call` method. Instead, the task automatically sets `doc.results[self.id] = None` and yields the document unchanged. This ensures the document remains in the pipeline for subsequent tasks while maintaining the original order of the document stream.

### Can I use different conditions for different tasks in the same pipeline?

Yes, each task in a Sieves pipeline maintains its own independent `condition` callable. You can configure `Task A` to process only PDFs while `Task B` processes only long documents, and both will execute their conditions separately as the pipeline forwards documents through each step.

### Does the condition parameter affect batch processing?

The condition is evaluated within the batch processing loop in `Task.__call__`. For each batch, the task identifies which documents satisfy the condition (`passing_indices`) and only processes those, while still yielding skipped documents to maintain stream continuity. This means batching efficiency is preserved while allowing per-document filtering.

### How do I access the document index in my condition function?

The condition callable receives only the `Doc` object as its argument by default. If you need access to the document index, you can use a closure with `itertools.count()` or a class-based callable that maintains state. For example: `lambda d, i=itertools.count(): next(i) < 100` processes only the first 100 documents.