Stop Re-embedding Everything! CocoIndex Only Processes the Δ

Your AI agent just answered a user's question with context from last Tuesday's meeting notes. The problem? Those notes changed twice yesterday, your competitor's pricing page updated this morning, and that critical bug fix in auth.rs landed three hours ago. Your agent is confidently hallucinating with stale data — and you're paying full price for the privilege.

Here's the brutal truth that keeps ML engineers awake: batch pipelines are broken for long-horizon agents. Every night, you re-embed 10,000 documents to catch the 47 that actually changed. Every morning, your vector database is six hours stale. Every quarter, your cloud bill makes finance cry.

What if you could declare what your target should contain — and have an engine keep it synchronized forever, recomputing only the delta? No DAGs to manage. No cron jobs to debug. No "oops, we forgot to re-index" incidents at 2 AM.

Enter CocoIndex: the incremental data transformation framework that's making batch pipelines obsolete. Built with a Rust core for production-grade reliability, CocoIndex brings React^{↗ Bright Coding Blog}'s declarative magic to data engineering. Your code stays as simple as the one-off version. The engine handles the rest — caching, lineage, retries, and that beautiful Δ-only recomputation that slashes your embedding bill by 90%.

What is CocoIndex?

CocoIndex is an incremental engine for long-horizon AI agents — an open-source Python^{↗ Bright Coding Blog} framework with a Rust core that transforms scattered enterprise data into live, continuously fresh context. Think codebases, Slack threads, meeting transcripts, PDFs, videos, and databases flowing through intelligent pipelines into vector stores, knowledge graphs, and feature stores that your agents can actually trust.

The project emerged from a simple observation: modern AI agents need to reason over dynamic, multi-source corpora, but existing ETL tools force engineers to choose between freshness and cost. Streaming pipelines are complex. Batch pipelines are stale. And hand-rolled incremental logic? That's a maintenance nightmare waiting to happen.

CocoIndex's creators took a different approach. They built what they call "React — for data engineering" — a persistent-state-driven dataflow where you declare Target = F(Source) and the engine maintains that invariant continuously. When source data changes, only affected records propagate. When your transformation code changes, only outputs dependent on the modified logic re-execute. The result? Sub-second freshness at a fraction of the compute cost.

The framework has gained serious traction in the AI infrastructure community, with growing GitHub stars, active Discord discussions, and a flagship product — CocoIndex-code — that serves as an MCP server giving Claude Code and Cursor instant, semantic understanding of entire repositories. The Rust core provides memory safety, zero-copy transforms where possible, and failure isolation that keeps one bad record from stalling your entire pipeline.

Key Features That Make CocoIndex Insane

Δ-Only Incremental Processing The headline feature: CocoIndex tracks per-row provenance so precisely that when a single file changes, only that file's embeddings recompute. The engine maintains fine-grained lineage from source bytes to target vectors, enabling surgical invalidation rather than carpet-bombing your entire corpus. For a 10,000-document collection with 0.1% daily churn, that's 99.9% compute savings.

Declarative Python API, Rust Core You write clean Python — decorators, async functions, familiar patterns. The heavy lifting happens in Rust: parallel chunking, efficient caching, robust scheduling. This split gives you ergonomics without sacrificing the performance and reliability that production AI systems demand. Python 3.10 through 3.13 supported, with PyPI distribution for trivial installation.

End-to-End Lineage Every target row traces back to its exact source bytes. Debug a bad embedding? Follow the lineage thread to the specific file, line, and transformation version. This isn't just nice-to-have — it's essential for auditable AI systems, regulatory compliance, and those 3 AM debugging sessions where you need answers fast.

Memoized Transformations The @coco.fn(memo=True) decorator caches results by hash of inputs plus hash of code. Change your embedding model? Only affected records recompute. Roll back a buggy transformation? Previous cached results restore instantly. It's like Git for your data pipeline — immutable, reproducible, branchable.

Parallel by Default No manual multiprocessing or asyncio gymnastics. CocoIndex's task scheduler distributes work across available cores automatically. Walk a directory tree, process files, chunk text, generate embeddings — all happening concurrently with dependency-aware execution order.

Production-Grade Resilience Rust doesn't just mean speed — it means reliability. Exponential backoff, dead-letter queues for poison records, retry loops with jitter, and no-data-loss guarantees. The control plane runs eight always-on subsystems: live caching, pipeline catalog, version tracking, continuous learning, lineage, task scheduling, metrics collection, and failure management.

Rich Connector Ecosystem Sources: codebases, meeting notes, web APIs, file systems, blob stores, databases, message queues, images, video, voice transcripts. Targets: relational databases, data warehouses, vector databases (pgvector, LanceDB), graph databases (Neo4j, Kuzu, SurrealDB), message queues (Kafka), feature stores. The framework grows weekly with community contributions.

