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.

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