Customizable Data Indexing Pipelines

Q: What is custom transformation logic in a data indexing pipeline?

Custom transformation logic means you decide what the pipeline does and how, rather than accepting a predesigned service. This includes basic component choices (which parser, chunking strategy, embedding model, vector database), the pipeline's goal (simple text embedding, building a knowledge graph, or summarizing each source without chunking), and quality work such as deduplication, looking up other sources to enrich data, and reconciling multiple documents.In CocoIndex, custom transformation logic is ordinary Python: any function decorated with @coco.fn can become a pipeline step, and memoization keeps expensive steps incremental.

CocoIndex is an open-source engine built for data indexing that supports both custom transformation logic and incremental processing. So, what is custom transformation logic? In CocoIndex it is ordinary Python: any function decorated with @coco.fn can become a pipeline step, and marking it with memo=True makes it incremental, so an unchanged document never pays for the same parse, LLM call, or embedding twice.

Index-as-a-service (or RAG-as-a-service) tends to package a predesigned service and expose two endpoints to users: one to configure the source, and an API to read from the index. Many predefined pipelines for unstructured documents do this. The requirements are fairly simple: parse PDFs, perform some chunking and embedding, and dump into vector stores. This works well if your requirements are simple and primarily focused on document parsing.

We’ve talked to many developers across various verticals that require data indexing, and being able to customize logic is essential for high-quality data retrieval. For example:

Basic choices for pipeline components
- which parser for different files?
- how to chunk files (documents with structure normally have different optimal chunking strategies)?
- which embedding model? which vector database?
What should the pipeline do?
- Is it a simple text embedding?
- Is it building a knowledge graph?
- Should it perform simple summarization for each source for retrieval without chunking?
What additional work is needed to improve pipeline quality?
- Do we need deduplication?
- Do we need to look up different sources to enrich our data?
- Do we need to reconcile and align multiple documents?

Here we’ll walk through some examples of the topology of index pipelines, and we can explore more in the future. If you’re new to the framework, the core concepts and the text embedding example are a good starting point.

Basic embedding

Basic embedding pipeline: source documents flow into a per-document loop that converts each file to Markdown and splits it to chunks, a nested per-chunk loop embeds each chunk, and the embeddings land in a vector index

In this example, we do the following:

Read from sources, for example, a list of PDFs
For each source file, parse it to markdown with a PDF parser. There are lots of choices out there: Llama Parse, Unstructured, Docling, or a vision model such as Gemini.
Chunk all the markdown files. This is a way to break text into smaller chunks, or units, to help organize and process information. Options range from flat to hierarchical chunking, with many publications in this area. There are also special chunking strategies for different verticals: for code, tools like Tree-sitter can help parse and chunk based on syntax. Normally the best choice is tied to your document structure and requirements.
Perform embedding for each chunk, with the model of your choice: Voyage, OpenAI, or a local model.
Collect the embeddings in vector stores. The embedding is normally attached with metadata, for example, which file this embedding belongs to. Popular choices include Chroma, Milvus, and Pinecone, and many databases now support vector indexing, for example PostgreSQL (pgvector) and MongoDB. CocoIndex ships target connectors for PostgreSQL, SQLite, LanceDB, Qdrant, SurrealDB, and Apache Doris.

In CocoIndex, this whole topology is a short piece of Python, and every step is a point where you plug in your own choice:

python

@coco.fn(memo=True)
async def process_file(file: FileLike, table: postgres.TableTarget[DocChunk]) -> None:
    text = await parse_to_markdown(file)  # your parser choice
    id_gen = IdGenerator()
    for chunk in splitter.split(text, chunk_size=2000, chunk_overlap=500):
        embedding = await coco.use_context(EMBEDDER).embed(chunk.text)  # your model choice
        table.declare_row(row=DocChunk(
            id=await id_gen.next_id(chunk.text),
            filename=str(file.file_path.path),
            text=chunk.text,
            embedding=embedding,
        ))

await coco.mount_each(process_file, files.items(), table)

The PDF embedding example is a runnable version of this pipeline.

A combination of keyword search and vector search

Anthropic has published a great article about Contextual Retrieval that suggests combining embedding-based retrieval with BM25, a lexical ranking function built on TF-IDF.

The way to think about a data flow for the pipeline is:

In addition to preparing the vector embedding as in the basic embedding example above, after the source data parsing, we can do the following:

For each document, extract keywords from it with their frequencies.
For each keyword, store the keyword with its per-document frequency into a keyword index.