Real-World Use Cases Where CocoIndex Dominates

Live RAG Pipelines for Customer Support Your documentation lives in Notion, your API reference auto-generates from code, and your Slack #support channel contains thousands of resolved threads. Traditional approach: nightly batch job that misses today's critical bug workaround. CocoIndex approach: continuous sync where a Notion edit propagates to your agent's context in under a second. Customer asks about the edge case you just documented? Your agent knows immediately.

Coding Agents with Semantic Repository Understanding The flagship CocoIndex-code demonstrates the power: AST-aware chunking, call graph extraction, semantic search by meaning rather than grep. Claude Code or Cursor queries "where do we handle OAuth token refresh?" and gets the exact function, its callers, and related tests — not 47 string matches. Incremental indexing means new commits integrate instantly, keeping the agent's mental model current.

Knowledge Graphs from Conversational Data Meeting transcripts, podcast episodes, support calls — extract people, topics, decisions, and action items with LLM-powered entity extraction, resolve duplicates across episodes with embeddings, and maintain a living graph in Neo4j or Kuzu. Only changed conversation turns trigger re-extraction, making continuous ingestion economically viable.

Multi-Source Competitive Intelligence Monitor competitor pricing pages, SEC filings, job postings, and press releases. Each source has different change patterns: hourly API polls, daily page scrapes, quarterly PDF drops. CocoIndex handles the heterogeneity, maintains per-source freshness guarantees, and gives your strategy agents unified, current context without manual orchestration.

Feature Stores for ML Pipelines Transform raw event streams into aggregated features with complex windowing and embedding logic. The incremental engine ensures your feature store stays current without recomputing historical windows that haven't changed. Backfill once, increment forever — the holy grail of feature engineering.

Step-by-Step Installation & Setup Guide

Getting started with CocoIndex takes under five minutes. The framework is distributed via PyPI with pre-built wheels for major platforms.

Prerequisites

Python 3.10, 3.11, 3.12, or 3.13
A target database (PostgreSQL^{↗ Bright Coding Blog} with pgvector recommended for vector workloads)
Optional: Rust toolchain if building from source

Installation

# Install the core package
pip install -U cocoindex

# For vector database targets, install additional drivers as needed
pip install psycopg2-binary  # PostgreSQL
pip install lancedb          # LanceDB alternative

Database Setup for PostgreSQL Target

# Ensure your PostgreSQL instance has pgvector extension
createdb my_agent_db
psql my_agent_db -c "CREATE EXTENSION IF NOT EXISTS vector;"

Environment Configuration

CocoIndex uses standard connection strings. Set your database URL:

export COCOINDEX_DATABASE_URL="postgresql://user:pass@localhost:5432/my_agent_db"

For embedding models, you'll typically need API keys:

export OPENAI_API_KEY="sk-..."
# or
export COHERE_API_KEY="..."

First Pipeline Verification

Create a minimal test file to verify installation:

# test_install.py
import cocoindex as coco

print(f"CocoIndex version: {coco.__version__}")
print("Installation verified!")

python test_install.py

Project Structure Convention

CocoIndex projects typically follow this pattern:

my-agent-pipeline/
├── flow.py          # Main pipeline definition
├── transforms.py    # Reusable transformation functions
├── requirements.txt
└── .env             # Environment variables (gitignored)

REAL Code Examples from the Repository

Let's examine the actual patterns from CocoIndex's documentation, with detailed explanations of how the incremental magic works.

Example 1: Basic Document Indexing Pipeline

This is the canonical "hello world" from the README — and it contains more sophistication than most production ETL systems:

import cocoindex as coco
from cocoindex.connectors import localfs, postgres
from cocoindex.ops.text import RecursiveSplitter

# The @coco.fn(memo=True) decorator enables code-aware caching.
# The engine hashes both the input data AND the function's source code.
# If you modify embed() or split logic, affected records recompute automatically.
@coco.fn(memo=True)
async def index_file(file, table):
    # RecursiveSplitter handles intelligent chunking — respects paragraph boundaries,
    # code blocks, and semantic breaks rather than naive character counting
    for chunk in RecursiveSplitter().split(await file.read_text()):
        # declare_row builds the target schema incrementally
        # The engine tracks which source bytes produced which target rows
        table.declare_row(
            text=chunk.text,
            embedding=embed(chunk.text)  # Your embedding function here
        )

