Rebuild Add and Cognify by Hand

You need to change something in the middle of cognee’s ingestion — swap a chunker, insert a task of your own, or just see which steps run in what order — and calling add() and cognify() hides all of it behind two function calls.

What You’ll Build

A short paragraph of text about natural language processing goes in, and a queryable knowledge graph comes out — except neither add() nor cognify() is ever called. The add stage is reassembled by hand from the two Task objects it wraps, and the cognify stage is run from the task list cognify() itself would have used, both handed to run_custom_pipeline(). A final GRAPH_COMPLETION search over the result proves the hand-driven graph is an ordinary cognee graph. The complete runnable script is examples/demos/custom_pipelines/custom_cognify_pipeline_example.py — this page walks through its key moments rather than reproducing it.

Features in Play

Custom Tasks and Pipelines — run_custom_pipeline() executes both stages from task lists the script controls
Tasks — Task(...) wraps resolve_data_directories and ingest_data together with the arguments they need
Add — the ingestion stage this demo reconstructs from its two underlying tasks
Cognify — get_default_tasks() hands over the real graph-building task list instead of running it
Inspecting Graph Completion Context — SearchType.GRAPH_COMPLETION queries the graph the custom pipelines built

Before You Start

Complete Quickstart to understand basic operations
Ensure you have LLM Providers configured — the cognify task list makes live LLM calls to extract entities and relationships. The script’s own header asks you to copy .env.template to .env and set LLM_API_KEY
Run it from a checkout of the cognee repo: it imports internals such as cognee.modules.pipelines.Task and cognee.api.v1.cognify.cognify.get_default_tasks, and the command below uses the repo-relative path
The script begins with prune_data() and prune_system(metadata=True), which wipe data and system state including users and pipeline runs — point it at a scratch instance rather than memory you want to keep

How It Works

Stage 1: Reset and Initialize the Databases

    # Create a clean slate for cognee -- reset data and system state
    print("Resetting cognee data...")
    await cognee.prune.prune_data()
    await cognee.prune.prune_system(metadata=True)
    print("Data reset complete.\n")

    # Create relational database and tables
    await setup()

Pruning the system with metadata=True drops the relational database, so the tables the pipelines write to have to be recreated before anything runs. setup() is the call the high-level operations make for you; driving pipelines directly means making it yourself.

Stage 2: Rebuild the Add Stage from Tasks

    # Let's recreate the cognee add pipeline through the custom pipeline framework
    from cognee.tasks.ingestion import ingest_data, resolve_data_directories

    user = await get_default_user()

    # Values for tasks need to be filled before calling the pipeline
    add_tasks = [
        Task(resolve_data_directories, include_subdirectories=True),
        Task(
            ingest_data,
            "main_dataset",
            user,
        ),
    ]
    # Forward tasks to custom pipeline along with data and user information
    await cognee.run_custom_pipeline(
        tasks=add_tasks, data=text, user=user, dataset="main_dataset", pipeline_name="add_pipeline"
    )

This is what add() does underneath: resolve whatever was passed into concrete data items, then ingest them into a dataset. A Task carries its own arguments — the dataset name and user are bound into ingest_data here — while the pipeline supplies the data flowing through. Naming the run add_pipeline keeps it identifiable in the pipeline-run history.

Stage 3: Borrow the Default Cognify Task List

    from cognee.api.v1.cognify.cognify import get_default_tasks

    cognify_tasks = await get_default_tasks(user=user)
    print("Recreating existing cognify pipeline in custom pipeline to create knowledge graph...\n")
    await cognee.run_custom_pipeline(
        tasks=cognify_tasks, user=user, dataset="main_dataset", pipeline_name="cognify_pipeline"
    )

Rather than hand-writing the graph-building steps, the script asks cognify for its own task list — document classification, chunking, entity extraction, summarization, storage — and runs it through the same custom pipeline entry point. No data argument is needed: the tasks pick up the data items the add pipeline already wrote to main_dataset.

Stage 4: Query the Hand-Built Graph

    query_text = "Tell me about NLP"
    print(f"Searching cognee for insights with query: '{query_text}'")
    # Query cognee for insights on the added text
    search_results = await cognee.search(
        query_type=SearchType.GRAPH_COMPLETION, query_text=query_text
    )

An ordinary search() closes the loop. Nothing about the query is aware that the graph was built task by task — which is the point of the demo: the custom pipeline path produces the same graph the built-in operations do.

Run It

uv run python examples/demos/custom_pipelines/custom_cognify_pipeline_example.py

A successful run prints the reset messages first, then echoes the NLP paragraph it is adding and confirms Text added successfully. once the add pipeline finishes. The cognify pipeline announces that it is recreating the existing cognify pipeline, takes the longest as it makes its LLM calls, and ends with Cognify process complete. The script then reports the query it is running and prints Search results: followed by the generated answer about natural language processing.

Where to Change It

Both stages are plain Python lists of Task objects, so this is the shape to start from when the built-in flow is close to what you want but not exact. get_default_tasks() returns the cognify list, and you can reorder it, drop a task, or splice your own in before passing it to run_custom_pipeline() — see Custom Tasks and Pipelines for writing that task. It also takes the knobs cognify takes, including graph_model and chunker, if adjusting the arguments is enough.

Custom Tasks and Pipelines

Write your own task and run it in a pipeline of your own.

Tasks

What a Task wraps, and how its arguments and batching work.

run_custom_pipeline()

Every parameter of the entry point both stages are run through.

Cognify

The operation whose default task list this demo runs by hand.

Getting Started

Core Concepts

Setup Configuration

Guides

Examples

CLI

Rust SDK

TypeScript SDK

OSS

What You’ll Build

Features in Play

Before You Start

How It Works

Stage 1: Reset and Initialize the Databases

Stage 2: Rebuild the Add Stage from Tasks

Stage 3: Borrow the Default Cognify Task List

Stage 4: Query the Hand-Built Graph

Run It

Where to Change It

Custom Tasks and Pipelines

Tasks

run_custom_pipeline()

Cognify

​What You’ll Build

​Features in Play

​Before You Start

​How It Works

​Stage 1: Reset and Initialize the Databases

​Stage 2: Rebuild the Add Stage from Tasks

​Stage 3: Borrow the Default Cognify Task List

​Stage 4: Query the Hand-Built Graph

​Run It

​Where to Change It

Custom Tasks and Pipelines

Tasks

run_custom_pipeline()

Cognify

What You’ll Build

Features in Play

Before You Start

How It Works

Stage 1: Reset and Initialize the Databases

Stage 2: Rebuild the Add Stage from Tasks

Stage 3: Borrow the Default Cognify Task List

Stage 4: Query the Hand-Built Graph

Run It

Where to Change It