CocoIndex empowers long-horizon AI agents with continuously fresh context through intelligent incremental processing that recomputes only data deltas, eliminating staleness and computational waste while scaling from single repositories to petabyte-scale deployments.
Long-horizon AI agents operate over extended timeframes, continuously reasoning over evolving knowledge bases such as code repositories, technical documentation, meeting transcripts, Slack threads, and PDF corpora. Traditional batch-oriented ETL or RAG pipelines create fundamental problems: full re-indexing on every change wastes compute, context drifts out of date between runs, and state management becomes brittle as sources mutate independently. An incremental engine addresses these issues by treating data synchronization as a stateful, delta-driven computation rather than periodic full rebuilds.
CocoIndex implements this paradigm as a declarative Python framework backed by a high-performance Rust core. It models the desired target state as a pure function of the current source state, then automatically reconciles differences using change data capture, memoization keyed on both input values and code hashes, and targeted propagation of updates across joins and lookups. Only affected records are reprocessed; unchanged data remains untouched. This design delivers sub-second freshness at any scale while providing failure isolation so a single malformed record never stalls the entire pipeline.
The framework supports diverse connectors out of the box, including local filesystem traversal with pattern matching, PostgreSQL, S3-compatible object storage, and emerging vector targets such as Turbopuffer. Built-in operations leverage tree-sitter for language-aware code chunking and integrate seamlessly with embedding models. For AI developers building production agentic systems, CocoIndex removes the data synchronization bottleneck that currently forces teams to choose between freshness and cost. The official repository at https://github.com/cocoindex-io/cocoindex contains the complete source, 20+ production-ready examples updated weekly, and a dedicated skill pack that enables AI coding agents to generate correct CocoIndex pipelines.
The architecture centers on a declarative App that orchestrates sources, processing functions, and targets. Memoized functions (@coco.fn(memo=True)) cache results by content hash, while mount_each distributes work across parallel workers. When a source file changes, only the impacted chunks are re-embedded and the vector index is updated transactionally. Stale rows are retired automatically, ensuring the agent always queries a consistent, up-to-date view.
Installation Guide
CocoIndex installs via standard Python package managers and runs on macOS 10.12+, Linux (glibc 2.28+), and Windows 10+ with Python 3.11–3.13.
Begin by creating an isolated project environment:
mkdir cocoindex-agent-project && cd cocoindex-agent-project
python -m venv .venv
source .venv/bin/activate # or `.venv\Scripts\activate` on WindowsInstall the core package:
pip install -U cocoindexFor PDF and document-heavy workloads, also install the Docling converter:
pip install -U cocoindex doclingThe CLI cocoindex becomes available immediately after installation. For state persistence in file-based examples, create a .env file:
echo "COCOINDEX_DB=./cocoindex.db" > .envWhen targeting PostgreSQL (recommended for production vector search), ensure the instance includes the pgvector extension and expose connection parameters via environment variables (PGHOST, PGPORT, PGUSER, PGPASSWORD, PGDATABASE). Turbopuffer targets require only an API key.
Verify the installation:
cocoindex --versionThe command should report the current release (v1.0.3 or newer). No Rust toolchain is required for end users; the high-performance engine is pre-compiled and distributed as a binary wheel.
Usage and Implementation
CocoIndex follows a clear declarative workflow: define processing logic as memoized functions, declare the desired target state, then instantiate an App that keeps the target synchronized with the source. The following example indexes a local documentation directory into a PostgreSQL table with vector embeddings, exactly the pattern used by long-horizon coding agents.
Create index_docs.py:
import cocoindex as coco
from cocoindex.connectors import localfs, postgres
from cocoindex.ops.text import RecursiveSplitter
PG = {
"host": "localhost",
"port": 5432,
"user": "postgres",
"password": "your_password",
"database": "agent_context",
}
@coco.fn(memo=True)
async def index_file(file, table):
text = await file.read_text()
for chunk in RecursiveSplitter().split(text):
table.declare_row(
text=chunk.text,
embedding=embed(chunk.text), # replace with your embedding call
source_path=str(file.path),
last_modified=file.mtime,
)
@coco.fn
async def main(src: str):
table = await postgres.mount_table_target(PG, table_name="docs")
table.declare_vector_index(column="embedding", metric="cosine")
await coco.mount_each(
index_file,
localfs.walk_dir(src, recursive=True).items(),
table
)
app = coco.App(
coco.AppConfig(name="docs-indexer"),
main,
src="./docs",
)
app.update_blocking()Run the initial backfill:
cocoindex update index_docs.pySubsequent executions process only changed or new files. Deleted source files automatically trigger row retirement in the target table. The declare_vector_index call ensures a cosine-similarity HNSW index is maintained for fast agent retrieval.
For a lighter file-transformation use case (e.g., maintaining a live Markdown knowledge base), the PDF-to-Markdown quickstart demonstrates the same incremental guarantees with local filesystem targets:
import pathlib
import cocoindex as coco
from cocoindex.connectors import localfs
from cocoindex.resources.file import PatternFilePathMatcher
from docling.document_converter import DocumentConverter
_converter = DocumentConverter()
@coco.fn(memo=True)
def process_file(file: localfs.File, outdir: pathlib.Path) -> None:
markdown = _converter.convert(file.file_path.resolve()).document.export_to_markdown()
outname = file.file_path.path.stem + ".md"
localfs.declare_file(outdir / outname, markdown, create_parent_dirs=True)
@coco.fn
async def app_main(sourcedir: pathlib.Path, outdir: pathlib.Path) -> None:
files = localfs.walk_dir(
sourcedir,
recursive=True,
path_matcher=PatternFilePathMatcher(included_patterns=["**/*.pdf"]),
)
await coco.mount_each(process_file, files.items(), outdir)
app = coco.App("PdfToMarkdown", app_main, sourcedir=pathlib.Path("./pdf_files"), outdir=pathlib.Path("./out"))
app.update_blocking()Long-horizon agents interact with CocoIndex indirectly through the target store. After indexing, an agent issues semantic queries against the vector table (or reads the latest Markdown files) to retrieve fresh context. Because updates are incremental and transactional, the agent never observes partial or stale snapshots. In production, schedule cocoindex update via cron, Git hooks, or a lightweight watcher process. For event-driven flows, combine with webhooks on source systems to trigger targeted updates.
The second diagram illustrates the full dataflow: source change detection feeds an incremental indexer that updates only affected embeddings, while the search pipeline continues to serve low-latency queries against the live vector store. This separation of indexing and serving planes is essential for agents that must reason continuously without blocking on reprocessing.
Advanced patterns include composing multiple mount_each calls for heterogeneous sources, using @coco.fn(memo=True) on expensive embedding or LLM calls, and leveraging the built-in tree-sitter splitters for language-specific chunking in coding agents. The official skill at https://github.com/cocoindex-io/cocoindex enables AI coding assistants to generate correct pipelines directly from natural language requirements.
Production Considerations and Best Practices
CocoIndex’s Rust engine provides parallel chunking, zero-copy transforms where possible, and strict failure isolation. Memory usage remains bounded even on petabyte-scale corpora because only deltas are materialized. For enterprise deployments, the same declarative code runs unchanged from a developer laptop to a Kubernetes job; simply point the source connector at S3 or a database replica and the target at a managed vector service.
Monitor progress with the redesigned PTY-based reporter and integrate health checks via the App’s programmatic API. When logic changes (for example, switching embedding models), the memoization layer automatically invalidates only the affected records, avoiding costly full backfills.
By adopting CocoIndex, teams building long-horizon agents replace fragile nightly batch jobs with a live, explainable data substrate. The result is agents that reason over genuinely current information at a fraction of the previous compute cost. Clone the repository at https://github.com/cocoindex-io/cocoindex, explore the examples directory, and deploy your first incremental pipeline in under ten minutes. The future of reliable agent memory is incremental.