@coco.fn
async def main(src):
    # Mount a PostgreSQL table as the target — the engine manages schema evolution
    table = await postgres.mount_table_target(PG, table_name="docs")
    
    # Declare a vector index for similarity search — created automatically if missing
    table.declare_vector_index(column="embedding")
    
    # mount_each applies index_file to every item from the source walker
    # Parallelism, batching, and retry logic are handled by the engine
    await coco.mount_each(index_file, localfs.walk_dir(src).items(), table)

# AppConfig names the pipeline for monitoring and versioning
coco.App(
    coco.AppConfig(name="docs"),
    main,
    src="./docs"
).update_blocking()  # Blocking call for scripts; async variant available for services

What makes this incremental? The first run backfills everything. Subsequent runs? The engine compares source file mtimes, hashes, and content against its lineage store. Unchanged files hit the cache. Modified files flow through index_file with their new content. Deleted files trigger target row retirement. You literally re-run the same script — no manual change detection.

Example 2: React-Style Declarative Pattern

The core mental model expressed in code:

# Target = F(Source) — declare the desired state, engine maintains it
# This is NOT a DAG definition. It's a persistent state declaration.
# The engine continuously reconciles actual state with declared state.

@coco.fn(memo=True)
async def transform_document(source_doc, target_table):
    """
    This function defines the transformation F.
    The engine handles:
    - When to call it (source changed, code changed, or dependency changed)
    - How to parallelize it (across files, across chunks within files)
    - What to cache (output keyed by hash(input) + hash(this function's AST))
    - How to recover from failures (retry with backoff, dead letter queue)
    """
    extracted = await extract_entities(source_doc.content)
    target_table.declare_row(
        doc_id=source_doc.id,
        entities=extracted.entities,
        embedding=extracted.embedding,
        updated_at=coco.now()  # Engine-managed timestamp for freshness tracking
    )

Example 3: Code-Aware Caching in Action

The memo=True parameter is where the "React for data engineering" analogy becomes concrete:

# Version 1: Using sentence-transformers/all-MiniLM-L6-v2
@coco.fn(memo=True)
async def embed(text: str) -> list[float]:
    return sentence_transformer.encode(text).tolist()

# You deploy this, backfill completes, everything works.
# Now you want to upgrade to a better model...

# Version 2: Using text-embedding-3-large
@coco.fn(memo=True)
async def embed(text: str) -> list[float]:
    return openai.embeddings.create(
        model="text-embedding-3-large",
        input=text
    ).data[0].embedding

# With a traditional pipeline: manual migration script, full recompute, hours of downtime
# With CocoIndex: the engine detects hash(code) changed, invalidates ONLY affected cache entries,
# incrementally re-embeds using the new model, maintains zero-downtime cutover

Example 4: Hacker News Trending Topics (Real Example)

From the examples directory, showing complex multi-stage transformation:

# Fetches HN threads via Algolia API, recursively pulls nested comments,
# LLM-extracts typed topic lists per message with Gemini 2.5 Flash,
# ranks topics by weighted mention count

@coco.fn(memo=True)
async def extract_topics(message: HNMessage) -> list[Topic]:
    """
    LLM extraction is expensive — memoization makes it viable to run continuously.
    Same message content + same extraction prompt = cached result, no API call.
    """
    return await gemini_extract_topics(
        text=message.text,
        schema=Topic,  # Pydantic model for structured output
        model="gemini-2.5-flash"
    )

@coco.fn
async def rank_topics(topics_table, messages_source):
    # Thread = 5 points, comment = 1 point — business logic in plain Python
    weights = {"story": 5, "comment": 1}
    
    async for msg in messages_source.items():
        extracted = await extract_topics(msg)
        for topic in extracted:
            topics_table.declare_row(
                name=topic.name,
                score=weights.get(msg.type, 1),
                source_url=msg.url
            )
    
    # Aggregation happens in the target store — efficient for large corpora
    topics_table.declare_aggregate(
        group_by="name",
        total_score=coco.sum("score"),
        latest_source=coco.max_by("source_url", "score")
    )

Advanced Usage & Best Practices

Design for Cacheability Structure transformations so that expensive operations (LLM calls, embeddings) are in memo=True functions with stable inputs. Avoid including timestamps or randomness in cache keys unless necessary. The engine's invalidation is precise — leverage it.

Monitor Lineage for Debugging When results seem wrong, use the lineage API to trace from suspicious target rows back to source bytes. This isn't just for debugging — it's your audit trail for compliance and your explanation mechanism for agent behavior.

Leverage Parallel Source Walkers For large corpora, use localfs.walk_dir() with appropriate glob patterns rather than loading file lists into memory. The engine's scheduler handles backpressure and parallelism automatically.

Schema Evolution Strategy When adding columns to target tables, the engine handles additive changes automatically. For destructive changes, version your pipeline name (docs_v2) and run parallel until cutover is verified.

