What it is
CocoIndex is an open-source incremental data transformation engine for AI workloads, built in Rust with a Python interface and positioned in the data engineering, ETL, indexing, and AI agent ecosystem. It turns codebases, meeting notes, inboxes, Slack, PDFs, and videos into continuously fresh context for AI agents and LLM applications.
The concrete problem it addresses is stale context in agent systems. Batch pipelines can leave an application using outdated documents, embeddings, or records, while full reprocessing can be expensive and slow. CocoIndex keeps target data in sync by recomputing only the changed portion, so developers can maintain fresh retrieval context without rebuilding the entire dataset on every run.
Key capabilities
- CocoIndex updates only the delta when source data changes, so a pipeline does not need a full reprocess on every run.
- It uses a declarative Python API where a function declares the rows that should exist in a target, and the engine keeps that target synchronized with mounted source items.
- It supports local filesystem traversal and Postgres target mounting, including declared vector indexes for embedding columns.
- It can cache computation by hash of input and code, so unchanged work is not repeated.
- It is described as parallel by default and intended for any scale, making it suitable for jobs that process many source items.
- It frames data transformation with lineage, so users can relate processed context to source data.
Who uses it and how
- AI application teams use it to build retrieval context from local documentation and store chunks plus embeddings in Postgres.
- Data engineers use it as a Python ETL and indexing layer for refreshed datasets that feed AI agents and LLM applications.
- Agent builders use it to reduce context gaps by re-running pipelines that update only changed files, notes, inboxes, Slack, PDFs, or videos.
- Developers using AI coding agents can install the CocoIndex skill so the agent generates correct v1 pipeline code.
Getting started
The README shows installation with pip install -U cocoindex for Python 3.10 through 3.13, followed by a Python script that defines a pipeline and calls update_blocking. The typical workflow is to run the pipeline once to backfill the target and then re-run it later so only changed source items are processed.
When to use it — and when not to
CocoIndex is a good fit when you want an incremental, declarative Python pipeline for AI context indexing and you are willing to operate the required target database, such as Postgres, yourself. It is less suitable if you need a mature platform with a long public track record, because the repository is listed as 0 years old, 0 contributors, and 83 open issues. It does not list paid products it replaces, so buyers should compare it directly against existing batch ETL, vector database, and agent context workflows.
project readme (upstream, from github) — read inline

Your agents deserve fresh context.
Star us ❤️ →
·
·
·
CocoIndex turns codebases, meeting notes, inboxes, Slack, PDFs, and videos into live, continuously fresh context for your AI agents and LLM apps to reason over effectively — with minimal incremental processing. Get your production AI agent ready in 10 minutes with reliable, continuously fresh data — no stale batches, no context gap
Incremental · only the delta · Any scale · parallel by default · Declarative · Python, 5 min



Deutsch |
English |
Español |
français |
日本語 |
한국어 |
Português |
Русский |
中文
Built with CocoIndex ❤️
See all 20+ examples · updated every week →
Get started
pip install -U cocoindex
Declare what should be in your target — CocoIndex keeps it in sync forever, recomputing only the Δ.
import cocoindex as coco
from cocoindex.connectors import localfs, postgres
from cocoindex.ops.text import RecursiveSplitter
@coco.fn(memo=True) # ← cached by hash(input) + hash(code)
async def index_file(file, table):
for chunk in RecursiveSplitter().split(await file.read_text()):
table.declare_row(text=chunk.text, embedding=embed(chunk.text))
@coco.fn
async def main(src):
table = await postgres.mount_table_target(PG, table_name="docs")
table.declare_vector_index(column="embedding")
await coco.mount_each(index_file, localfs.walk_dir(src).items(), table)
coco.App(c