oms-cocoindex

oms-cocoindex

Overview

cocoindex is a Python ETL framework with a Rust engine for building incremental data indexes (embeddings, knowledge graphs, vector search, LLM extraction). Users author flows in Python; the Rust engine handles incremental recomputation and target state management.

• Source: cocoindex-io/cocoindex @ v0.3.37 (commit 87c5dbf0)
• Forge tier: Deep — AST structural extraction + QMD temporal/docs enrichment
• Exports documented: 102 public exports (T1 AST-verified) across flow, lib, index, llm, setting, auth_registry, query_handler, typing, op, sources, targets, functions, cli, utils
• Confidence distribution: T1 = 102, T2 = 15, T3 = 10 (docs), T1-low = 0

Note on stability: cocoindex is Development Status 3 — Alpha. This skill is pinned to tag v0.3.37; upstream has since moved to v1.0.0-alpha*. Re-forge for newer versions.

Quick Start

End-to-end text-embedding flow — read markdown files, chunk, embed with SentenceTransformer, export to Postgres + pgvector:

import cocoindex

@cocoindex.flow_def(name="TextEmbedding")
def text_embedding_flow(
    flow_builder: cocoindex.FlowBuilder,
    data_scope: cocoindex.DataScope,
):
    data_scope["documents"] = flow_builder.add_source(
        cocoindex.sources.LocalFile(path="markdown_files")
    )
    doc_embeddings = data_scope.add_collector()

    with data_scope["documents"].row() as doc:
        doc["chunks"] = doc["content"].transform(
            cocoindex.functions.SplitRecursively(),
            language="markdown", chunk_size=2000, chunk_overlap=500,
        )
        with doc["chunks"].row() as chunk:
            chunk["embedding"] = chunk["text"].transform(
                cocoindex.functions.SentenceTransformerEmbed(
                    model="sentence-transformers/all-MiniLM-L6-v2"
                )
            )
            doc_embeddings.collect(
                filename=doc["filename"],
                location=chunk["location"],
                text=chunk["text"],
                embedding=chunk["embedding"],
            )

    doc_embeddings.export(
        "doc_embeddings",
        cocoindex.targets.Postgres(),
        primary_key_fields=["filename", "location"],
        vector_indexes=[
            cocoindex.VectorIndexDef(
                field_name="embedding",
                metric=cocoindex.VectorSimilarityMetric.COSINE_SIMILARITY,
            )
        ],
    )