The corpus-wide part of the computation belongs at query time. Ranking functions like TF-IDF and BM25 need to know, for each keyword, how many documents contain it, and that number changes whenever any document changes. Instead of recomputing it in the pipeline on every update, keep the pipeline per-document and let the database aggregate: a GROUP BY over the keyword index yields the document frequencies, and the score combines them with the per-document frequencies already stored.

At query time, we query both the vector index and the keyword index, score the keyword matches, and combine results from both. The HackerNews trending topics example uses the same split: per-item extraction in the pipeline, cross-item aggregation in SQL.

Simple data lookup/enrichment example

Sometimes, you want to enrich your data with metadata looked up from other sources. For example, if we want to create an index on diagnostic reports, which use ICD-10 (International Classification of Diseases Version 10) codes to describe diseases, we can have a pipeline like this:

In this example, we do the following:

On the first path, build an ICD-10 dictionary by
1. For each ICD-10 description document, convert it to markdown with a PDF parser.
2. Split into items.
3. For each item, extract the ICD-10 code and description, and collect them into a dictionary.
On the second path, for each report
1. Parse it to markdown with a PDF parser.
2. Split into items.
3. For each item, look up the ICD-10 dictionary prepared above and enrich the item with the descriptions for ICD-10 codes.

The ordering between the two paths is exactly what use_mount expresses: mount the dictionary-building component, wait for its result, then fan out over the reports with it.

python

@coco.fn
async def app_main(dict_dir: pathlib.Path, report_dir: pathlib.Path) -> None:
    icd10 = await coco.use_mount(build_icd10_dictionary, dict_dir)  # first path
    reports = localfs.walk_dir(report_dir, recursive=True)
    await coco.mount_each(process_report, reports.items(), icd10)   # second path

Now we have a vector index, built based on diagnostic reports enriched with ICD-10 descriptions.

CocoIndex

Fresh context for long-horizon agents.

Learn the concept → View on GitHub

Frequently asked questions.

What is custom transformation logic in a data indexing pipeline?

Custom transformation logic means you decide what the pipeline does and how, rather than accepting a predesigned service. This includes basic component choices (which parser, chunking strategy, embedding model, vector database), the pipeline's goal (simple text embedding, building a knowledge graph, or summarizing each source without chunking), and quality work such as deduplication, looking up other sources to enrich data, and reconciling multiple documents.

In CocoIndex, custom transformation logic is ordinary Python: any function decorated with @coco.fn can become a pipeline step, and memoization keeps expensive steps incremental.

How is a custom indexing pipeline different from index-as-a-service or RAG-as-a-service?

Index-as-a-service (or RAG-as-a-service) packages a predesigned service and exposes two endpoints: one to configure the source and an API to read from the index. That works when requirements are simple (parse PDFs, chunk, embed, dump into a vector store). But developers across many verticals find that customizing logic is essential for high-quality retrieval, which is what a custom pipeline gives you.

What does a basic embedding pipeline look like?

A basic embedding pipeline reads from sources (for example a list of PDFs), parses each file to markdown with a PDF parser, chunks the markdown, computes an embedding for each chunk, and collects the embeddings (with metadata such as the source file) into a vector store. The post notes many choices at each step, including parsers like Llama Parse, Unstructured, and Docling, and vector stores like Chroma, Milvus, Pinecone, and pgvector.

See Basic embedding.

How do you combine vector search with keyword search?

Following Anthropic's Contextual Retrieval idea, which recommends pairing embeddings with BM25 (a lexical ranking function built on TF-IDF), you prepare the vector embeddings as usual, and in addition extract keywords with their frequencies per document and store each keyword with its per-document frequency in a keyword index. The corpus-wide statistics that BM25 needs, such as how many documents contain each keyword, are computed at query time by aggregating over the keyword index. You then query both the vector index and the keyword index, score the keyword matches, and combine the results.

See A combination of keyword search and vector search.

How do you enrich indexed data with metadata looked up from another source?

The post's ICD-10 example uses two paths. The first builds a dictionary: convert each ICD-10 description document to markdown, split into items, and for each item extract the ICD-10 code and description into a dictionary. The second path processes each report: parse to markdown, split into items, and for each item look up the dictionary to enrich it with the matching ICD-10 descriptions, producing a vector index over reports enriched with those descriptions.

See Simple data lookup/enrichment example.

Why does chunking strategy matter for retrieval quality?

Chunking breaks text into smaller units to help organize and process information, and there is no single best option: strategies range from flat to hierarchical chunking, along with vertical-specific approaches (for code, tools like Tree-sitter chunk based on syntax). The best choice is normally tied to your document structure and requirements.