- name
- cognee-custom-pipelines
- 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).
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:
returns (the started run when ). With it runs over the dataset's existing documents ( rows).