Embedding Cost Optimization The Δ-only processing typically reduces embedding API calls by 90%+ for typical churn patterns. For additional savings, implement batch embedding within your @coco.fn — the engine will still cache at the individual row level.

Comparison with Alternatives

Feature	CocoIndex	Airflow	dbt	LangChain Indexing	Custom Scripts
Incremental by default	✅ Native	❌ Manual	❌ Manual	⚠️ Partial	❌ You build it
Δ-only processing	✅ Fine-grained lineage	❌ Full re-runs	❌ Full re-runs	⚠️ Document-level	❌ Rarely implemented
Code-change invalidation	✅ Hash-based	❌ N/A	❌ N/A	❌ N/A	❌ N/A
Declarative API	✅ React-style	❌ DAG imperative	⚠️ SQL-centric	❌ Imperative	❌ Ad-hoc
Rust core performance	✅ Zero-copy, parallel	❌ Python overhead	❌ SQL execution	❌ Python overhead	⚠️ Varies
Vector DB targets	✅ First-class	⚠️ Via operators	❌ No	✅ Yes	⚠️ Manual
Production resilience	✅ Built-in retries, DLQ	⚠️ Via plugins	❌ Limited	❌ None	❌ You build it
Setup complexity	⭐ 5 minutes	⭐⭐⭐ Hours	⭐⭐⭐ Hours	⭐⭐ Moderate	⭐⭐⭐⭐ Days
Cost at scale	💰 10× cheaper	💰💰💰 Full recompute	💰💰💰 Full recompute	💰💰 Re-embed docs	💰💰💰💰 Unoptimized

When to choose what:

CocoIndex: Live agent context, RAG pipelines, continuous knowledge graphs, any scenario where freshness matters and compute costs scale with corpus size.
Airflow/dbt: Traditional analytics with daily/hourly freshness sufficient, heavy SQL transformations, existing team expertise.
LangChain Indexing: Quick prototypes, simple document loaders, when you don't need incremental processing or production resilience.
Custom scripts: Never, unless you're specifically building a CocoIndex competitor.

FAQ

Does CocoIndex replace my vector database? No — it feeds it. CocoIndex is the pipeline layer that keeps your vector database (pgvector, LanceDB, Pinecone, etc.) synchronized with source data. You still query the vector database directly.

How does the incremental engine handle schema changes in source data? The engine tracks schema versions and triggers appropriate invalidation. For compatible changes (new optional fields), existing cached results remain valid. For breaking changes, affected records recompute automatically.

Can I use CocoIndex with my existing Python data processing code? Yes — wrap existing functions with @coco.fn(memo=True) and integrate with CocoIndex connectors. The framework is designed to adopt incrementally (fittingly) rather than requiring greenfield rewrites.

What's the overhead of the Rust core for small projects? Minimal. The Rust engine is compiled to native extensions with minimal Python interop overhead. For small corpora, you might not notice the difference. For large corpora, the performance advantage becomes dramatic.

How does CocoIndex compare to dedicated CDC tools like Debezium? CocoIndex includes CDC capabilities but goes further — it handles transformation logic versioning, target schema management, and end-to-end lineage. For simple database replication, Debezium might suffice. For AI agent context pipelines, CocoIndex provides the complete solution.

Is there a managed/cloud version? CocoIndex Enterprise offers PB-scale deployment with additional features. The open-source version is production-ready for most use cases — check the GitHub repository for current capabilities.

What embedding models work with CocoIndex? Any — you bring your own embed() function. Popular choices include OpenAI's text-embedding series, Cohere, sentence-transformers, and custom fine-tuned models. The memoization works regardless of provider.

Conclusion

The era of nightly batch jobs for AI agent context is ending. Agents that reason over stale data are liabilities, not assets. CocoIndex offers a fundamentally better approach: declare your desired target state in clean Python, and let a Rust-powered incremental engine maintain it with surgical precision — sub-second freshness, 10× cost reduction, and production-grade reliability built in.

I've evaluated dozens of pipeline frameworks, and the React-for-data-engineering mental model is the first that genuinely simplifies rather than complicates. The @coco.fn(memo=True) decorator alone eliminates entire categories of bugs and optimization work that consumed my previous teams.

The community is growing fast, the examples are expanding weekly, and the core team ships with impressive velocity. Whether you're building coding agents, customer support bots, competitive intelligence systems, or knowledge graphs — if your agents need fresh context from dynamic sources, CocoIndex deserves your evaluation.

Ready to stop re-embedding everything? Star the repository, run pip install -U cocoindex, and get your first incremental pipeline running in five minutes. Your future self — and your cloud bill — will thank you.

Built with 🥥 by the CocoIndex community. Apache 2.0 licensed.