# How to Run Custom Pipelines in Cognee: A Complete Guide

> Learn to run custom pipelines in Cognee by wrapping functions in Task objects and using cognee.run_custom_pipeline(). Master background and blocking execution modes for efficient workflow orchestration.

- Repository: [Topoteretes/cognee](https://github.com/topoteretes/cognee)
- Tags: how-to-guide
- Published: 2026-03-16

---

**Run custom pipelines in Cognee by wrapping functions in `Task` objects and passing them to `cognee.run_custom_pipeline()`, which orchestrates execution through the core `run_pipeline` engine with support for blocking or background modes.**

Cognee is an open-source knowledge graph engine that enables developers to build complex data processing workflows. While the framework provides built-in CLI commands for common operations, the `run_custom_pipeline` interface in `topoteretes/cognee` allows you to compose arbitrary sequences of tasks for specialized ingestion, enrichment, or graph construction needs. This guide explains how to construct and execute custom pipelines using the underlying Task API and pipeline execution layers.

## Understanding the Custom Pipeline Architecture

Custom pipelines in Cognee are built on three foundational components: the `Task` wrapper, the orchestration layer, and the core execution engine. Understanding how these pieces fit together enables you to construct everything from simple text ingestion to complex LLM-driven knowledge graph creation.

### Task Definition with the Task Class

Every step in a Cognee pipeline is encapsulated by a `Task` object defined in [`cognee/modules/pipelines/tasks/task.py`](https://github.com/topoteretes/cognee/blob/main/cognee/modules/pipelines/tasks/task.py). This wrapper records the callable function, default arguments, batch size, and determines execution strategy—whether the underlying code is a standard function, coroutine, generator, or async generator.

The `Task` class abstracts away execution complexity, allowing you to focus on business logic. When you instantiate a `Task`, you provide the function reference and any default parameters, which the pipeline engine later invokes with runtime data.

### Pipeline Orchestration Layer

The `run_custom_pipeline` function in [`cognee/modules/run_custom_pipeline/run_custom_pipeline.py`](https://github.com/topoteretes/cognee/blob/main/cognee/modules/run_custom_pipeline/run_custom_pipeline.py) serves as the primary entry point for custom workflows. This function performs three critical operations:

- **Task normalization** – Validates and prepares the list of `Task` objects for execution
- **Executor selection** – Retrieves a pipeline executor via `get_pipeline_executor` that determines whether the pipeline runs blocking or in the background (defined in [`cognee/modules/pipelines/layers/pipeline_execution_mode.py`](https://github.com/topoteretes/cognee/blob/main/cognee/modules/pipelines/layers/pipeline_execution_mode.py))
- **Core delegation** – Forwards all parameters (tasks, data, dataset, user, configuration flags) to the central `run_pipeline` function

### Core Runner and Execution Flow

The `run_pipeline` function in [`cognee/modules/pipelines/operations/pipeline.py`](https://github.com/topoteretes/cognee/blob/main/cognee/modules/pipelines/operations/pipeline.py) implements the actual execution logic. It validates the task list through `validate_pipeline_tasks`, pre-checks the environment via `setup_and_check_environment`, and resolves authorized datasets for the supplied user through `resolve_authorized_user_datasets`.

For each authorized dataset, the runner streams `PipelineRunInfo` objects while delegating work to `run_tasks`. This architecture yields progress updates, start/completion events, and error states, enabling real-time monitoring of long-running operations.

## Execution Modes and Configuration Options

Cognee supports two primary execution strategies for custom pipelines, controlled through the orchestration layer in [`pipeline_execution_mode.py`](https://github.com/topoteretes/cognee/blob/main/pipeline_execution_mode.py).

### Blocking vs. Background Execution

**Blocking execution** (`run_pipeline_blocking`) runs the coroutine directly, yielding results to the caller until completion. This mode is suitable for scripts and notebooks where you need immediate confirmation of success or failure.

**Background execution** (`run_pipeline_as_background_process`) launches the pipeline in a separate async task, immediately returning a `pipeline_run_id` that you can use to monitor progress later. Enable this mode by setting `run_in_background=True` when calling `run_custom_pipeline`.

### Performance and Caching Controls

The pipeline engine exposes several optimization flags:

- **`use_pipeline_cache`** – Skips re-processing if a pipeline with the same ID has already run or is currently running
- **`incremental_loading`** – Processes only newly added or changed data, requiring Cognee’s `Data` model to track state
- **`data_per_batch`** – Controls parallel batch size for task execution, tuning throughput for I/O-bound operations

## Code Examples

### Minimal Custom Pipeline for Text Ingestion

The following example demonstrates adding plain text to Cognee using the Task API:

```python
import asyncio
import cognee
from cognee.modules.pipelines import Task
from cognee.modules.users.methods import get_default_user

async def main() -> None:
    user = await get_default_user()

    # Define the tasks that compose the pipeline

    add_tasks = [
        Task(cognee.modules.tasks.ingestion.resolve_data_directories, include_subdirectories=True),
        Task(cognee.modules.tasks.ingestion.ingest_data, "main_dataset", user),
    ]

    # Run the pipeline – blocking mode (default)

    await cognee.run_custom_pipeline(
        tasks=add_tasks,
        data="Natural language processing (NLP) is a subfield of AI.",
        user=user,
        dataset="main_dataset",
        pipeline_name="add_text",
    )
    print("✅ Text added to Cognee")

if __name__ == "__main__":
    asyncio.run(main())

```

Key implementation details:

- `Task` objects wrap low-level ingestion helpers from `cognee.modules.tasks.ingestion`
- `run_custom_pipeline` forwards the `data` parameter to the first task (`resolve_data_directories`)
- The call blocks until completion because `run_in_background` defaults to `False`

### Background Execution with Caching and Incremental Loading

For production workflows processing large datasets, use background execution with caching:

```python
import asyncio, cognee
from cognee.modules.pipelines import Task
from cognee.modules.users.methods import get_default_user

async def main():
    user = await get_default_user()

    # Re‑use the default Cognify tasks (LLM‑driven KG creation)

    cognify_tasks = await cognee.api.v1.cognify.cognify.get_default_tasks(user=user)

    # Launch the pipeline in the background, enable cache & incremental mode

    pipeline_info = await cognee.run_custom_pipeline(
        tasks=cognify_tasks,
        user=user,
        dataset="main_dataset",
        pipeline_name="cognify_pipeline",
        use_pipeline_cache=True,
        incremental_loading=True,
        run_in_background=True,   # ← fire‑and‑forget

    )
    print(f"🚀 Background pipeline started – run_id={pipeline_info.pipeline_run_id}")
    # You can later inspect status with `cognee.get_pipeline_status(pipeline_info.pipeline_run_id)`

if __name__ == "__main__":
    asyncio.run(main())

```

This configuration:

- Returns immediately with a `PipelineRunStarted` object containing the `pipeline_run_id`
- Avoids duplicate work through `use_pipeline_cache`
- Processes only new documents via `incremental_loading=True`

### Full-Featured Example

For a complete end-to-end demonstration, reference [`examples/custom_pipelines/custom_cognify_pipeline_example.py`](https://github.com/topoteretes/cognee/blob/main/examples/custom_pipelines/custom_cognify_pipeline_example.py) in the Cognee repository. This script illustrates the full workflow:

1. Reset Cognee state using `cognee.prune.prune_data()`
2. Setup a relational database via `cognee.modules.engine.operations.setup.setup`
3. Create and run an "add" pipeline using `Task` objects
4. Execute built-in Cognify tasks inside a custom pipeline
5. Query the resulting graph with `cognee.search`

Copy this example, adjust the `LLM_API_KEY` in your `.env` file, and run it directly to observe custom pipeline behavior in action.

## Summary

- **Custom pipelines** in Cognee are constructed by chaining `Task` objects and executing them via `cognee.run_custom_pipeline()`
- The **Task class** in [`cognee/modules/pipelines/tasks/task.py`](https://github.com/topoteretes/cognee/blob/main/cognee/modules/pipelines/tasks/task.py) wraps functions, coroutines, and generators with execution metadata
- **Execution mode** is determined by `get_pipeline_executor` in [`pipeline_execution_mode.py`](https://github.com/topoteretes/cognee/blob/main/pipeline_execution_mode.py), supporting both blocking and background processing
- **Core orchestration** happens in [`cognee/modules/pipelines/operations/pipeline.py`](https://github.com/topoteretes/cognee/blob/main/cognee/modules/pipelines/operations/pipeline.py), which streams `PipelineRunInfo` objects for progress monitoring
- **Performance features** include `use_pipeline_cache` for deduplication, `incremental_loading` for delta processing, and `data_per_batch` for parallelism tuning

## Frequently Asked Questions

### What is the difference between `run_custom_pipeline` and the built-in CLI commands?

`run_custom_pipeline` in [`cognee/modules/run_custom_pipeline/run_custom_pipeline.py`](https://github.com/topoteretes/cognee/blob/main/cognee/modules/run_custom_pipeline/run_custom_pipeline.py) provides programmatic access to Cognee's pipeline engine, allowing you to compose arbitrary sequences of `Task` objects for specific data processing needs. CLI commands wrap predefined task sequences for standard workflows, while the custom pipeline API enables ad-hoc ingestion, custom LLM-based enrichment, or partial graph reconstruction without modifying the framework's core code.

### How do I monitor a pipeline running in background mode?

When you set `run_in_background=True`, `run_custom_pipeline` returns a `PipelineRunStarted` object containing a `pipeline_run_id`. You can later query the status of this execution using `cognee.get_pipeline_status(pipeline_run_id)` to retrieve progress updates, completion status, or error states. The underlying `run_pipeline` function yields `PipelineRunInfo` objects that track the pipeline lifecycle through the streaming interface implemented in [`cognee/modules/pipelines/operations/pipeline.py`](https://github.com/topoteretes/cognee/blob/main/cognee/modules/pipelines/operations/pipeline.py).

### Can I mix built-in Cognify tasks with custom functions in the same pipeline?

Yes. Retrieve the default Cognify task list via `cognee.api.v1.cognify.cognify.get_default_tasks(user=user)`, then append or prepend your own `Task` objects before passing the combined list to `run_custom_pipeline`. This approach allows you to insert custom preprocessing steps before LLM-driven knowledge graph construction or add post-processing logic after the standard Cognify workflow completes.

### What happens when `use_pipeline_cache` is enabled?

When `use_pipeline_cache=True`, the pipeline engine checks whether a pipeline with the same ID has already run or is currently running. If a cached result exists, the engine skips re-processing and returns the existing output. This optimization is particularly valuable for expensive operations like LLM-based entity extraction or large-scale document ingestion where duplicate work would waste computational resources and API costs.