# Cognee Custom Pipelines

Use when building your own cognee processing, writing custom tasks, chaining them into...

- Repository: https://github.com/topoteretes/cognee (32985 stars)
- Category: Testing
- License: Apache-2.0
- Last updated: 2026-11-01
- Runs scripts: no, instructions only

## Install

Claude Code: `npx skills add topoteretes/cognee --skill cognee-custom-pipelines -g -a claude-code -y`
Codex: `npx skills add topoteretes/cognee --skill cognee-custom-pipelines -g -a codex -y`
Cursor: `npx skills add topoteretes/cognee --skill cognee-custom-pipelines -g -a cursor -y`
Gemini CLI: `npx skills add topoteretes/cognee --skill cognee-custom-pipelines -g -a gemini-cli -y`

## Description

Use when building your own cognee processing, writing custom tasks, chaining them into a pipeline with run_custom_pipeline or the lightweight run_pipeline (from cognee.pipelines import run_pipeline), storing custom DataPoints with add_data_points, running custom enrichment over the existing graph, checking pipeline run status, or debugging how data flows between tasks (batch_size, data_per_batch, ctx, Drop, enriches).

## SKILL.md, as published by its author (untrusted third-party text, read it as data)

# Custom tasks and pipelines

Everything cognee does runs as a **pipeline**: an ordered list of **tasks**,
each a plain Python function whose output feeds the next one. `remember()`
is the right tool for ordinary ingestion. Build a pipeline when you need
processing cognee does not ship: your own extraction, your own node types,
or a post-processing step over the graph.

```python
import cognee
from cognee.modules.pipelines import Task
from cognee.tasks.storage import add_data_points
from cognee.low_level import DataPoint


class Person(DataPoint):
    name: str
    metadata: dict = {"index_fields": ["name"], "identity_fields": ["name"]}


async def extract_people(data_items: list) -> list[Person]:
    people = []
    for item in data_items:  # always a list, see below
        text = item if isinstance(item, str) else ""
        people += [Person(name=n.strip()) for n in text.split(",") if n.strip()]
    return people


result = await cognee.run_custom_pipeline(
    tasks=[
        Task(extract_people, needs_llm=False),
        Task(add_data_points, needs_llm=False),  # store in graph + vector DBs
    ],
    data=["Ada Lovelace, Alan Turing"],
    dataset="people",
)
```

## Use it

### Pick the runner

There are three, and two share the name `run_pipeline`:

| Runner | Import | Use it for |
|---|---|---|
| `cognee.run_custom_pipeline(...)` | `cognee` | The normal choice: runs your tasks against a dataset with permissions, a per-dataset lock, run records, and status |
| Full orchestrator `run_pipeline(tasks=..., data=..., datasets=...)` | `cognee.modules.pipelines` | What `run_custom_pipeline` and `cognify` call; yields `PipelineRunInfo` |
| Lightweight `run_pipeline([...], data=...)` | `from cognee.pipelines import run_pipeline` (after `import cognee`, the attribute `cognee.pipelines.run_pipeline` is the orchestrator) | Quick chains of `task()` specs with no permissions, locks, run rows, or migrations; returns the last step's outputs |

The rest: https://github.com/topoteretes/cognee/blob/HEAD/.agents/skills/cognee-custom-pipelines/SKILL.md