> ## Documentation Index
> Fetch the complete documentation index at: https://docs.cognee.ai/llms.txt
> Use this file to discover all available pages before exploring further.
# Pipelines
> Orchestrate tasks into coordinated data processing workflows.
## What pipelines are
Pipelines coordinate ordered [Tasks](../building-blocks/tasks) into a reproducible workflow. Default Cognee operations like [Remember](../main-operations/remember) run on top of the same execution layer. You typically do not call low-level functions directly; you trigger pipelines through the higher-level operations unless you need staged control.
## Prerequisites
* **Dataset**: a container (name or UUID) where your data is stored and processed. Every document remembered by Cognee belongs to a dataset.
* **User**: the identity for ownership and access control. A default user is created and used if none is provided.
* More details are available below
## How pipelines run
Somewhat unsurprisingly, the function used to run pipelines is called `run_pipeline`.
Cognee uses a **layered execution model**: a single call to `run_pipeline` orchestrates **multi-dataset processing** by running **per-file pipelines** through the sequence of tasks.
* **Statuses** are yielded as the pipeline runs and written to **databases** where appropriate
* **User access** to datasets and files is carefully verified at each layer
* **Pipeline run information** includes dataset IDs, completion status, and error handling
* **Background execution** uses queues to manage status updates and avoid database conflicts
Every `run_pipeline` call takes a `pipeline_name` parameter (default: `"custom_pipeline"`) and a `use_pipeline_cache` flag (default: `False`). These two values together control whether a pipeline re-processes a dataset that was already handled.
### Reserved pipeline names
Two pipeline names are used internally and carry special meaning:
| Name | Used by | Behavior |
| ------------------ | ------------------ | ----------------------------------------------------------------------------------------------------------------------- |
| `cognify_pipeline` | `cognee.cognify()` | Runs with `use_pipeline_cache=False`; starts a new dataset-level run instead of skipping because of a prior dataset run |
| `add_pipeline` | `cognee.add()` | Runs with `use_pipeline_cache=False`; starts a new dataset-level run instead of skipping because of a prior dataset run |
Both built-in operations run with `use_pipeline_cache=False`, so they do **not** short-circuit based on a dataset-level `DATASET_PROCESSING_COMPLETED` or `DATASET_PROCESSING_STARTED` record. They start a new dataset-level run, while per-document pipeline status can still skip data items that already completed for that pipeline. Concurrent runs on the **same** dataset are kept safe by a per-dataset lock (see the "Per-dataset serialization" section below) rather than by the cache check.
The lower-level `cognee.add()` step, which is also used inside `remember()`, always resets the stored status for **both** `add_pipeline` and `cognify_pipeline` before running, so that new data can be re-processed by the downstream `cognify()` step on the next call.
Do not use `cognify_pipeline` or `add_pipeline` as `pipeline_name` values in your own `run_pipeline` calls. Reusing these names causes your pipeline to read and write the same status records as the built-in operations, which can lead to unexpected skipping or incorrect state resets.
### How `use_pipeline_cache` works
When `use_pipeline_cache=True`, Cognee checks the relational database for the most recent run of `pipeline_name` on the target dataset before executing:
* If the stored status is **`DATASET_PROCESSING_COMPLETED`** → the pipeline yields the cached result and returns immediately without re-running the tasks.
* If the stored status is **`DATASET_PROCESSING_STARTED`** → the pipeline yields the in-progress status and returns, preventing duplicate concurrent runs.
* If there is **no prior record** (new dataset or new pipeline name) → the pipeline runs normally.
When `use_pipeline_cache=False` (the default for custom pipelines, and the mode used by `cognee.add()` and `cognee.cognify()`), the dataset-level qualification check is skipped entirely — the prior dataset run status is not read — and the pipeline starts a new dataset-level run regardless of any prior dataset completion status. Per-document pipeline status is checked later during task execution, so individual data items that already completed for that pipeline can still be skipped. Safety against concurrent runs on the same dataset is provided by the per-dataset lock described below instead of by this check.
Pipeline runs are serialized **per dataset**. Before a run starts, `run_pipeline_per_dataset` acquires a lock keyed on the dataset ID, so two runs that target the **same** dataset execute one after another — the second waits until the first finishes — while runs on **different** datasets still proceed in parallel. This protects each dataset from concurrent writers (for example, two `cognify()` calls on the same dataset) without globally serializing all pipeline activity.
Delete operations share the **same** per-dataset lock. Deleting a dataset or a single data item (including the memory-clearing `forget()` paths) acquires the lock keyed on that dataset ID before it mutates anything, so a delete waits for an in-flight `add()`, `cognify()`, or `memify()` run on the same dataset to finish — and a pipeline run started while a delete is in progress waits for the delete. Two deletes targeting the same dataset serialize the same way. Deletes on **different** datasets still proceed in parallel.
The lock is **process-local** — it is an in-memory `asyncio.Lock`. It only serializes runs and deletes within a single process/event loop and does **not** guard against multiple processes or workers operating on the same dataset at once. Cognee is designed to run as a single process; do not point multiple processes or workers at the same stores.
### Nested (re-entrant) runs
A pipeline task may legitimately start another pipeline on the same dataset — for example, a session-driven run calling `add()` or `cognify()` on the dataset it is already processing. Because the per-dataset lock is not re-entrant, re-acquiring it from within the same execution would self-deadlock. Cognee detects that the current execution already holds the dataset's lock and lets the nested run proceed **without re-locking**; external runs (and deletes) on that dataset stay queued behind the lock the ancestor run holds. The same re-entrancy applies to deletes, since they acquire the lock from the same registry.
For your own pipelines, choose a unique `pipeline_name` that does not conflict with `cognify_pipeline` or `add_pipeline`. Using a unique name means:
* State tracking is isolated to your pipeline — a completed run of the built-in `cognify()` will not affect your pipeline's qualification check.
* If you enable `use_pipeline_cache=True` for your custom pipeline, you must reset its status manually (via `reset_dataset_pipeline_run_status`) when you want to re-process a dataset.
```python theme={null}
# Custom pipeline with a unique name — safe to use alongside the built-in memory workflows
async for run_info in run_pipeline(
tasks=tasks,
data=text,
datasets=["my_dataset"],
pipeline_name="my_enrichment_pipeline", # unique name, no conflict
use_pipeline_cache=False, # default: always re-runs
):
pass
```
How an interrupted run ends depends on **how** it was interrupted:
* A **hard kill** — `SIGKILL`, a container OOM, power loss — stops the process before any cleanup code can run, so the pipeline run record in the relational database is left with a `DATASET_PROCESSING_STARTED` status.
* A **cancellation** — a graceful server shutdown or restart, or any other `asyncio.CancelledError` delivered to the task driving the run — is cooperative, so the run's terminal handler gets to run. The run is finalized as `DATASET_PROCESSING_ERRORED` rather than being left in progress, and the cancellation is then re-raised so it still propagates to the caller exactly as before.
Marking cancelled runs as errored is new. Previously the terminal handler caught only `Exception`, and `asyncio.CancelledError` is a `BaseException`, so a cancelled run never reached the error-logging step and its record stayed at `DATASET_PROCESSING_STARTED` indefinitely.
On the cancellation path the run is now finalized like any other failed run: the rollback handler runs first, then the terminal `pipeline_runs` row is written with `outcome` `FAILED`, `error_class` `CancelledError`, and a scrubbed `error_message`, and a `PipelineRunErrored` event is yielded before the `CancelledError` is re-raised. Behavior for ordinary exceptions is unchanged.
Because `cognify()` and `add()` run with `use_pipeline_cache=False`, they do not consult the dataset-level status at all — the next call starts a new dataset-level run, serialized by the per-dataset lock. A stuck `DATASET_PROCESSING_STARTED` record therefore does not block the built-in operations either way, although completed data items can still be skipped by their per-document pipeline status.
| How the run was interrupted | Pipeline run status left behind | What `cognify()` does on retry | Outcome |
| ----------------------------------------------------------- | ------------------------------- | ------------------------------------------------------------------------------ | -------------------------------------------------------- |
| Hard kill (process killed, OOM, power loss) | `DATASET_PROCESSING_STARTED` | Dataset-level cache check skipped; starts a new run under the per-dataset lock | Runs normally; completed data items may still be skipped |
| Cancelled (graceful shutdown or restart, task cancellation) | `DATASET_PROCESSING_ERRORED` | Dataset-level cache check skipped; starts a new run under the per-dataset lock | Runs normally; completed data items may still be skipped |
| Raised an error | `DATASET_PROCESSING_ERRORED` | Dataset-level cache check skipped; starts a new run under the per-dataset lock | Runs normally; completed data items may still be skipped |
### Automatic recovery
Two automatic layers clean up failed or abandoned cognify runs, so a re-run can resume instead of redoing or half-skipping work:
* **Rollback on error** — when a cognify run fails (including when it is cancelled), an error handler rolls back the partial graph, vector, and relational artifacts written by that run and clears the per-document `cognify_pipeline` status of the data items the failed run touched. Data items completed in earlier successful runs keep their status and are still skipped, so the next run resumes at the first unprocessed document and reprocesses the rolled-back ones cleanly.
* **Startup recovery for stale runs** — when the API server starts, it finds datasets whose latest cognify run is still `DATASET_PROCESSING_STARTED`, rolls back those older than `COGNEE_STALE_RUN_RECOVERY_MIN_AGE_SECONDS` (env var, default `3600` seconds), and resets their status to `DATASET_PROCESSING_INITIATED` so the dataset is no longer reported as "already being processed" and can be cognified again. Younger runs are left alone because they may still be executing in another live worker or replica. This runs only during API server startup — library-only usage does not trigger it. Because it keys on `DATASET_PROCESSING_STARTED`, it covers hard kills; a cancelled run already reaches the terminal `DATASET_PROCESSING_ERRORED` status on its own, so there is nothing for the reaper to pick up and no wait for the minimum-age threshold.
By default a per-document processing error aborts the whole run; set the `RAISE_INCREMENTAL_LOADING_ERRORS` env var to `false` to log the error and continue with the remaining data items instead.
### Manual reset
The `reset_dataset_pipeline_run_status` helper below is still useful for **custom pipelines that opt into `use_pipeline_cache=True`**, where a stuck `DATASET_PROCESSING_STARTED` record would otherwise cause the cache check to report the dataset as "already being processed" and skip a re-run. That now applies to hard kills only: the cache check lets a `DATASET_PROCESSING_ERRORED` record through, so a run ended by cancellation re-runs on its own without a manual reset.
To unblock a stuck pipeline that uses `use_pipeline_cache=True`, call `reset_dataset_pipeline_run_status`. It writes a new `DATASET_PROCESSING_INITIATED` record, which clears the stuck status so the next cached run is no longer skipped.
```python theme={null}
from uuid import UUID
from cognee.modules.pipelines.layers.reset_dataset_pipeline_run_status import (
reset_dataset_pipeline_run_status,
)
# Reset all pipelines on a dataset
await reset_dataset_pipeline_run_status(dataset_id=my_dataset.id, user=current_user)
# Or reset only specific pipelines by name
await reset_dataset_pipeline_run_status(
dataset_id=my_dataset.id,
user=current_user,
pipeline_names=["cognify_pipeline"],
)
```
**Parameters**
| Parameter | Type | Required | Description |
| ---------------- | ------------------- | -------- | ---------------------------------------------------------------------------- |
| `dataset_id` | `UUID` | Yes | The ID of the dataset whose pipeline runs should be reset |
| `user` | `User` | Yes | The user object used for ownership lookup |
| `pipeline_names` | `list[str] \| None` | No | If provided, only runs for these pipeline names are reset; omit to reset all |
Once reset, calling `cognify()` again is safe:
* Documents that **fully completed** before the crash (their per-document `pipeline_status` entry is `DATA_ITEM_PROCESSING_COMPLETED`) are skipped — no duplicate graph nodes or embeddings are written.
* Documents that were **mid-processing** when the crash occurred will be reprocessed from the beginning. These items will be re-chunked, re-extracted, and re-embedded.
`reset_dataset_pipeline_run_status` resets the dataset-level run status only. It does not clear per-document status. Documents that completed before the crash remain marked as completed and are not reprocessed.
Pipeline runs are persisted in the relational `pipeline_runs` table. Alongside the status, IDs, and pipeline name, the record keeps a `run_info` column with an **audit-only preview** of the input the run was started with. This preview is bounded so a single run cannot grow the table without limit:
* If the input is a list of Cognee `Data` records, only their IDs are stored.
* Any other input is stringified and, if longer than **512 characters**, truncated to a preview that ends with `... [truncated, chars total]`.
* Empty or missing input is recorded as `"None"`.
This preview is intended for inspection and debugging only — Cognee never reads it back during processing. If you need the full input payload (for example, large raw text passed to `add()` or `cognify()`), persist it yourself in object storage or a linked record rather than relying on `run_info` to retain it verbatim.
`run_info` also holds a second key, `progress`, which *is* read back — by `GET /api/v1/datasets/status/progress`. It is an in-flight snapshot of `{completed_items, total_items, current_stage}` written as items finish, and it is the one exception to the append-only rule below: a progress tick **updates the run's existing `DATASET_PROCESSING_STARTED` row in place** rather than inserting. Inserting per tick would grow `pipeline_runs` without bound as batches accumulate, so progress is stored as metadata inside the started state — `PipelineRunStatus` gains no new member for it. Three consequences worth knowing:
* Ticks are throttled to roughly 20 database writes per run (`max(1, total_items // 20)`, with the first and last item always persisted), so the snapshot is coarse by design — that keeps write pressure off the default SQLite backend, which serializes writers behind a file lock.
* Concurrent ticks for the same run race on a read-modify-write with no locking: last write wins. For a display-only signal that only risks a slightly stale snapshot between ticks.
* A late tick can never make a finished run read as running again. Status readers pick the newest row, so a tick that lands after the terminal row merely updates the older `STARTED` row in place, invisibly; and the defensive path for a `STARTED` row that has gone missing drops the tick outright rather than inserting a new `STARTED` row with a later `created_at` than the terminal one.
**The table is otherwise not one row per pipeline run.** It is append-only, and it holds two kinds of row:
* A pipeline run writes **several** rows that share one `pipeline_run_id` (initiated → started → terminal). Only the terminal row carries the run's `outcome` and `tokens_in` / `tokens_out`.
* Non-pipeline operations (`search`, `recall`, `remember`, `forget`, `delete`, `prune`) write exactly **one** row each, with `status`, `pipeline_name` and `pipeline_id` left `NULL` — which is what keeps them invisible to readers that look up a dataset's latest pipeline status.
Two rules follow when you aggregate over the table (for example over [`GET /api/v1/activity/pipeline-runs`](/guides/deploy-rest-api-server)):
1. Deduplicate by `pipeline_run_id` before summing, or a single run is counted once per row it wrote.
2. `parent_operation_id` links a child operation to its parent's `pipeline_run_id`, forming a tree — but token counts already chain up into the parent. Sum one level only; summing across levels double-counts.
Because the table now grows with every operation rather than with every pipeline run, it carries a composite `(created_at, id)` index matching the newest-first, id-tiebroken order that the activity feed and the time-bucketed usage queries read it in. Existing databases pick it up from the usual Alembic upgrade, so there is no manual step unless you run with `ENABLE_AUTO_MIGRATIONS=false` — then apply it with `cognee-cli upgrade` as you would any other revision.
`pipeline_runs` is no longer pipeline-only: it is Cognee's **local activity record** — an audit table of every operation run on the deployment, kept in the deployment's own relational database (SQLite or Postgres, whichever is configured). The rows never leave your instance, are readable by whoever can query that database, and are unrelated to Cognee's anonymous product telemetry.
Operations that never run a pipeline — `search()`, `recall()`, `remember()`, `improve()`, `forget()`, `datasets.delete_data()`, and both prune paths — each write **exactly one** self-contained row when they finish, so a single indexed SQL query answers "what did this user run, did it work, how long did it take, and what did it cost in tokens" without parsing JSON.
These operation rows are written with `status`, `pipeline_name`, and `pipeline_id` left `NULL`. That is deliberate: every status reader filters on `pipeline_name`/`status`, so operation rows are invisible to the caching, crash-recovery, and status-reporting logic described above. Each row still gets its own `pipeline_run_id` — the operation's id, which children reference as `parent_operation_id`. The append-only pattern and the `PipelineRunStatus` enum are unchanged.
### Operation-record columns
All of these columns are nullable and are populated on both operation rows and the terminal (completed/errored) row of a pipeline run:
| Column | Meaning |
| ------------------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
| `user_id`, `tenant_id` | The identity that triggered the operation |
| `operation_name` | The operation (`"search"`, `"recall"`, `"remember"`, `"improve"`, `"forget"`, `"delete"`, `"prune_data"`, `"prune_system"`) — or the pipeline name on pipeline rows |
| `started_at`, `ended_at` | Timezone-aware start and finish timestamps; subtract for duration |
| `outcome` | `"succeeded"` or `"failed"` (plain string, from the `OperationOutcome` enum) |
| `error_class` | Exception class name on failure, e.g. `"DatasetNotFoundError"` |
| `error_message` | Failure message, PII-scrubbed (emails, secrets, home-directory user names, long digit runs) and truncated to 512 characters |
| `tokens_in`, `tokens_out` | LLM tokens spent inside the operation. `NULL` means not measured; `0` means measured as zero |
| `origin` | The surface that initiated the work: `"sdk"` (default), `"api"`, `"cli"`, `"mcp"`, or `"background"` |
| `session_id` | The session-cache id, when the operation ran in a session |
| `parent_operation_id` | The enclosing run's `pipeline_run_id`, making nesting such as `remember` → `add`/`cognify`/`improve` a queryable tree |
| `background` | `True` when the call only *launched* background work — `outcome="succeeded"` then means "accepted and started", not "the background work finished" |
Token counts **chain to parents**: a parent row's `tokens_in`/`tokens_out` already include everything its children spent. Filter by `parent_operation_id IS NULL`, or read one level at a time — never `SUM` token columns across nesting levels, or you double-count.
### Recording is fail-open
The recorder can never break the operation it records. The wrapped operation's exceptions always propagate unchanged, and a failure to persist the row is logged and swallowed. One known consequence: `prune_system(metadata=True)` drops the relational database including `pipeline_runs`, so its own record is self-erasing by design.
### Cost and accuracy
Recording is a single `INSERT` off the hot path — roughly **2.5 ms per operation**, with no additional LLM or network calls. The recorded token counts are the provider-billed `response.usage` figures when the provider reports usage (falling back to a character-based estimate when it does not), so they include hidden reasoning tokens; usage you read out of this table may therefore be higher than counts derived from visible output alone. Nothing about billing changes — only the accuracy of what Cognee reports.
### Upgrading
The columns are added by a single idempotent Alembic migration (`a7f3c9e1b5d2`). Old rows keep `NULL` in the new columns — there is no backfill, and existing readers are unaffected. The API server applies migrations at startup; for standalone scripts, CI, or self-managed database lifecycles, run [`run_migrations`](/python-api/run-migrations) after upgrading the `cognee` package.
`PipelineContext` is the runtime context object that Cognee automatically builds and injects into any task that declares a `ctx` parameter. It carries the user, dataset, and per-item context for the current pipeline run, and provides an `extras` dict for custom state.
| Field | Type | Description |
| --------------- | ---------------- | ------------------------------------------------------------------------------------------------ |
| `user` | `Any` | The user that triggered the pipeline. Used for access control and provenance. |
| `dataset` | `Any` | The resolved dataset object for the current run. |
| `data_item` | `Any` | The individual data item being processed in this pipeline execution. |
| `pipeline_name` | `Optional[str]` | The name passed to `run_pipeline` or `run_tasks`. |
| `extras` | `Dict[str, Any]` | Arbitrary key/value state you can pass into the pipeline and read in any task. Defaults to `{}`. |
The framework inspects each task function's signature. If it finds a parameter named `ctx`, it passes the current `PipelineContext` when the task runs. Matching is by parameter name, not by type annotation.
Tasks that do not declare `ctx` simply receive no context and are unaffected.
Pass a dict as the `context` argument to `run_pipeline` or `extras` to `run_tasks`. Every task in the pipeline can read those values from `ctx.extras`.
```python theme={null}
from cognee.modules.pipelines import Task, run_pipeline
from cognee.modules.pipelines.models.PipelineContext import PipelineContext
async def score_items(data, ctx: PipelineContext = None):
multiplier = ctx.extras.get("score_multiplier", 1) if ctx else 1
return [item * multiplier for item in data]
async for _ in run_pipeline(
tasks=[Task(score_items)],
data=[1, 2, 3],
datasets=["my_dataset"],
pipeline_name="scoring_pipeline",
context={"score_multiplier": 10},
):
pass
```
`ctx.extras` is always a plain dict, never `None`.
The `user` and `dataset` fields are most useful when you need to write provenance records or apply per-tenant logic:
```python theme={null}
async def store_result(data_points, ctx: PipelineContext = None):
user = ctx.user if ctx else None
dataset = ctx.dataset if ctx else None
data_item = ctx.data_item if ctx else None
for dp in data_points:
await write_to_store(dp, user_id=user.id, dataset_id=dataset.id)
return data_points
```
The built-in `add_data_points` task already does this automatically, so you typically only need to read these fields when writing your own storage tasks.
The built-in `ingest_data` task reads `ctx.dataset` too, but conditionally, and the condition matters if you compose it into your own pipeline. `run_pipeline` resolves and write-checks the run's dataset once, so `ingest_data` reuses `ctx.dataset` instead of re-resolving it per item — but only when that dataset demonstrably is the one the call targets: matching the `dataset_id` argument, or matching the `dataset_name` argument for a dataset the calling user owns in the same tenant. Point `ingest_data` at any other dataset and it ignores `ctx.dataset`, resolving the dataset itself and running the usual write-permission check, so the reuse is a saved lookup and never a widened permission.
Always default `ctx` to `None` so the task can also be called directly in tests or scripts without a running pipeline:
```python theme={null}
async def my_task(data, ctx: PipelineContext = None):
name = ctx.pipeline_name if ctx else "standalone"
...
```
When a task raises while processing a data item, the pipeline logs the error, yields a `PipelineRunErrored` status, and then re-raises the original exception to the caller. A failing data item therefore both surfaces a `PipelineRunErrored` event **and** propagates the underlying exception out of the pipeline run, rather than being silently collapsed into an error status only.
Because of this, wrap pipeline runs in `try`/`except` when you iterate them, so you can react to the propagated exception:
```python theme={null}
try:
async for run_info in run_pipeline(
tasks=tasks,
data=text,
datasets=["my_dataset"],
pipeline_name="my_pipeline",
):
...
except Exception as error:
# the original task exception propagates here after the
# PipelineRunErrored status has been yielded
handle(error)
```
**What a propagated Cognee exception looks like.** Cognee's own exception types derive from `CogneeApiError`, which carries three attributes you can read in the handler: `message`, `name`, and `status_code`. Its `__str__` formats them as `"{name}: {message} (Status code: {status_code})"` — for example, passing a list containing something other than `Task` instances to a custom pipeline raises `"WrongTaskTypeError: tasks argument must be a list of Task class instances, got str in the list. (Status code: 400)"`. The base constructor also populates `Exception.args` with `(message, name)`, so `repr()` and `raise ... from error` chaining behave the way they do for any standard Python exception:
```python theme={null}
from cognee.exceptions import CogneeApiError
try:
...
except CogneeApiError as error:
logger.warning("%s failed with %s", error.name, error.status_code)
raise MyAppError(error.message) from error # error stays reachable as __cause__
```
Not every propagated error is a `CogneeApiError`, though — plain built-in exceptions and errors from underlying database or LLM libraries surface too, so keep the broad `except Exception` above as the safety net.
`CogneeApiError.__init__` also logs the exception centrally, at `ERROR` by default. Some exceptions that represent expected control flow rather than failures — such as an adapter reporting an unsupported capability — opt out with `log=False`, so a missing `ERROR` log line for one of those does not mean the exception was swallowed. It still propagates to your `except` block exactly as above.
* Innermost layer: individual task execution with telemetry and recursive task running in batches
* Middle layer: per-dataset pipeline management and task orchestration
* Outermost layer: multi-dataset orchestration and overall pipeline execution
* Execution modes: blocking (wait for completion) or background (return immediately with "started" status)
* In background mode with no `datasets` passed, the run resolves to every dataset the run's user has write access to — the user supplied in the run's params, or the default user when none is given
* Use [Remember](../main-operations/remember) for the default ingestion path
* Modify transformation steps without touching low-level functions, avoid going below `run_pipeline`
* Custom tasks let you extend or replace default behavior
* Identity: represents who owns and acts on data. If omitted, a default user is used
* Ownership: every ingested item is tied to a user; content is deduplicated per owner
* Permissions: enforced per dataset (read/write/delete/share) during processing and API access
* Container: a named or UUID-scoped collection of related data and derived knowledge
* Scoping: `remember()` writes into a specific dataset, and dataset-scoped pipelines process the dataset(s) you pass
* Lifecycle: new names create datasets and grant the calling user permissions; UUIDs let you target existing datasets (given permission)
Learn about the individual processing units that make up pipelines
Understand the structured outputs that pipelines produce
See how pipelines are used in Remember and lower-level ingestion workflows