| name | cocoindex-v1 |
| description | This skill should be used when building data processing pipelines with CocoIndex v1, a Python library for incremental data transformation. Use when the task involves processing files/data into databases, creating vector embeddings, building knowledge graphs, ETL workflows, or any data pipeline requiring automatic change detection and incremental updates. CocoIndex v1 is Python-native (supports any Python types), has no DSL, and is currently under pre-release (version 1.0.0a1 or later). Use when this capability is needed. |
| metadata | {"author":"cocoindex-io"} |
CocoIndex v1
CocoIndex v1 is a Python library for building incremental data processing pipelines with declarative target states. Think spreadsheets or React for data pipelines: declare what the output should look like based on current input, and CocoIndex automatically handles incremental updates, change detection, and syncing to external systems.
Overview
CocoIndex v1 enables building data pipelines that:
- Automatically handle incremental updates: Only reprocess changed data
- Use declarative target states: Declare what should exist, not how to update
- Support any Python types: No custom DSLโuse dataclasses, Pydantic, NamedTuple
- Provide function memoization: Skip expensive operations when inputs/code unchanged
- Sync to multiple targets: PostgreSQL, SQLite, LanceDB, Qdrant, file systems
Key principle: TargetState = Transform(SourceState)
When to Use This Skill
Use this skill when building pipelines that involve:
- Document processing: PDF/Markdown conversion, text extraction, chunking
- Vector embeddings: Embedding documents/code for semantic search
- Database transformations: ETL from source DB to target DB
- Knowledge graphs: Extract entities and relationships from data
- LLM-based extraction: Structured data extraction using LLMs
- File-based pipelines: Transform files from one format to another
- Incremental indexing: Keep search indexes up-to-date with source changes
Quick Start: Creating a New Project
Initialize Project
Use the built-in CLI to create a new project:
cocoindex init my-project
cd my-project
This creates:
main.py - Main app definition
pyproject.toml - Dependencies with pre-release config
.env - Environment configuration
README.md - Quick start guide
Add Dependencies for Specific Use Cases
Add dependencies to pyproject.toml based on your needs:
dependencies = ["cocoindex>=1.0.0a1", "sentence-transformers", "asyncpg"]
dependencies = ["cocoindex>=1.0.0a1", "asyncpg"]
dependencies = ["cocoindex>=1.0.0a1", "litellm", "instructor", "pydantic>=2.0"]
See references/setup_project.md for complete examples.
Set Up Database (if using Postgres/Qdrant)
For PostgreSQL with Docker:
docker-compose up -d
For Qdrant with Docker:
docker-compose up -d
See references/setup_database.md for detailed setup instructions.
Run the Pipeline
pip install -e .
cocoindex update main.py
Core Concepts
1. Apps
An app is the top-level executable that binds a main function with parameters:
import cocoindex as coco
@coco.function
def app_main(sourcedir: pathlib.Path, outdir: pathlib.Path) -> None:
...
app = coco.App(
coco.AppConfig(name="MyApp"),
app_main,
sourcedir=pathlib.Path("./data"),
outdir=pathlib.Path("./output"),
)
if __name__ == "__main__":
app.update(report_to_stdout=True)
2. Processing Components
A processing component groups an item's processing with its target states.
Mount independent components with coco_aio.mount_each() (preferred) or coco_aio.mount():
await coco_aio.mount_each(process_file, files.items(), target_table)
for key, f in files.items():
await coco_aio.mount(coco.component_subpath(key), process_file, f, target_table)
coco.mount(coco.component_subpath(str(f.file_path.path)), process_file, f, target_table)
Mount dependent components with use_mount() when you need the return value:
result = await coco_aio.use_mount(subpath, fn, *args)
Mount targets using connector convenience methods (async, subpath is automatic):
target_table = await target_db.mount_table_target(
table_name="my_table",
table_schema=await postgres.TableSchema.from_class(MyRecord, primary_key=["id"]),
)
Key points:
- Each component runs independently
- Async-first: prefer
coco_aio.mount_each() / coco_aio.mount() for all components; use sync coco.mount() only for CPU-intensive leaf work (no I/O)
- Use
use_mount() when you need the return value of a child component
- Use stable paths for proper memoization
- Component path determines target state ownership
3. Function Memoization
Add memo=True to skip re-execution when inputs/code unchanged:
@coco.function(memo=True)
def expensive_operation(data: str) -> Result:
result = expensive_transform(data)
return result
4. Target States
Declare what should existโCocoIndex handles creation/update/deletion:
localfs.declare_file(outdir / "output.txt", content)
table.declare_row(row=MyRecord(id=1, name="example"))
collection.declare_point(point=PointStruct(id="1", vector=[...]))
5. Context for Shared Resources
Use ContextKey to share expensive resources across components:
EMBEDDER = coco.ContextKey[SentenceTransformerEmbedder]("embedder")
@coco.lifespan
def coco_lifespan(builder: coco.EnvironmentBuilder):
embedder = SentenceTransformerEmbedder("all-MiniLM-L6-v2")
builder.provide(EMBEDDER, embedder)
yield
The @coco.lifespan decorator registers the function to the default CocoIndex environment, which is shared among all apps by default.
@coco.function
def process_item(text: str) -> None:
embedder = coco.use_context(EMBEDDER)
embedding = embedder.embed(text)
6. ID Generation
Generate stable, unique identifiers that persist across incremental updates:
from cocoindex.resources.id import generate_id, IdGenerator
chunk_id = generate_id(chunk.content)
id_gen = IdGenerator()
for chunk in chunks:
chunk_id = id_gen.next_id(chunk.content)
table.declare_row(row=Row(id=chunk_id, content=chunk.content))
Use generate_id(dep) when same content should yield same ID. Use IdGenerator when you need distinct IDs even for duplicate content. See ID Generation docs for details.
Common Pipeline Patterns
Pattern 1: File Transformation
Transform files from input to output directory:
import pathlib
import cocoindex as coco
import cocoindex.asyncio as coco_aio
from cocoindex.connectors import localfs
from cocoindex.resources.file import PatternFilePathMatcher
@coco.function(memo=True)
def process_file(file, outdir):
content = file.read_text()
transformed = transform_content(content)
outname = file.file_path.path.stem + ".out"
localfs.declare_file(outdir / outname, transformed, create_parent_dirs=True)
@coco.function
async def app_main(sourcedir, outdir):
files = localfs.walk_dir(
sourcedir,
recursive=True,
path_matcher=PatternFilePathMatcher(
included_patterns=["*.txt", "*.md"],
excluded_patterns=[".*/**"],
),
)
await coco_aio.mount_each(process_file, files.items(), outdir)
app = coco_aio.App(coco_aio.AppConfig(name="Transform"), app_main, sourcedir=pathlib.Path("./data"), outdir=pathlib.Path("./out"))
Pattern 2: Vector Embedding Pipeline
Chunk and embed documents for semantic search:
import pathlib
from dataclasses import dataclass
from typing import Annotated, AsyncIterator
import cocoindex as coco
import cocoindex.asyncio as coco_aio
from cocoindex.connectors import localfs, postgres
from cocoindex.ops.text import RecursiveSplitter
from cocoindex.ops.sentence_transformers import SentenceTransformerEmbedder
from cocoindex.resources.chunk import Chunk
from cocoindex.resources.file import FileLike, PatternFilePathMatcher
from cocoindex.resources.id import IdGenerator
from numpy.typing import NDArray
PG_DB = coco.ContextKey[postgres.PgDatabase]("pg_db")
_embedder = SentenceTransformerEmbedder("sentence-transformers/all-MiniLM-L6-v2")
_splitter = RecursiveSplitter()
@dataclass
class DocEmbedding:
id: int
filename: str
text: str
embedding: Annotated[NDArray, _embedder]
chunk_start: int
chunk_end: int
@coco_aio.lifespan
async def coco_lifespan(builder: coco_aio.EnvironmentBuilder) -> AsyncIterator[None]:
async with postgres.create_pool(DATABASE_URL) pool:
builder.provide(PG_DB, postgres.register_db(, pool))
():
table.declare_row(
row=DocEmbedding(
= id_gen.next_id(chunk.text),
filename=(filename),
text=chunk.text,
embedding= _embedder.embed(chunk.text),
chunk_start=chunk.start.char_offset,
chunk_end=chunk.end.char_offset,
),
)
():
text = file.read_text()
chunks = _splitter.split(text, chunk_size=, chunk_overlap=)
id_gen = IdGenerator()
coco_aio.(process_chunk, chunks, file.file_path.path, id_gen, table)
():
target_db = coco.use_context(PG_DB)
target_table = target_db.mount_table_target(
table_name=,
table_schema= postgres.TableSchema.from_class(
DocEmbedding, primary_key=[],
),
)
files = localfs.walk_dir(sourcedir, recursive=)
coco_aio.mount_each(process_file, files.items(), target_table)
app = coco_aio.App(coco_aio.AppConfig(name=), app_main, sourcedir=Path())
Pattern 3: LLM-Based Extraction
Extract structured data using LLMs:
import instructor
from pydantic import BaseModel
from litellm import acompletion
_instructor_client = instructor.from_litellm(acompletion, mode=instructor.Mode.JSON)
class ExtractionResult(BaseModel):
title: str
topics: list[str]
@coco.function(memo=True)
async def extract_and_store(content, message_id, table):
result = await _instructor_client.chat.completions.create(
model="gpt-4",
response_model=ExtractionResult,
messages=[{"role": "user", "content": f"Extract topics: {content}"}],
)
table.declare_row(row=Message(id=message_id, title=result.title, content=content))
Connectors and Operations
CocoIndex v1 provides connectors for reading from and writing to various external systems including databases (SQL and vector), file systems, and more.
For detailed connector documentation, see:
Text and Embedding Operations
Text Splitting
from cocoindex.ops.text import RecursiveSplitter, detect_code_language
splitter = RecursiveSplitter()
language = detect_code_language(filename="example.py")
chunks = splitter.split(
text,
chunk_size=1000,
min_chunk_size=300,
chunk_overlap=200,
language=language,
)
Embeddings
from cocoindex.ops.sentence_transformers import SentenceTransformerEmbedder
embedder = SentenceTransformerEmbedder("sentence-transformers/all-MiniLM-L6-v2")
embedding = embedder.embed(text)
embedding = await embedder.embed_async(text)
CLI Commands
Run Pipeline
cocoindex update main.py
cocoindex update main.py:my_app
cocoindex update my_module:my_app
Drop All State
cocoindex drop main.py [-f]
List Apps
cocoindex ls main.py
cocoindex ls --db ./cocoindex.db
Show Component Paths
cocoindex show main.py
Best Practices
1. Use Stable Component Paths
coco.component_subpath("file", str(file.file_path.path))
coco.component_subpath("record", record.id)
coco.component_subpath("file", file)
coco.component_subpath("idx", idx)
2. Add Memoization for Expensive Operations
@coco.function(memo=True)
async def process_chunk(chunk, table):
embedding = await embedder.embed_async(chunk.text)
table.declare_row(...)
@coco.function
async def process_chunk(chunk, table):
embedding = await embedder.embed_async(chunk.text)
3. Use Context for Shared Resources
@coco.lifespan
def coco_lifespan(builder):
model = load_expensive_model()
builder.provide(MODEL_KEY, model)
yield
@coco.function
def process(data):
model = load_expensive_model()
4. Use Type Annotations
from dataclasses import dataclass
from typing import Annotated
from numpy.typing import NDArray
@dataclass
class Record:
id: int
name: str
vector: Annotated[NDArray, embedder]
record = {"id": 1, "name": "example", "vector": [...]}
5. Use Convenience APIs for Targets and Iteration
table = await target_db.mount_table_target(
table_name="my_table",
table_schema=await postgres.TableSchema.from_class(MyRecord, primary_key=["id"]),
)
await coco_aio.mount_each(process_item, items.items(), table)
6. Prefer Async Mount
@coco.function
async def app_main(sourcedir):
await coco_aio.mount_each(process_file, files.items(), table)
await coco_aio.mount(coco.component_subpath("setup"), setup_fn)
@coco.function(memo=True)
def cpu_heavy_leaf(data: str) -> Result:
return expensive_computation(data)
@coco.function
async def app_main(sourcedir):
for key, f in files.items():
coco.mount(coco.component_subpath(key), process_file, f)
Migration from Old API
| Before | After |
|---|
await mount_run(subpath, fn, *args).result() | await use_mount(subpath, fn, *args) |
for key, item in items: mount(subpath(key), fn, item, *args) | mount_each(fn, items, *args) |
with component_subpath("setup"): await mount_run(...) | await mount_target(target) or await db.mount_table_target(...) |
await asyncio.gather(*(fn(item) for item in items)) | await map(fn, items) |
Troubleshooting
"Module not found" Error
Ensure pyproject.toml has pre-release config:
[tool.uv]
prerelease = "explicit"
PostgreSQL pgvector Not Found
Enable the pgvector extension:
psql "postgres://localhost/db" -c "CREATE EXTENSION IF NOT EXISTS vector;"
See references/setup_database.md for detailed setup instructions.
Memoization Not Working
Check component paths are stable:
coco.component_subpath(file.stable_key)
coco.component_subpath(file)
Everything Reprocessing
Add memo=True to expensive functions:
@coco.function(memo=True)
async def process_item(item):
...
Resources
references/
assets/
- simple-template/: Minimal project template structure
Additional Resources
For AI Agents:
For Humans:
Version Note
This skill is for CocoIndex v1 (pre-release: >=1.0.0a1). It uses a completely different API from v0. Key differences:
- Python-native (no DSL)
- Any Python types supported
- No flow definitions required
- More flexible and seamless experience
Converted and distributed by TomeVault โ claim your Tome and manage your conversions.