DEV Community: CocoIndex

How I Built a Semantic Search Engine with CocoIndex

Linghua Jin — Tue, 02 Dec 2025 22:04:22 +0000

Introduction

In this tutorial, I'll walk you through how I built a semantic search engine using CocoIndex, an open-source Python library for creating powerful search experiences. If you've ever wanted to build a search feature that understands context and meaning (not just exact keyword matches), this post is for you!

What is CocoIndex?

CocoIndex is a lightweight semantic search library that makes it easy to index and search through documents using vector embeddings. Unlike traditional keyword-based search, semantic search understands the meaning behind queries, allowing users to find relevant results even when they use different words.

Why I Chose CocoIndex

I needed a search solution that was:

Easy to integrate - No complex setup or infrastructure required
Fast - Quick indexing and search performance
Semantic - Understanding context, not just keywords
Open source - Free to use and modify

CocoIndex checked all these boxes!

Getting Started

First, install CocoIndex:

pip install cocoindex

Building the Search Engine

Here's how I implemented the core functionality:

1. Initialize CocoIndex

from cocoindex import CocoIndex

2. Add Documents

@cocoindex.flow_def(name="TextEmbedding")
def text_embedding_flow(flow_builder: cocoindex.FlowBuilder, data_scope: cocoindex.DataScope):
    """
    Define an example flow that embeds text into a vector database.
    """
    data_scope["documents"] = flow_builder.add_source(
        cocoindex.sources.LocalFile(path="markdown_files"))

    doc_embeddings = data_scope.add_collector()

Index the documents

Process each document

with data_scope["documents"].row() as doc:
    doc["chunks"] = doc["content"].transform(
        cocoindex.functions.SplitRecursively(),
        language="markdown", chunk_size=2000, chunk_overlap=500)

Embed

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"])

Export

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

3. Perform Semantic Search

def search(pool: ConnectionPool, query: str, top_k: int = 5):
    table_name = cocoindex.utils.get_target_storage_default_name(text_embedding_flow, "doc_embeddings")
    query_vector = text_to_embedding.eval(query)

    with pool.connection() as conn:
        with conn.cursor() as cur:
            cur.execute(f"""
                SELECT filename, text, embedding <=> %s::vector AS distance
                FROM {table_name} ORDER BY distance LIMIT %s
            """, (query_vector, top_k))
            return [
                {"filename": row[0], "text": row[1], "score": 1.0 - row[2]}
                for row in cur.fetchall()
            ]

Key Features I Implemented

Fast Indexing

CocoIndex uses efficient vector storage, making indexing thousands of documents quick and painless.

Semantic Understanding

The search understands that "teaching computers" relates to "machine learning" even without exact keyword matches.

Customizable Embeddings

You can use different embedding models depending on your use case and accuracy requirements.

Real-World Example

I built a documentation search for my project with 500+ markdown files. With CocoIndex:

Indexing took less than 30 seconds
Search response time averaged 50ms
Users found relevant docs even with vague queries

Performance Tips

Batch indexing - Add multiple documents at once for better performance
Choose the right embedding model - Balance between accuracy and speed
Cache frequently accessed results - Store common queries for instant responses

Challenges I Faced

Challenge 1: Choosing Embedding Dimensions

Higher dimensions = better accuracy but slower performance. I settled on 384 dimensions as a sweet spot.

Challenge 2: Handling Large Document Collections

For collections over 10k documents, I implemented pagination and lazy loading.

Results

After implementing CocoIndex:

User satisfaction increased significantly
Implementation took only 2 days vs weeks for alternatives

Conclusion

CocoIndex made building a semantic search engine surprisingly simple. Whether you're building a documentation site, blog search, or product catalog, it's a fantastic tool that punches above its weight.

The library is actively maintained, well-documented, and the community is helpful. I highly recommend giving it a try for your next search implementation!

Resources

GitHub: CocoIndex Repository
Documentation: Official Docs
My Demo Project: Simple Vector Index Demo

Have you used CocoIndex or other semantic search libraries? Share your experience in the comments below!

Happy coding! 🚀

Multimodal Face Recognition Pipeline with CocoIndex: Real-Time Image and Vector Search

Linghua Jin — Tue, 29 Jul 2025 22:53:01 +0000

This post is also on cocoindex.io, with interactive figures. The code uses CocoIndex v1.

A folder of group photos is full of people, but a file index sees only filenames and bytes. Ask it for "every photo with this person in it" and it has nothing to compare.

This post builds the index that can answer that question. It finds every face in every photo, turns each face into 128 numbers, and stores one Qdrant point per face, with the filename and the face's box. A face you search with then comes back with the photos it appears in and where it is in each one. The interactive demo on cocoindex.io uses the 1927 Solvay Conference photo, one of the 4 images that ship with the example. The boxes are the real output of the example's code.

Detection and embedding come from the face_recognition library, which wraps dlib. CocoIndex v1 runs the pipeline around it: it walks the image folder, runs each new or changed photo once, fans each photo out into one point per face, and removes a photo's points when the photo leaves the folder. The full code is in the CocoIndex repo.

What a face index lets you build

Face embeddings identify people, and under laws such as the EU's GDPR (Article 9) and Illinois' BIPA they can count as biometric data. Index only photos you have the right to process this way.

Image in, one point per face out

A text pipeline splits a document into chunks. This pipeline splits a photo into faces, so 1 image becomes N points:

Walk images/ and match .jpg, .jpeg and .png files. Each file becomes its own processing component, keyed by its path.
Detect every face with dlib's CNN detector, and crop each one out of the full-size photo.
Embed each crop into a 128-float vector, and declare one Qdrant point per face. The point ID is a uuid5 of the filename and the box, and the payload carries both.

You write each step as a plain Python function, target_state = transformation(source_state). CocoIndex tracks which files changed, skips unchanged photos through memoization, and keeps the Qdrant collection in step with the folder.

One photo, end to end

The stepper follows the Solvay photo through the 3 functions that do the work. Step with the buttons or the arrow keys.

The rest of the post shows the code in the order it runs.

Prerequisites

Qdrant on the default ports: docker run -d -p 6333:6333 -p 6334:6334 qdrant/qdrant.
Python 3.11 or later. face_recognition needs dlib, which builds from source on some platforms; its installation notes cover CMake and the other build tools.

CocoIndex v1 keeps its own processing state in a local directory set by COCOINDEX_DB (the example's .env.example sets ./cocoindex.db), so you do not run a separate database for it.

Define the pipeline

Everything is in main.py. It starts with the imports, the constants and the Qdrant client's context key:

```python title=main.py
from future import annotations

import io
import os
import pathlib
import sys
import uuid
from dataclasses import dataclass
from typing import Any, AsyncIterator

import face_recognition
import numpy as np
from PIL import Image
from qdrant_client import QdrantClient

import cocoindex as coco
from cocoindex.connectors import localfs, qdrant
from cocoindex.resources.file import FileLike, PatternFilePathMatcher
from cocoindex.resources.schema import VectorSchema

QDRANT_COLLECTION = "face_embeddings"
FACE_EMBED_DIM = 128 # face_recognition (dlib) face encodings are 128-d
MAX_IMAGE_WIDTH = 1280 # the CNN detector is slow on large images; downscale first
TOP_K = 5

def qdrant_url() -> str:
return os.getenv("QDRANT_URL", "http://localhost:6334/")

QDRANT_DB = coco.ContextKeyQdrantClient




`MAX_IMAGE_WIDTH` bounds the cost of detection: the CNN detector is slow on large images, so wider photos are shrunk for it. A face is a box plus the cropped pixels:



```python title=main.py
@dataclass
class ImageRect:
    min_x: int
    min_y: int
    max_x: int
    max_y: int

@dataclass
class Face:
    """One detected face: its bounding box and the cropped PNG."""

    rect: ImageRect
    image: bytes

Detect and crop the faces

extract_faces runs the detector on a copy that is at most 1280 pixels wide, then multiplies each box by ratio so it lands on the original photo. The crop comes from the original, so the embedding sees every pixel of the face:

```python title=main.py
@coco.fn.as_async(runner=coco.GPU)
def extract_faces(content: bytes) -> list[Face]:
"""Detect faces in an image and return each as a (bounding box, cropped PNG)."""
orig = Image.open(io.BytesIO(content)).convert("RGB")

# Downscale large images for the detector, then map boxes back to full size.
if orig.width > MAX_IMAGE_WIDTH:
    ratio = orig.width / MAX_IMAGE_WIDTH
    small = orig.resize(
        (MAX_IMAGE_WIDTH, int(orig.height / ratio)),
        resample=Image.Resampling.BICUBIC,
    )
else:
    ratio = 1.0
    small = orig

faces: list[Face] = []
for top, right, bottom, left in face_recognition.face_locations(
    np.array(small), model="cnn"
):
    rect = ImageRect(
        min_x=int(left * ratio),
        min_y=int(top * ratio),
        max_x=int(right * ratio),
        max_y=int(bottom * ratio),
    )
    buf = io.BytesIO()
    orig.crop((rect.min_x, rect.min_y, rect.max_x, rect.max_y)).save(
        buf, format="PNG"
    )
    faces.append(Face(rect=rect, image=buf.getvalue()))

return faces




`face_locations` returns boxes as `(top, right, bottom, left)`; `ImageRect` stores them as `min_x`, `min_y`, `max_x`, `max_y`. On the Solvay photo (2560 × 1853), the detector sees 1280 × 926 and finds 29 faces.

### Embed one face

`embed_face` passes the whole crop to dlib's encoder as the face location and returns 128 floats:



```python title=main.py
@coco.fn.as_async(runner=coco.GPU)
def embed_face(face_png: bytes) -> list[float]:
    """Embed a single cropped face into a 128-d vector."""
    img = Image.open(io.BytesIO(face_png)).convert("RGB")
    arr = np.array(img)
    encoding = face_recognition.face_encodings(
        arr, known_face_locations=[(0, img.width - 1, img.height - 1, 0)]
    )[0]
    return encoding.tolist()

Both functions are synchronous dlib calls, so they are declared with @coco.fn.as_async(runner=coco.GPU). Callers await them, and the GPU runner runs them in a worker thread, one call at a time per GPU it finds (a machine without a GPU counts as 1).

One point per face

process_face embeds a face and declares its point. The ID comes from _face_id, a uuid5 of the filename and the box, so the same box in the same file always maps to the same point:

```python title=main.py
def _face_id(filename: str, rect: ImageRect) -> str:
key = f"{filename}|{rect.min_x},{rect.min_y},{rect.max_x},{rect.max_y}"
return str(uuid.uuid5(uuid.NAMESPACE_URL, key))

@coco.fn
async def process_face(
face: Face,
filename: str,
target: qdrant.CollectionTarget,
) -> None:
"""Embed one face and declare it as a Qdrant point."""
embedding = await embed_face(face.image)
target.declare_point(
qdrant.PointStruct(
id=_face_id(filename, face.rect),
vector=embedding,
payload={
"filename": filename,
"min_x": face.rect.min_x,
"min_y": face.rect.min_y,
"max_x": face.rect.max_x,
"max_y": face.rect.max_y,
},
)
)




`declare_point` does not write anything itself. It states which point should exist, and CocoIndex creates, updates or deletes points to match after the component finishes (see [target states](https://cocoindex.io/docs/programming_guide/target_state/)).

### One component per photo

`process_file` handles one image: it detects the faces, then runs `process_face` for each with [`coco.map`](https://cocoindex.io/docs/programming_guide/processing_component/#using-cocoindex-map-to-apply-transformations):



```python title=main.py
@coco.fn(memo=True)
async def process_file(file: FileLike, target: qdrant.CollectionTarget) -> None:
    """Detect every face in one image and index each one."""
    faces = await extract_faces(await file.read())
    filename = str(file.file_path.path)
    await coco.map(process_face, faces, filename, target)

coco.map runs the calls concurrently inside the current component. It creates no child components, so all the points of one photo belong to that photo's component, and they are synced together.
memo=True memoizes the whole component. While a photo and the code are unchanged, the photo skips detection and embedding, and its points are carried over as they are.

Wire up the app

The lifespan creates one Qdrant client for the app, and app_main declares the collection and mounts one process_file per image:

```python title=main.py
@coco.lifespan
async def coco_lifespan(builder: coco.EnvironmentBuilder) -> AsyncIterator[None]:
builder.provide(QDRANT_DB, qdrant.create_client(qdrant_url(), prefer_grpc=True))
yield

@coco.fn
async def app_main(sourcedir: pathlib.Path) -> None:
target_collection = await qdrant.mount_collection_target(
QDRANT_DB,
collection_name=QDRANT_COLLECTION,
schema=await qdrant.CollectionSchema.create(
vectors=qdrant.QdrantVectorDef(
schema=VectorSchema(dtype=np.dtype(np.float32), size=FACE_EMBED_DIM),
distance="euclid", # dlib face encodings compare by Euclidean distance
)
),
)

files = localfs.walk_dir(
    sourcedir,
    recursive=True,
    path_matcher=PatternFilePathMatcher(
        included_patterns=["**/*.jpg", "**/*.jpeg", "**/*.png"]
    ),
    live=True,  # api supports live watch; pass -L to `cocoindex update` to run live
)
await coco.mount_each(process_file, files.items(), target_collection)

app = coco.App(
coco.AppConfig(name="FaceRecognitionV1"),
app_main,
sourcedir=pathlib.Path("./images"),
)




- **`distance="euclid"`**: dlib's encodings are compared by Euclidean distance. `face_recognition` treats 2 faces at a distance of 0.6 or less as the same person by default (the `tolerance` of `compare_faces`).
- **`files.items()`** keys each file by its path relative to `images/`. That key names the file's component, so CocoIndex can tell which files are new, changed or gone.
- **`live=True`** lets the source keep watching the folder when the app runs in live mode (`cocoindex update -L main`). A plain `cocoindex update main` processes what changed and exits.

## Run it

From a clone of the [CocoIndex repo](https://github.com/cocoindex-io/cocoindex):



```sh
cd cocoindex/examples/face_recognition
docker run -d -p 6333:6333 -p 6334:6334 qdrant/qdrant
cp .env.example .env     # QDRANT_URL (defaults to the local container above)
pip install -e .
cocoindex update main    # or: cocoindex update -L main  (keep watching the folder)

The example ships 4 photos in images/. The first run finds 36 faces: 29 in the Solvay photo, 3 in Carter_welcomes_Reagan.jpg, and 2 each in the Steve Jobs and Bill Gates photo and in einplanck3.jpg. Each face is one point in the face_embeddings collection.

What an edit to the folder costs

Run the update again after a change, and CocoIndex runs only the photos that changed. Each button below is one edit followed by cocoindex update main:

What decides each outcome:

New photo. A new path mounts a new component with no memo entry, so extract_faces runs once and embed_face runs once per face.
Same bytes, new modification time. The memo state of a file is its modification time plus a fingerprint of its bytes. The time changed, so CocoIndex fingerprints the bytes, finds them unchanged, and skips the component. Its points stay.
New bytes. The component runs again: one detection, and one embedding for every face in the photo, including faces that did not change, because neither process_face nor embed_face is memoized. Cropping the left or top edge shifts every box, so every face gets a new ID; the new points are upserted and the old ones, no longer declared, are deleted.
Rename. The old path's component is gone, so its points are deleted. The new path has no memo entry and runs in full, and since the filename is part of the ID, its points get new IDs.
Delete. The component is removed, and CocoIndex deletes all of its points.

A change to the logic of process_file, or of the @coco.fn functions it calls (extract_faces, process_face, embed_face), invalidates the memo too, and the next update re-runs every photo. Comments and formatting do not count.

Search by face

The original version of this post sketched the goal: a face found in one photo, looked up in another.

main.py also has a small query command. It detects the faces in a query image, embeds the first one it finds with the same dlib encoder (inside the whole photo rather than as a crop), and asks Qdrant for the 5 nearest points:

```python title=main.py
def _qdrant_search(
client: QdrantClient, query_vector: list[float], limit: int
) -> list[Any]:
return client.query_points(
collection_name=QDRANT_COLLECTION,
query=query_vector,
limit=limit,
with_payload=True,
).points

def query(image_path: str, *, top_k: int = TOP_K) -> None:
content = pathlib.Path(image_path).read_bytes()
img = Image.open(io.BytesIO(content)).convert("RGB")
arr = np.array(img)
locs = face_recognition.face_locations(arr, model="cnn")
if not locs:
print("No face found in the query image.")
return
query_vec = face_recognition.face_encodings(arr, known_face_locations=locs[:1])[
0
].tolist()

client = qdrant.create_client(qdrant_url(), prefer_grpc=True)
for r in _qdrant_search(client, query_vec, top_k):
    payload = r.payload or {}
    # Euclidean distance: smaller is more similar (same person typically < 0.6).
    print(f"[{r.score:.3f}] {payload.get('filename')}  {payload}")






```sh
python main.py query images/einplanck3.jpg

einplanck3.jpg shows Max Planck and Albert Einstein, who both sit in the front row of the Solvay photo. The interactive demo on cocoindex.io runs the query for each of them against the 36 indexed faces.

The results show how to use this kind of index:

The photo's own crop comes first. einplanck3.jpg is in the index too, and its stored crop of the query face is 0.230 away for Planck and 0.187 for Einstein, not 0, because the query encodes the face inside the whole photo and the index encodes the tight crop.
Small, old faces crowd together. Most Solvay faces are about 80 pixels wide in a 1927 scan. Planck's face in the Solvay photo is 0.472 away, 0.009 behind a different man. Einstein's is 0.587 away, under 0.6, but 11 points are closer.
Ranked results are candidates. Show the top hits with their boxes and let a person confirm, or require a tighter distance before you treat a hit as a match.

To go further, search with every face in a photo instead of the first one, filter the hits by filename in the payload, or draw each hit's box on its photo.

The face recognition walkthrough in the docs covers the same example, and the image search post shows a query API and a live index inside a FastAPI server.

Build Real-Time Codebase Indexing for AI Coding agents

Linghua Jin — Sun, 13 Jul 2025 08:10:15 +0000

This post is also on cocoindex.io, with interactive figures. The code uses CocoIndex v1.

Build a live semantic index over a codebase with CocoIndex V1. Point it at a repo, and you get a vector index you can search in natural language ("where do we embed chunks?") that updates itself as you edit. It's the kind of fresh, low-latency code context a coding agent needs, and it's about 100 lines of plain async Python.

CocoIndex has built-in support for codebase chunking, with native Tree-sitter parsing and live updates.

The heavy lifting (incremental processing, change tracking, managed targets) runs in a Rust engine underneath. With a source that watches for changes, like the local filesystem in live mode, the index tracks your edits as they happen: save a file, and only the chunks that changed get re-embedded and re-upserted.

Use case

Code context for agents: semantic context for Claude, Codex, OpenCode, Factory.
Code search: natural-language and semantic search over your repo.
Review & refactor agents: context for code review, security analysis, and large-scale refactoring.

Why CocoIndex for codebase indexing

A codebase is hard to keep indexed well, and it exercises most of what CocoIndex was built for:

Chunks follow real code structure (functions, classes, blocks) instead of arbitrary line windows. Tree-sitter parsing is built in for every major language.
Code changes constantly: every commit, save, and rebase mutates a handful of files. CocoIndex re-embeds only the chunks that changed, and with live=True plus cocoindex update -L it keeps watching and applies each edit as it happens.
The engine is Rust. The run below indexes 622 files in 43 seconds cold; after that, a save costs one file.
The pipeline is ordinary async Python and your own types. Pick your embedding model, chunking strategy, and vector database; if your agent can write Python, it can extend this flow.

href="https://github.com/cocoindex-io/cocoindex/tree/main/examples/code_embedding"
path="examples/code_embedding"
description="The whole example is ~100 lines of plain Python. Clone it, point it at your repo, and swap in your favorite model or vector store."
/>

Overview

From a high level, these are the steps:

Read code files from a local directory.
Split each file into syntax-aware chunks with Tree-sitter, then embed every chunk.
Store the chunks and their embeddings in Postgres (as target states).

With CocoIndex, you declare the transformation logic with native Python, without worrying about handling updates.

Think: target_state = transformation(source_state).

When your codebase changes, or your processing logic changes (a different embedding model, a new chunk size), only the difference is reprocessed. The Incremental updates section walks through exactly what gets recomputed when.

Setup

In this example we store the indexed rows in Postgres with the pgvector extension. Your metadata and the vector search index both live there. CocoIndex supports many targets, so you can pick the one that works best for you.

If you don't have Postgres running, use the Docker Compose file in the cocoindex repo to start one.

   docker compose -f dev/postgres.yaml up -d

Point CocoIndex at it

   export POSTGRES_URL="postgres://cocoindex:cocoindex@localhost/cocoindex"

Install CocoIndex and the dependencies

   pip install -U "cocoindex[postgres,sentence_transformers]" asyncpg pgvector numpy python-dotenv

Define the App

Apps are the top-level runnable unit in CocoIndex. An app names your pipeline and binds a main function with its parameters.

Define the data and shared resources

This block sets up two things the rest of the code builds on. CodeEmbedding defines one row of the output table: each chunk of code becomes one row, with its text, location, and embedding vector. coco_lifespan provides the shared resources every step needs, the Postgres connection pool and the embedding model, built once at startup instead of per file.

import os
import pathlib
from dataclasses import dataclass
from typing import AsyncIterator, Annotated

import asyncpg
from numpy.typing import NDArray

import cocoindex as coco
from cocoindex.connectors import localfs, postgres
from cocoindex.ops.text import RecursiveSplitter, detect_code_language
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

DATABASE_URL = os.getenv("POSTGRES_URL", "postgres://cocoindex:cocoindex@localhost/cocoindex")
TABLE_NAME = "code_embeddings"
EMBED_MODEL = "sentence-transformers/all-MiniLM-L6-v2"

PG_DB = coco.ContextKey[asyncpg.Pool]("code_embedding_db")
EMBEDDER = coco.ContextKey[SentenceTransformerEmbedder]("embedder", detect_change=True)

_splitter = RecursiveSplitter()

@dataclass
class CodeEmbedding:
    id: int
    filename: str
    code: str
    embedding: Annotated[NDArray, EMBEDDER]
    start_line: int
    end_line: int

@coco.lifespan
async def coco_lifespan(builder: coco.EnvironmentBuilder) -> AsyncIterator[None]:
    async with asyncpg.create_pool(DATABASE_URL) as pool:
        builder.provide(PG_DB, pool)
        builder.provide(EMBEDDER, SentenceTransformerEmbedder(EMBED_MODEL))
        yield

Define file processing

With the row schema and shared resources in place, we can write the actual work: turning a file into rows.

Process file

@coco.fn(memo=True)
async def process_file(
    file: FileLike,
    table: postgres.TableTarget[CodeEmbedding],
) -> None:
    text = await file.read_text()
    language = detect_code_language(filename=str(file.file_path.path.name))
    chunks = _splitter.split(
        text,
        chunk_size=1000,
        min_chunk_size=300,
        chunk_overlap=300,
        language=language,
    )
    id_gen = IdGenerator()
    await coco.map(process_chunk, chunks, file.file_path.path, id_gen, table)

process_file runs once per file. It reads the file, detects the language so Tree-sitter can parse it, splits the code along the syntax tree, and maps each chunk to process_chunk.

Each chunk is a coherent syntactic unit, so retrieval returns whole functions or blocks, never a fragment cut mid-statement. All major languages are supported (Python, Rust, JavaScript, TypeScript, Java, C++, and more); unknown types fall back to plain text.

Note that @coco.fn with memo=True is what makes this incremental: if a file's content and this function's code are both unchanged, the whole file is skipped on the next run. coco.map fans out to one process_chunk call per chunk.

Here is what chunking produces: each file is split into syntactic chunks, each with its own location and text.

Process chunk

Next, we embed the chunk with the shared embedder and declare the target row.

@coco.fn
async def process_chunk(
    chunk: Chunk,
    filename: pathlib.PurePath,
    id_gen: IdGenerator,
    table: postgres.TableTarget[CodeEmbedding],
) -> None:
    embedding = await coco.use_context(EMBEDDER).embed(chunk.text)
    table.declare_row(
        row=CodeEmbedding(
            id=await id_gen.next_id(chunk.text),
            filename=str(filename),
            code=chunk.text,
            embedding=embedding,
            start_line=chunk.start.line,
            end_line=chunk.end.line,
        ),
    )

We use SentenceTransformerEmbedder with all-MiniLM-L6-v2; there are 12k+ sentence-transformer models on Hugging Face, so swap in whichever you prefer. Note chunk.start.line and chunk.end.line: V1 carries line numbers through, so search results point straight at the lines that matched.

Define the main function

app_main wires the source to the target. It mounts the Postgres table (with a vector index), walks the codebase, and mounts one processing component per file.

@coco.fn
async def app_main(sourcedir: pathlib.Path) -> None:
    target_table = await postgres.mount_table_target(
        PG_DB,
        table_name=TABLE_NAME,
        table_schema=await postgres.TableSchema.from_class(
            CodeEmbedding, primary_key=["id"],
        ),
    )
    target_table.declare_vector_index(column="embedding")

    files = localfs.walk_dir(
        sourcedir,
        recursive=True,
        path_matcher=PatternFilePathMatcher(
            included_patterns=["**/*.py", "**/*.rs", "**/*.toml", "**/*.md", "**/*.mdx"],
            excluded_patterns=["**/.*", "**/target", "**/node_modules"],
        ),
        live=True,  # watch for changes; pass -L to `cocoindex update` to run live
    )
    await coco.mount_each(process_file, files.items(), target_table)

We walk the codebase from sourcedir (point this at the repository you want to index). We index files with the extensions .py, .rs, .toml, .md, .mdx, and skip dot-directories, target, and node_modules.

mount_table_target creates and manages the Postgres table for you: schema, the pgvector index, idempotent upserts, and orphan cleanup when a file disappears. live=True makes the filesystem source watch for changes, and mount_each runs one component per file so the engine can track and update them independently.

Create the App

Bind app_main into a coco.App and point it at the codebase root.

app = coco.App(
    coco.AppConfig(name="CodeEmbeddingV1"),
    app_main,
    sourcedir=pathlib.Path(__file__).parent / ".." / "..",  # index from repo root
)

That is the entire indexing path.

Run the pipeline

Run the cocoindex CLI to set up and update the index. Choose catch-up (scan, sync, exit) or live (catch up, then keep watching):

# Catch-up run
cocoindex update main

# Live run: keep watching for file changes
cocoindex update -L main

You'll see the index update state in the terminal:

✅ app_main: 1 total | 1 added
✅ declare_target_state_with_child: 1 total | 1 added
✅ _MountEachLiveComponent.process: 1 total | 1 added
✅ process_file: 622 total | 622 added
⏳ Elapsed: 43.4s

Query the index

We match user text against the index with a plain SQL query, reusing the same embedder from the indexing flow so indexing and querying stay consistent.

async def query_once(pool, embedder, query: str, *, top_k: int = 5) -> None:
    query_vec = await embedder.embed(query)
    async with pool.acquire() as conn:
        rows = await conn.fetch(
            f"""
            SELECT filename, code, embedding <=> $1 AS distance, start_line, end_line
            FROM "{TABLE_NAME}"
            ORDER BY distance ASC
            LIMIT $2
            """,
            query_vec, top_k,
        )
    for r in rows:
        score = 1.0 - float(r["distance"])
        print(f"[{score:.3f}] {r['filename']} (L{r['start_line']}-L{r['end_line']})")
        print(f"    {r['code']}")
        print("---")

The <=> operator is pgvector's cosine distance. We turn it into a similarity score and print the filename, the matched line range, and the code snippet. Run it against your index:

python main.py "where do we embed chunks"

It prints the top five chunks, each with its similarity score, file, and line range. Snippets are cut to their first lines here (…):

[0.527] rust/ops_text/src/split/by_separators.rs (L72-L94)
            let bytes = text.as_bytes();

        // Collect raw chunks (byte ranges)
        struct RawChunk {
            start: usize,
            end: usize,
    …
---
[0.459] rust/ops_text/src/split/by_separators.rs (L116-L138)
            let mut positions: Vec<Position> = raw_chunks
            .iter()
            .flat_map(|c| vec![Position::new(c.start), Position::new(c.end)])
            .collect();

        set_output_positions(text, positions.iter_mut());
    …
---
[0.448] rust/ops_text/src/split/recursive.rs (L560-L578)
        fn split_root_chunk(&self, kind: ChunkKind<'t>) -> Vec<ChunkOutput> {
        let mut atom_collector = AtomChunksCollector {
            full_text: self.full_text,
            min_level: 0,
            curr_level: 0,
            atom_chunks: Vec::new(),
    …
---
[0.447] rust/sdk/cocoindex/tests/code.rs (L300-L314)
    #[test]
fn code_ast_split_chunk_positions_are_accurate() {
    // Three-line Python source; verify start/end positions of first chunk.
    let src = "a = 1\nb = 2\nc = 3\n";
    let ast = CodeAst::new(src, "python").unwrap();
    let chunks = ast.split(200, None, None).unwrap(); // one big chunk
    …
---
[0.444] rust/sdk/cocoindex/src/ops/code.rs (L315-L341)
            let config = RecursiveChunkConfig {
            chunk_size,
            min_chunk_size,
            chunk_overlap,
        };
        let raw = chunker.split(&self.source, config);
    …
---

Incremental updates

You never compute a diff or write update logic: you change something, and CocoIndex works out what to embed, upsert, and delete. @coco.fn(memo=True) decides what to recompute: a file is skipped when its content and the function's code are both unchanged, and an embedding is reused when its chunk text is unchanged. mount_table_target decides what to write: each row's id is derived from its chunk's content, so only rows that actually changed are upserted, and rows whose source is gone are deleted.

The same machinery covers two kinds of change: changes to your data (the code being indexed) and changes to your logic (the pipeline itself).

Data changes.

A file is added: only that file is chunked and embedded, and its rows are inserted. The rest of the repo is untouched.
A file is deleted: its rows are removed from the target.
A file is edited: the file is re-chunked, and the new chunks usually overlap the old ones. Chunks whose text is unchanged keep their id and embedding, so they are left as-is; genuinely new chunks are embedded and inserted; chunks that no longer exist are deleted. Edit one function and only that function's chunks are re-embedded, even though the whole file was re-read.

Logic changes. A pipeline change is reconciled the same way: CocoIndex compares the new output against what is already in Postgres and applies only the difference.

Change the file patterns (included_patterns / excluded_patterns): files that now match are added automatically; files that no longer match have their rows deleted automatically.
Tune the chunking (chunk size, overlap): files are re-chunked, producing the same partial-overlap diff shown above: unchanged chunks are no-ops, new chunks are embedded and inserted, dropped chunks are deleted.
Swap the embedding model: the vectors genuinely change, so all embeddings are recomputed, but row identity is stable: it is an in-place update of the embedding column, with no rows added or removed.

A catch-up run (cocoindex update main) does this once and exits; live mode (cocoindex update -L main) keeps watching and applies each change as you code.

Next steps

The full, runnable example is in the CocoIndex repo: examples/code_embedding, with a walkthrough in the index-codebase example.

Point sourcedir at your own repository.
Swap the embedding model or the target store; the pipeline is plain Python.
Read the CocoIndex V1 docs for more on incremental processing, components, and targets.

Issues and contributions are welcome on GitHub.

Beyond Embeddings: Building Metadata-Rich Indexes from Academic PDFs

Linghua Jin — Fri, 11 Jul 2025 06:20:41 +0000

This post is also on cocoindex.io, with interactive figures. The code uses CocoIndex v1.

The first page of a paper holds almost everything you'd want to query: title, authors, abstract. But it's locked in PDF prose. This walkthrough reads just that page, hands the text to an LLM with a strict schema, gets back clean typed JSON, and embeds the metadata so you can search papers by meaning. One PDF fans out into three Postgres tables, and CocoIndex keeps all three in sync as the folder changes.

The full code is in the CocoIndex v1 repo.

What we'll build

Each PDF flows through one processing component: slice out the first page, pull its text, extract metadata with the LLM, and embed the title and abstract. The component declares rows into three tables.

The three tables serve three kinds of queries. paper_metadata holds one row per paper for lookups and joins. author_papers maps each author name to a filename, so "all the papers by Jeff Dean" is a plain SQL WHERE. metadata_embeddings holds one vector per title and per abstract chunk, for semantic search with pgvector.

Define the models and rows

Two pydantic models describe what the LLM must return. Validation is the contract: a malformed response fails loudly instead of writing junk.

```python title=models.py
from pydantic import BaseModel, Field

class AuthorModel(BaseModel):
name: str
email: str | None = None
affiliation: str | None = None

class PaperMetadataModel(BaseModel):
title: str
authors: list[AuthorModel] = Field(default_factory=list)
abstract: str




The pydantic models are the LLM contract only. What lands in Postgres is described separately, by three dataclasses, one per table. Keeping the two layers apart lets the table shape differ from the LLM's: rows add `filename` and `num_pages`, and authors are stored as plain dicts. The `embedding` field is annotated with the embedder so CocoIndex infers the vector size:



```python title=main.py
import uuid
from dataclasses import dataclass
from typing import Annotated
from numpy.typing import NDArray

import cocoindex as coco
from cocoindex.ops.sentence_transformers import SentenceTransformerEmbedder

EMBED_MODEL = "sentence-transformers/all-MiniLM-L6-v2"
EMBEDDER = coco.ContextKey[SentenceTransformerEmbedder]("embedder", detect_change=True)

@dataclass
class PaperMetadataRow:
    filename: str
    title: str
    authors: list[dict[str, str | None]]
    abstract: str
    num_pages: int

@dataclass
class AuthorPaperRow:
    author_name: str
    filename: str

@dataclass
class MetadataEmbeddingRow:
    id: uuid.UUID
    filename: str
    location: str
    text: str
    embedding: Annotated[NDArray, EMBEDDER]

The database pool and the embedder are created once, in the app lifespan:

```python title=main.py
import asyncpg
from cocoindex.connectors import postgres

PG_DB = coco.ContextKeyasyncpg.Pool

@coco.lifespan
async def coco_lifespan(builder: coco.EnvironmentBuilder) -> AsyncIterator[None]:
async with asyncpg.create_pool(os.getenv("POSTGRES_URL")) as pool:
builder.provide(PG_DB, pool)
builder.provide(EMBEDDER, SentenceTransformerEmbedder(EMBED_MODEL))
yield




## Read the first page

A paper can be large. The Attention Is All You Need PDF in the sample folder is 2,215,244 bytes, and other arXiv PDFs run to dozens of pages. The metadata we want lives on page one, so we slice it out with pypdf and never send the rest anywhere:



```python title=main.py
import io
from pypdf import PdfReader, PdfWriter

@dataclass
class PaperBasicInfo:
    num_pages: int
    first_page: bytes

@coco.fn
def extract_basic_info(content: bytes) -> PaperBasicInfo:
    """Extract first page bytes and page count from a PDF."""
    reader = PdfReader(io.BytesIO(content))

    output = io.BytesIO()
    writer = PdfWriter()
    writer.add_page(reader.pages[0])
    writer.write(output)

    return PaperBasicInfo(num_pages=len(reader.pages), first_page=output.getvalue())

A second small function pulls the text off that single page:

```python title=main.py
@coco.fn
def pdf_to_markdown(content: bytes) -> str:
"""Convert PDF bytes to text using pypdf."""
reader = PdfReader(io.BytesIO(content))
page_text = reader.pages[0].extract_text() if reader.pages else ""
return page_text or ""




pypdf keeps the example dependency-light. If your PDFs are heavier on layout, swap this function's body for [Marker](https://github.com/datalab-to/marker) or [Docling](https://github.com/docling-project/docling); the rest of the pipeline does not change.

## Extract metadata with an LLM

One call turns the first-page prose into typed fields. The system prompt pins the schema, `response_format` forces JSON, `temperature=0` keeps it deterministic, and `model_validate_json` parses the response into the pydantic model:



```python title=main.py
import functools
from openai import OpenAI

LLM_MODEL = "gpt-4o"

@functools.cache
def openai_client() -> OpenAI:
    return OpenAI()

@coco.fn
def extract_metadata(markdown: str) -> PaperMetadataModel:
    """Extract paper metadata from first-page text using an LLM."""
    client = openai_client()
    response = client.chat.completions.create(
        model=LLM_MODEL,
        messages=[
            {
                "role": "system",
                "content": (
                    "You extract metadata from academic paper first pages. "
                    "Return only JSON with keys: title, authors, abstract. "
                    "authors is a list of {name, email, affiliation}. "
                    "Use null for missing fields."
                ),
            },
            {
                "role": "user",
                "content": markdown[:4000],
            },
        ],
        response_format={"type": "json_object"},
        temperature=0,
    )

    content = response.choices[0].message.content
    if not content:
        raise RuntimeError("LLM returned empty content.")
    return PaperMetadataModel.model_validate_json(content)

Process one paper into three tables

process_file is the whole per-paper pipeline: read, slice, extract, embed, declare. memo=True caches it, so an unchanged PDF is skipped on the next run:

```python title=main.py
from cocoindex.ops.text import CustomLanguageConfig, RecursiveSplitter
from cocoindex.resources.file import FileLike

_abstract_splitter = RecursiveSplitter(
custom_languages=[
CustomLanguageConfig(
language_name="abstract",
separators_regex=[r"[.?!]+\s+", r"[:;]\s+", r",\s+", r"\s+"],
)
]
)

@coco.fn(memo=True)
async def process_file(
file: FileLike,
metadata_table: postgres.TableTarget[PaperMetadataRow],
author_table: postgres.TableTarget[AuthorPaperRow],
embedding_table: postgres.TableTarget[MetadataEmbeddingRow],
) -> None:
content = await file.read()

basic_info = extract_basic_info(content)
first_page_md = pdf_to_markdown(basic_info.first_page)
metadata = extract_metadata(first_page_md)

metadata_table.declare_row(
    row=PaperMetadataRow(
        filename=str(file.file_path.path),
        title=metadata.title,
        authors=[a.model_dump() for a in metadata.authors],
        abstract=metadata.abstract,
        num_pages=basic_info.num_pages,
    ),
)

for author in metadata.authors:
    if author.name:
        author_table.declare_row(
            row=AuthorPaperRow(
                author_name=author.name,
                filename=str(file.file_path.path),
            ),
        )

title_embedding = await coco.use_context(EMBEDDER).embed(metadata.title)
embedding_table.declare_row(
    row=MetadataEmbeddingRow(
        id=uuid.uuid4(),
        filename=str(file.file_path.path),
        location="title",
        text=metadata.title,
        embedding=title_embedding,
    ),
)

abstract_chunks = _abstract_splitter.split(
    metadata.abstract,
    chunk_size=500,
    min_chunk_size=200,
    chunk_overlap=150,
    language="abstract",
)
for chunk in abstract_chunks:
    embedding_table.declare_row(
        row=MetadataEmbeddingRow(
            id=uuid.uuid4(),
            filename=str(file.file_path.path),
            location="abstract",
            text=chunk.text,
            embedding=await coco.use_context(EMBEDDER).embed(chunk.text),
        ),
    )




Run the numbers for one concrete paper. Attention Is All You Need has 8 authors, and its 1,139-character abstract splits into 3 chunks with these settings (sentence-aware separators, chunk size 500, overlap 150). So the function declares 13 rows: 1 metadata row, 8 author rows, and 4 embedding rows (the title plus the 3 chunks):

The abstract is chunked because a long abstract blurs into a single vector; per-chunk embeddings keep search sharp. And everything is declarative: remove the PDF from the folder and all 13 rows across the three tables are cleaned up on the next run.

## Wire up the app

`app_main` mounts the three target tables, walks the folder of PDFs, and runs one `process_file` per file:



```python title=main.py
import pathlib
from cocoindex.connectors import localfs
from cocoindex.resources.file import PatternFilePathMatcher

PG_SCHEMA_NAME = "coco_examples_v1"

@coco.fn
async def app_main(sourcedir: pathlib.Path) -> None:
    metadata_table = await postgres.mount_table_target(
        PG_DB,
        table_name="paper_metadata",
        table_schema=await postgres.TableSchema.from_class(
            PaperMetadataRow,
            primary_key=["filename"],
        ),
        pg_schema_name=PG_SCHEMA_NAME,
    )
    author_table = await postgres.mount_table_target(
        PG_DB,
        table_name="author_papers",
        table_schema=await postgres.TableSchema.from_class(
            AuthorPaperRow,
            primary_key=["author_name", "filename"],
        ),
        pg_schema_name=PG_SCHEMA_NAME,
    )
    embedding_table = await postgres.mount_table_target(
        PG_DB,
        table_name="metadata_embeddings",
        table_schema=await postgres.TableSchema.from_class(
            MetadataEmbeddingRow,
            primary_key=["id"],
        ),
        pg_schema_name=PG_SCHEMA_NAME,
    )

    files = localfs.walk_dir(
        sourcedir,
        recursive=True,
        path_matcher=PatternFilePathMatcher(included_patterns=["**/*.pdf"]),
        live=True,
    )
    await coco.mount_each(
        process_file, files.items(), metadata_table, author_table, embedding_table
    )

app = coco.App(
    coco.AppConfig(name="PaperMetadataV1"),
    app_main,
    sourcedir=pathlib.Path("./papers"),
)

mount_table_target creates and manages each table: schema, idempotent upserts, and removal of rows whose source vanished. walk_dir(..., live=True) means the source can watch the folder; drop a new PDF in and it gets indexed without re-running anything.

Run it

Start Postgres with pgvector (the repo ships a compose file), set POSTGRES_URL and OPENAI_API_KEY, and build the index. The example folder ships sample papers, including Attention Is All You Need, so it works out of the box:

docker compose -f ../../dev/postgres.yaml up -d
pip install -e .
cocoindex update main

The first run extracts and embeds everything; re-runs only process changed files. Add -L (cocoindex update -L main) to watch the folder and update the index live.

Query the index

Semantic search embeds the query with the same model and asks Postgres for the nearest vectors; <=> is pgvector's cosine distance:

```python title=main.py
async def query_once(pool, embedder, query: str, *, top_k: int = 5) -> None:
query_vec = await embedder.embed(query)
async with pool.acquire() as conn:
rows = await conn.fetch(
"""
SELECT filename, location, text, embedding <=> $1 AS distance
FROM "coco_examples_v1"."metadata_embeddings"
ORDER BY distance ASC
LIMIT $2
""",
query_vec,
top_k,
)
for r in rows:
score = 1.0 - float(r["distance"])
print(f"[{score:.3f}] {r['filename']} ({r['location']})")
print(f" {r['text']}")






```sh
python main.py "sequence models without recurrence"

Matches come back from both titles and abstract chunks, with the location column telling you which. The author index needs no embeddings at all:

SELECT filename FROM "coco_examples_v1"."author_papers"
WHERE author_name = 'Ashish Vaswani';

Where to go next

The shape here generalizes past papers: any document family with a metadata-rich region (contracts, invoices, intake forms) can use the same slice-extract-validate-embed pipeline. Swap the source for Amazon S3 or Google Drive, swap the target for Qdrant, or feed the metadata tables to an agent. For full-text embedding of the whole PDF body, see text embeddings 101.

Build Real-Time Knowledge Graphs from Documents Using CocoIndex + Kuzu (with LLMs & Live Updates)

Linghua Jin — Wed, 04 Jun 2025 05:01:25 +0000

A blazing-fast, end-to-end open source pipeline for turning documents into queryable knowledge graphs using LLMs, CocoIndex, and the Kuzu graph database.

🔍 Why Real-Time Knowledge Graphs?
If you're working with unstructured documents and want to make the data inside searchable, relational, and up-to-date — this guide is for you. With just ~200 lines of Python, we’ll show you how to:

Extract semantic relationships using an LLM
Stream structured graph data into Kuzu
Build a real-time, self-updating knowledge base

What You’ll Use

✅ CocoIndex
https://github.com/cocoindex-io/cocoindex
A high-performance data transformation engine using a dataflow programming model. It enables incremental processing, modular plugins, and live updates for knowledge graphs.

✅ Kuzu
https://github.com/kuzudb/kuzu
An open-source, fast, and lightweight graph database designed for high performance. It’s easy to run locally or in production, and now natively supported in CocoIndex.

✅ Large Language Models (LLMs)
Used for extracting subject-object-predicate relationships from document content.

We understand preparing data is highly use-case based and there is no one-size-fits-all solution.

We take the composition approach, and instead of building everything, we provide native plugins to embrace the ecosystem and make it easier to plug in and swap any module by standardizing the interface - exactly like LEGO.

If you are using CocoIndex to build your knowledge graph, you can use Kuzu as a target graph data store.

How to map to Kuzu in CocoIndex

The GraphDB interface in CocoIndex is standardized, if you are already using Neo4j, you just need to switch the configuration to export to Kuzu as below.
CocoIndex supports exporting to Kuzu through its API server. You can bring up a Kuzu API server locally by running:

KUZU_DB_DIR=$HOME/.kuzudb
KUZU_PORT=8123
docker run -d --name kuzu -p ${KUZU_PORT}:8000 -v ${KUZU_DB_DIR}:/database kuzudb/api-server:latest

In your CocoIndex flow, you need to add the Kuzu connection spec to your flow.

kuzu_conn_spec = cocoindex.add_auth_entry(
    "KuzuConnection",
    cocoindex.storages.KuzuConnection(
        api_server_url="http://localhost:8123",
    ),
)

What does it look like to build an indexing flow with CocoIndex + Kuzu

A CocoIndex knowledge graph example that got the most love is to build knowledge graph with LLM, here is a detailed step-by-step blog.
In the project, we process a list of documents,and use LLM to extract relationships between the concepts in each document.

We will generate two kinds of relationships from the documents:

Relationships between subjects and objects. E.g., "CocoIndex supports Incremental Processing"
Mentions of entities in a document. E.g., "core/basics.mdx" mentions CocoIndex and Incremental Processing.

The indexing flow looks like this for Kuzu:

The code is available here.

Ingest the documents into CocoIndex
Process the documents, for each document:
1. Map document nodes: Use LLM to generate summary, and map the documents to Graph nodes in Kuzu.
2. Map relationship nodes: Use LLM to extract relationships, and export the relationships to Kuzu.

Notably, it only takes ~200 lines of python to have a production ready knowledge graph; including class definitions, prompts, and configs.

To highlight how the relationship extraction works, you will define a python class for structured extraction.

@dataclasses.dataclass
class Relationship:
    """
    Describe a relationship between two entities.
    Subject and object should be Core CocoIndex concepts only, should be nouns. For example, `CocoIndex`, `Incremental Processing`, `ETL`,  `Data` etc.
    """

    subject: str
    predicate: str
    object: str

If you have a predefined set of ontology, you can skip the entity extraction use existing entities.

Call a transformation in the flow to extract the relationships from the document.

with data_scope["documents"].row() as doc:

    # extract relationships from document
    doc["relationships"] = doc["content"].transform(
        cocoindex.functions.ExtractByLlm(
            llm_spec=cocoindex.LlmSpec(
                # Supported LLM: https://cocoindex.io/docs/ai/llm
                api_type=cocoindex.LlmApiType.OPENAI,
                model="gpt-4o",
            ),
            output_type=list[Relationship],
            instruction=(
                "Please extract relationships from CocoIndex documents. "
                "Focus on concepts and ignore examples and code. "
            ),
        )
    )

You could use CocoInsight to verify each pair of the relationships.

and then collect the relationship use entity_relationship collector.

with doc["relationships"].row() as relationship:
    # relationship between two entities
    entity_relationship.collect(
        id=cocoindex.GeneratedField.UUID,
        subject=relationship["subject"],
        object=relationship["object"],
        predicate=relationship["predicate"],
    )

CocoIndex follows a dataflow programming model.
Rather than defining data operations like creations, updates or deletions, developers only need to focus on transformations or formulas based on source data.
The framework takes care of the data operations such as when to create, update, or delete.

Once you have collected the relationships, you can directly map it to Kuzu as below.

    entity_relationship.export(
        "entity_relationship",
         cocoindex.storages.Kuzu(
            connection=conn_spec,
            mapping=cocoindex.storages.Relationships(
                rel_type="RELATIONSHIP",
                source=cocoindex.storages.NodeFromFields(
                    label="Entity",
                    fields=[
                        cocoindex.storages.TargetFieldMapping(
                            source="subject", target="value"
                        ),
                    ],
                ),
                target=cocoindex.storages.NodeFromFields(
                    label="Entity",
                    fields=[
                        cocoindex.storages.TargetFieldMapping(
                            source="object", target="value"
                        ),
                    ],
                ),
            ),
        ),
        primary_key_fields=["id"],
    )

Amazingly, while working on this Kuzu example, I had a previous flow that I ran locally with Neo4j.
It was instant to export to Kuzu. CocoIndex is based on incremental processing, and if you have already run this flow before and just switched targets, the intermediate transformation results can be reused.

To run the Kuzu Explorer - an open source UI for Kuzu, you need to first bring down the Kuzu API server.

And then you can run the following command to start the Kuzu Explorer:

KUZU_EXPLORER_PORT=8124
docker run -d --name kuzu-explorer -p ${KUZU_EXPLORER_PORT}:8000  -v ${KUZU_DB_DIR}:/database -e MODE=READ_ONLY  kuzudb/explorer:latest

We could then access the explorer at http://localhost:8124. We could run a Cypher query to explore the graph.

MATCH p=()-->() RETURN p

Support us

We are constantly improving, and more features and examples are coming soon. If this article is helpful, please drop us a star ⭐ at GitHub to help us grow.

Thanks for reading!

Stream from S3 in Real-Time: CocoIndex Brings True Incremental Processing to the Cloud

Linghua Jin — Fri, 30 May 2025 17:46:31 +0000

This post is also on cocoindex.io, with interactive figures. The code uses CocoIndex v1.

CocoIndex has native support for Amazon S3 as a data source. The pipeline reads Markdown files from an S3 bucket, chunks and embeds them, and writes the vectors to a Postgres pgvector table for semantic search.

The point of the example is incremental processing: adding one object to the bucket embeds one object, not the whole bucket. It is Semantic Search 101 with the source swapped from a local folder to S3; everything downstream is the same.

The full project is in the CocoIndex v1 repo.

Why incremental processing

Incremental processing processes only the data that changed since the last run, instead of reprocessing everything. It matters most in three situations.

High freshness requirements. For user-facing search, stale results are a real problem: when someone updates a document, they expect it reflected in search right away. Feed a stale result into an AI agent and the model answers from out-of-date information, often without anyone noticing.

Expensive transforms. When each item goes through an embedding model or an LLM, the transform costs far more than the retrieval. Recomputing unchanged items burns money for no benefit.

Large datasets. With terabytes in S3, reprocessing everything on each new file is impractical on time, cost, and compute. Incremental processing keeps the work proportional to the change, not the corpus.

A useful way to think about it: if T is the staleness you can tolerate, and you do not want to recompute the whole corpus every T, you need incremental processing.

How CocoIndex processes only what changed

CocoIndex tracks each unit of work by a fingerprint of its input and the transform code. On the next run, anything whose fingerprint is unchanged is skipped, and its downstream rows are left untouched.

The granularity is finer than the file. If a Markdown file has many chunks and you edit one paragraph, only the chunks that actually changed are re-embedded; the rest reuse their cached vectors:

That is where the savings come from. The expensive embedding calls run only for content CocoIndex has not seen, and the vector store ends up consistent either way.

The indexing flow

The flow reads Markdown files from S3, and for each file splits it into chunks, embeds each chunk, and declares a row in the vector table:

Each file is handled by its own processing component, so change detection and parallelism are per file.

Define the pipeline

Shared resources (the Postgres pool, the S3 client, and the embedder) are provided once in the app lifespan and read anywhere with coco.use_context:

```python title=main.py
import cocoindex as coco
from cocoindex.connectors import amazon_s3, postgres
from cocoindex.ops.sentence_transformers import SentenceTransformerEmbedder

EMBED_MODEL = "sentence-transformers/all-MiniLM-L6-v2"

PG_DB = coco.ContextKeyasyncpg.Pool
S3_CLIENT = coco.ContextKeyAioBaseClient
EMBEDDER = coco.ContextKeySentenceTransformerEmbedder

@coco.lifespan
async def coco_lifespan(builder: coco.EnvironmentBuilder) -> AsyncIterator[None]:
async with asyncpg.create_pool(DATABASE_URL) as pool:
builder.provide(PG_DB, pool)
builder.provide(EMBEDDER, SentenceTransformerEmbedder(EMBED_MODEL))

    session = aiobotocore.session.get_session()
    async with session.create_client("s3") as s3_client:
        builder.provide(S3_CLIENT, s3_client)
        yield




`detect_change=True` on the embedder means swapping the model re-runs only the embedding step, not the whole pipeline.

The target row is a plain dataclass. The `embedding` column is annotated with the embedder so CocoIndex knows the vector dimension:



```python title=main.py
@dataclass
class DocEmbedding:
    id: int
    filename: str
    chunk_start: int
    chunk_end: int
    text: str
    embedding: Annotated[NDArray, EMBEDDER]

Chunk, embed, declare

Each chunk becomes one row. process_chunk embeds the chunk text and declares the row; CocoIndex works out whether that is an insert or an update:

```python title=main.py
@coco.fn
async def process_chunk(
chunk: Chunk,
filename: str,
id_gen: IdGenerator,
table: postgres.TableTarget[DocEmbedding],
) -> None:
table.declare_row(
row=DocEmbedding(
id=await id_gen.next_id(chunk.text),
filename=filename,
chunk_start=chunk.start.char_offset,
chunk_end=chunk.end.char_offset,
text=chunk.text,
embedding=await coco.use_context(EMBEDDER).embed(chunk.text),
),
)




`process_file` reads one S3 file, splits it into chunks, and maps `process_chunk` over them. `memo=True` caches per-file work, so an unchanged file is skipped entirely:



```python title=main.py
_splitter = RecursiveSplitter()

@coco.fn(memo=True)
async def process_file(
    file: amazon_s3.S3File,
    table: postgres.TableTarget[DocEmbedding],
) -> None:
    text = await file.read_text()
    chunks = _splitter.split(
        text, chunk_size=2000, chunk_overlap=500, language="markdown"
    )
    id_gen = IdGenerator()
    await coco.map(process_chunk, chunks, file.file_path.path.as_posix(), id_gen, table)

Wire up the app

app_main mounts the pgvector table target, lists the Markdown objects in the bucket, and fans out one process_file component per file with mount_each:

```python title=main.py
@coco.fn
async def app_main() -> None:
target_table = await postgres.mount_table_target(
PG_DB,
table_name=TABLE_NAME,
table_schema=await postgres.TableSchema.from_class(
DocEmbedding, primary_key=["id"]
),
pg_schema_name=PG_SCHEMA_NAME,
)

client = coco.use_context(S3_CLIENT)
files = amazon_s3.list_objects(
    client,
    S3_BUCKET,
    prefix=S3_PREFIX,
    path_matcher=PatternFilePathMatcher(included_patterns=["**/*.md"]),
)
await coco.mount_each(process_file, files.items(), target_table)

app = coco.App(
coco.AppConfig(name="AmazonS3Embedding"),
app_main,
)




`amazon_s3.list_objects` returns the objects under the prefix; `files.items()` yields `(key, file)` pairs, and `mount_each` gives each file a stable component path keyed on its S3 path.

## Set up the S3 bucket

Follow the [Amazon S3 connector docs](https://cocoindex.io/docs/connectors/amazon_s3/) to create a bucket and grant read access. The pipeline needs read-only S3 permissions (`AmazonS3ReadOnlyAccess` is enough), and standard AWS credentials in the environment. Point the pipeline at the bucket with a `.env` file:



```sh
POSTGRES_URL=postgresql://cocoindex:cocoindex@localhost/cocoindex
S3_BUCKET=your-bucket-name
S3_PREFIX=

Then upload your Markdown files to the bucket.

Run the pipeline

Build the index. This is a one-shot catch-up over the objects in the bucket:

cocoindex update main

The first run embeds everything. Every run after that does only the new work: new and changed files are processed, unchanged files are skipped, and within a changed file only the changed chunks are re-embedded.

For fresh data, schedule the update on whatever cadence you need (cron, Airflow, GitHub Actions, an Azure or Lambda function). Because re-runs only touch what changed, running the pipeline often is cheap. If you need event-driven, second-level freshness instead of scheduled runs, use one of CocoIndex's streaming sources (Kafka, Iggy, or Valkey) as the source of your pipeline.

Once the index is built, query it:

python main.py "your query"

How to build image search with semantic understanding

Linghua Jin — Sun, 25 May 2025 23:27:45 +0000

This post is also on cocoindex.io, with interactive figures. The code uses CocoIndex v1.

Search a folder of photos for "long neck" and a filename search returns nothing, unless someone named the giraffe photo that way.

This post builds image search that matches by meaning instead: you type "long neck" and the giraffe comes back, without captions or manual labels. Try it on the 4 photos that ship with the example. The scores are real CLIP scores.

The model that makes this work is CLIP. It embeds an image and a piece of text into the same vector space, so a text query and a matching picture land near each other. CocoIndex v1 walks a local image folder, embeds each picture with CLIP, and stores the vectors in Qdrant. The index runs in live mode inside a FastAPI server, so a photo you drop into the folder becomes searchable as soon as it is embedded, with no rebuild step.

The full code is in the CocoIndex repo and runs on CocoIndex v1.

Why search photos by meaning

You write the transformation in plain Python, target_state = transformation(source_state). CocoIndex handles the rest: the file watcher, change tracking, memoization, and a managed Qdrant collection, with change tracking and memoization running in its Rust engine.

The flow

Each image is its own processing component: read the bytes, embed them with the CLIP image encoder, and declare a Qdrant point. There is no text to chunk, so the indexing path is one embedding per image.

The shared image-text space

CLIP is trained to place a picture and its description near each other in one embedding space. At index time we only ever embed pixels (get_image_features); at query time we embed text (get_text_features). Both produce 768 floats in the same space, so "long neck" ends up closest to the giraffe photo even though that image was never described in words.

The vector database never learns that one vector came from an image and the other from text. It returns nearest neighbors by cosine similarity. A new image therefore needs no description to become searchable; adding the file to the folder is enough.

CLIP's image-to-text scores sit in a narrow band. In the post's demo, the best match for each query scores 0.21 to 0.23 and the other photos 0.11 to 0.19. Search ranks by those scores, so the order matters more than the absolute value.

One image, end to end

The stepper below follows a single photo through process_file before the code sections. Use the buttons, or let it play.

The rest of the post builds this pipeline in 3 pieces: the shared resources, the per-image function, and the app that connects them.

Technologies

CLIP ViT-L/14

CLIP ViT-L/14 is a vision-language model that embeds images and text into a shared 768-dimensional space. We use it two ways: to embed each image at index time, and to convert a natural-language query into that same space at search time.

The collection's vector size comes from model.config.projection_dim, so a smaller variant like CLIP ViT-B/32 (faster, slightly lower accuracy) needs only a new CLIP_MODEL_NAME, and the collection is created to match. A swap that changes the size (ViT-B/32 has 512 dimensions, against 768) makes CocoIndex drop and recreate the collection and re-embed every image on the next start. A swap that keeps the size is not noticed: the memo of process_file tracks the function's code, not the CLIP_MODEL_NAME constant. For that case, declare the dependency with @coco.fn(memo=True, deps=CLIP_MODEL_NAME), or stop the server and run cocoindex update pipeline.py --full-reprocess once. See memoization for what invalidates a memo.

CocoIndex

CocoIndex is an open-source data transformation framework for AI, with incremental processing built in. You declare the transformation in plain Python and your own types; change tracking and memoization run in a Rust engine underneath.

Qdrant

Qdrant is a vector database. We store the image vectors as points and query them by cosine distance through the CocoIndex Qdrant connector.

FastAPI

FastAPI serves the search API and hosts the live index in its lifespan, so one process both keeps the collection current and answers queries.

Prerequisites

Install Qdrant: docker run -d -p 6333:6333 -p 6334:6334 qdrant/qdrant.
Python 3.11+ with torch, transformers, and pillow for CLIP. pip install -e . in the example folder installs them.

CocoIndex v1 keeps its own processing state in a local LMDB directory, set by COCOINDEX_DB (the example's .env sets it to ./cocoindex.db), so there is no separate tracking database to run.

Define the pipeline

The pipeline lives in pipeline.py. The excerpts below leave out the standard-library, torch, transformers, Pillow, NumPy, and qdrant_client imports; the linked file has them all. CocoIndex's Qdrant client is created in the CocoIndex lifespan and reached through the QDRANT_DB context key (api.py opens its own client for search):

```python title=pipeline.py
import cocoindex as coco
from cocoindex.connectors import localfs, qdrant
from cocoindex.resources.file import FileLike, PatternFilePathMatcher
from cocoindex.resources.schema import VectorSchema

QDRANT_COLLECTION = "ImageSearch"
CLIP_MODEL_NAME = "openai/clip-vit-large-patch14"
TOP_K = 5

def qdrant_url() -> str:
return os.getenv("QDRANT_URL", "http://localhost:6334/")

QDRANT_DB = coco.ContextKeyQdrantClient
QDRANT_CLIENT = coco.ContextKeyQdrantClient






```python title=pipeline.py
@coco.lifespan
async def coco_lifespan(
    builder: coco.EnvironmentBuilder,
) -> AsyncIterator[None]:
    client = qdrant.create_client(qdrant_url(), prefer_grpc=True)
    builder.provide(QDRANT_DB, client)
    builder.provide(QDRANT_CLIENT, client)
    yield

CLIP is loaded once and reused. The image encoder is an ordinary helper, so nothing about it is CocoIndex-specific. _projected_features handles a return-type change between transformers 4 and 5:

```python title=pipeline.py
@functools.cache
def get_clip_model() -> tuple[CLIPModel, CLIPProcessor]:
model = CLIPModel.from_pretrained(CLIP_MODEL_NAME)
processor = CLIPProcessor.from_pretrained(CLIP_MODEL_NAME)
return model, processor

def _projected_features(out: Any) -> torch.Tensor:
# transformers >=5 returns BaseModelOutputWithPooling with the projected features in pooler_output;
# transformers <5 returns the projected features tensor directly.
return out.pooler_output if hasattr(out, "pooler_output") else out






```python title=pipeline.py
def embed_image_bytes(img_bytes: bytes) -> list[float]:
    model, processor = get_clip_model()
    image = Image.open(io.BytesIO(img_bytes)).convert("RGB")
    inputs = processor(images=image, return_tensors="pt")
    with torch.no_grad():
        out = model.get_image_features(**inputs)
    return _projected_features(out)[0].tolist()

One component per image

process_file handles a single image: read its bytes, embed them, and declare a Qdrant point keyed by a stable uuid5 of the path. memo=True skips an image whose bytes did not change, and because each image is its own component, deleting a file removes its point:

```python title=pipeline.py
@coco.fn(memo=True)
async def process_file(
file: FileLike,
target: qdrant.CollectionTarget,
) -> None:
content = await file.read()
embedding = embed_image_bytes(content)
row_id = _image_id(file.file_path.path)
point = qdrant.PointStruct(
id=row_id,
vector=embedding,
payload={"filename": str(file.file_path.path)},
)
target.declare_point(point)




`_image_id` is one line, `str(uuid.uuid5(uuid.NAMESPACE_URL, str(path)))`. The same path always produces the same point ID, so re-embedding an edited photo overwrites its point instead of adding a second one.

### Wire up the app

`app_main` creates the Qdrant collection (its vector size comes straight from CLIP's `projection_dim`), walks the image folder in live mode, and mounts one `process_file` per image with `mount_each`:



```python title=pipeline.py
@coco.fn
async def app_main(sourcedir: pathlib.Path) -> None:
    model, _ = get_clip_model()
    dim: int = model.config.projection_dim  # type: ignore[assignment]

    target_collection = await qdrant.mount_collection_target(
        QDRANT_DB,
        collection_name=QDRANT_COLLECTION,
        schema=await qdrant.CollectionSchema.create(
            vectors=qdrant.QdrantVectorDef(
                schema=VectorSchema(dtype=np.dtype(np.float32), size=dim),
                distance="cosine",
            )
        ),
    )

    files = localfs.walk_dir(
        sourcedir,
        recursive=True,
        path_matcher=PatternFilePathMatcher(
            included_patterns=["**/*.jpg", "**/*.jpeg", "**/*.png"]
        ),
        live=True,  # source supports live watch; api.py runs the app with live=True
    )
    await coco.mount_each(process_file, files.items(), target_collection)

app = coco.App(
    coco.AppConfig(name="ImageSearchQdrantV1"),
    app_main,
    sourcedir=pathlib.Path("./img"),
)

walk_dir(..., live=True) lists the matching files and, when the app runs live, keeps watching the folder. See the local file system connector for how live watching works.
files.items() keys each file by its path relative to img/. That key names the file's component, so CocoIndex can tell which files are new, which changed, and which disappeared.
mount_collection_target creates and manages the Qdrant collection: the schema, upserts, and removal of points whose target state is no longer declared.

Query the index

Search embeds the text with the same CLIP model, this time through the text encoder, and asks Qdrant for the nearest image vectors:

```python title=pipeline.py
def embed_query(text: str) -> list[float]:
model, processor = get_clip_model()
inputs = processor(text=[text], return_tensors="pt", padding=True)
with torch.no_grad():
out = model.get_text_features(**inputs)
return _projected_features(out)[0].tolist()




The FastAPI `/search` endpoint embeds the query and returns the top matches with their scores. `_qdrant_search` in `pipeline.py` calls whichever search method the installed `qdrant-client` version has:



```python title=api.py
@app.get("/search")
async def search(
    q: str = Query(..., description="Search query"),
    limit: int = Query(5, description="Number of results"),
) -> dict[str, Any]:
    query_embedding = pipeline.embed_query(q)

    if _client is None:
        raise RuntimeError("Qdrant client is not initialized.")

    results = pipeline._qdrant_search(
        _client,
        pipeline.QDRANT_COLLECTION,
        query_embedding,
        limit,
    )

    return {
        "results": [
            {
                "filename": (r.payload or {}).get("filename"),
                "score": r.score,
            }
            for r in results
        ]
    }

Run it as a live service

There is no separate build step: api.py starts the app in live mode inside the FastAPI lifespan. It blocks startup until the first sweep reports READY, so the collection is queryable, then keeps watching img/ in the background while it serves requests:

```python title=api.py
@asynccontextmanager
async def lifespan(app: FastAPI) -> AsyncIterator[None]: # type: ignore[override]
global _client
async with coco.runtime():
_client = qdrant.create_client(pipeline.qdrant_url(), prefer_grpc=True)

    # Start a live update; block startup until the initial sweep finishes so
    # the collection is queryable, then keep it running in the background.
    update_handle = pipeline.app.update(live=True)
    async for snap in update_handle.watch():
        if snap.status is coco.UpdateStatus.READY:
            break
    update_task = asyncio.create_task(update_handle.result())

    try:
        yield
    finally:
        update_task.cancel()
        with contextlib.suppress(asyncio.CancelledError):
            await update_task
        _client = None




## Changes in the folder reach the index

Once the server is up, every change to a `.jpg`, `.jpeg`, or `.png` file in `img/` reaches the collection without a command (the patterns are case-sensitive, so `IMG_001.JPG` is skipped). The interactive demo on cocoindex.io shows what each kind of change costs.

What decides each outcome:

- **New file.** The watcher reports `created`, `mount_each` mounts a component for the new path, and its memo has no entry, so CLIP runs. A file still being copied in may fire several events; each one goes through the same memo check.
- **Same bytes, new modification time.** A `touch`, or any write of identical bytes, reports a change, usually `modified`. The memo state of a local file is its modification time plus a fingerprint of its bytes. The time changed, so CocoIndex fingerprints the bytes, finds them unchanged, and skips the function. The point stays as it is.
- **New bytes.** An edited photo has a new fingerprint, so `process_file` runs again and the point is overwritten under the same ID.
- **Rename.** A move is handled as a delete of the old path and a create of the new one. The memo is keyed by the path, so the renamed photo is embedded again and gets a new point ID; the old point is deleted.
- **Delete.** The component for the path is removed, and CocoIndex deletes the point it declared.

Code changes invalidate the memo too: changing the logic of `process_file` (comments and formatting do not count) re-runs it for every image on the next start. Edits to plain helpers it calls, like `embed_image_bytes`, are not tracked. As a safety net against a watcher that silently stops (macOS FSEvents can), `walk_dir` rescans the whole folder once an hour by default; set `rescan_interval` to change that.

## Run the example

From a clone of the [CocoIndex repo](https://github.com/cocoindex-io/cocoindex), start Qdrant, install the example, and run the server. The example ships an `img/` folder with a cat, a dog, an elephant, and a giraffe:



```sh
cd cocoindex/examples/image_search
docker run -d -p 6333:6333 -p 6334:6334 qdrant/qdrant
cp .env.example .env     # QDRANT_URL (defaults to the local container above)
pip install -e .
python -m uvicorn api:app --reload --host 0.0.0.0 --port 8000

Then run the frontend and open it in the browser:

cd frontend && npm install && npm run dev   # http://localhost:5173

Query "elephant" and the elephant ranks first; query "long neck" and the giraffe wins, then the other animals fall in line by CLIP similarity. The images have no tags or captions:

Now drop another image into the img/ folder, for example this squirrel. Because the folder is watched in live mode, it becomes searchable once CLIP has embedded it, with no command to re-run. Search "a bushy tail" and it ranks first.

Acknowledgement

This image search project was largely contributed by @par4m, Param Arora, an open source developer at Google Summer of Code.

How to build index with text embeddings

Linghua Jin — Sat, 24 May 2025 00:23:59 +0000

This post is also on cocoindex.io, with interactive figures. The code uses CocoIndex v1.

Keyword search only finds the words you typed. Ask "how do I log back in?" and a page titled "recover your account access" never comes up, because it shares no words with your question. A page about log files does come up, because it contains "log".

Semantic search matches on meaning, so a question finds its answer even when the words are different. This guide explains the 4 ideas behind it (embeddings, similarity, chunks and an index) and then builds a semantic search index over a folder of Markdown with CocoIndex v1 and Postgres. You do not need to know anything about embeddings to start.

Every number in the figures is real. We computed it for this post with the model the example uses, all-MiniLM-L6-v2, and the 3 sample files that come with the example: the Transformer paper (1706.03762v7.md), the BERT paper (1810.04805v2.md) and the JSON spec, RFC 8259 (rfc8259.md). The full code is in the CocoIndex v1 repo.

What is a text embedding?

An embedding is a list of numbers (a vector) that a model gives to a piece of text. The model is trained so that texts with similar meanings get similar lists. all-MiniLM-L6-v2 gives every text 384 numbers, whether it is 3 words or a full paragraph.

No single number means "password" or "account". The meaning is in the pattern of all 384 together. You can think of the list as coordinates: it puts the text at one point in a space with 384 directions. Texts about the same thing end up near each other in that space.

You cannot draw 384 directions, but you can flatten them to 2 with PCA (principal component analysis), which keeps the largest differences between the points. Here are 10 everyday phrases after that flattening:

We gave the model no labels or groups. It still put "my cat sleeps all day" next to "kittens love to nap" (they share no words), and it put all the ways of saying "I can't get into my account" next to the question.

How close is close?

Search needs a number for "how close". The usual one is cosine similarity: it compares the directions of 2 vectors. 1 means the same direction, 0 means unrelated, and a value below 0 points the other way. This model scales every vector to length 1, so cosine similarity equals the dot product: multiply the 2 lists number by number and add the results.

Postgres with the pgvector extension computes this for you. Its <=> operator returns the cosine distance, which is 1 minus the similarity, so a smaller distance means a closer match. A search is then: embed the question, sort the rows by distance, keep the first few.

Why split files into chunks?

Our sample files are long. The Transformer paper in the sample folder is 46,145 characters, which is 12,746 tokens (the word pieces a model reads). all-MiniLM-L6-v2 reads at most 256 tokens. If you embed the whole file as one text, the model uses the first 2% and drops the rest without a warning.

Even a model with a longer input limit gives a file only 1 vector. A paper about many things gets 1 point somewhere between all of them, and a question about one section has to match that average. When you split the file into chunks and embed each chunk, every passage gets its own point, and the search can return the exact passage that answers the question.

The example uses CocoIndex's RecursiveSplitter. With language="markdown" it picks split points from the Markdown structure, and it aims for chunks of about 2,000 bytes (for English text, about 2,000 characters). Neighboring chunks can overlap by up to 500 bytes, so text near a boundary often appears in both. The 3 sample files give 106 chunks.

The pipeline: 3 files in, 106 rows out

The index is a short pipeline: read the files, split each one into chunks, embed each chunk, and store one row per chunk in Postgres.

Keeping the index correct is the hard part

You could write this pipeline as a plain loop, and it would embed every file once. Keeping the index correct as files change takes more code: on every run you must find what changed, embed only the new text, and make the table match.

CocoIndex is declarative: you describe what the table should contain, not the steps to update it. You write the target as a function of the source, target_state = transformation(source_state), and its Rust engine keeps the 2 in sync. You write the transformation (chunk, embed, and declare the row each chunk should have). The engine detects changes, embeds only the text it has not seen, and applies the inserts and deletes, including cleanup when a file is removed.

Build it

Prerequisites

Clone the CocoIndex repo and work in examples/text_embedding; the commands below expect its main.py and markdown_files/ folder.

Install Postgres with the pgvector extension; that is where the embeddings live. Set the connection string:

export POSTGRES_URL="postgres://cocoindex:cocoindex@localhost/cocoindex"

Install CocoIndex and the packages the example uses:

pip install -U "cocoindex[postgres,sentence_transformers]" asyncpg pgvector numpy python-dotenv

Step 1: define the row

Each chunk becomes one row. A plain dataclass describes the columns. The embedding field is annotated with the embedder, so CocoIndex gets the vector size from the model:

```python title=main.py
import os
import pathlib
from dataclasses import dataclass
from typing import Annotated, AsyncIterator
from numpy.typing import NDArray

import cocoindex as coco
from cocoindex.ops.sentence_transformers import SentenceTransformerEmbedder

DATABASE_URL = os.getenv("POSTGRES_URL", "postgres://cocoindex:cocoindex@localhost/cocoindex")
EMBED_MODEL = "sentence-transformers/all-MiniLM-L6-v2"
EMBEDDER = coco.ContextKeySentenceTransformerEmbedder

@dataclass
class DocEmbedding:
id: int
filename: str
chunk_start: int
chunk_end: int
text: str
embedding: Annotated[NDArray, EMBEDDER]




This is one real row from the sample files, the chunk that holds section 4 of the Transformer paper:

The database connection and the embedder are created once when the app starts (in the `coco_lifespan` function) and shared through [context](https://cocoindex.io/docs/programming_guide/context/):



```python title=main.py
import asyncpg
from cocoindex.connectors import postgres

PG_DB = coco.ContextKey[asyncpg.Pool]("text_embedding_db")

@coco.lifespan
async def coco_lifespan(builder: coco.EnvironmentBuilder) -> AsyncIterator[None]:
    async with asyncpg.create_pool(DATABASE_URL) as pool:
        builder.provide(PG_DB, pool)
        builder.provide(EMBEDDER, SentenceTransformerEmbedder(EMBED_MODEL))
        yield

detect_change=True on EMBEDDER tells CocoIndex to watch the embedder itself. If you change EMBED_MODEL, the next run embeds every chunk again with the new model.

Step 2: chunk and embed

process_chunk embeds one chunk and declares its row. process_file reads a file, splits it into chunks, and runs process_chunk on each one. memo=True turns on memoization: CocoIndex caches the result, and when a file's content and the function's code are unchanged, it skips the file on the next run.

```python title=main.py
from cocoindex.ops.text import RecursiveSplitter
from cocoindex.resources.chunk import Chunk
from cocoindex.resources.file import FileLike
from cocoindex.resources.id import IdGenerator

_splitter = RecursiveSplitter()

@coco.fn
async def process_chunk(
chunk: Chunk,
filename: pathlib.PurePath,
id_gen: IdGenerator,
table: postgres.TableTarget[DocEmbedding],
) -> None:
table.declare_row(
row=DocEmbedding(
id=await id_gen.next_id(chunk.text),
filename=str(filename),
chunk_start=chunk.start.char_offset,
chunk_end=chunk.end.char_offset,
text=chunk.text,
embedding=await coco.use_context(EMBEDDER).embed(chunk.text),
),
)

@coco.fn(memo=True)
async def process_file(
file: FileLike,
table: postgres.TableTarget[DocEmbedding],
) -> None:
text = await file.read_text()
chunks = _splitter.split(text, chunk_size=2000, chunk_overlap=500, language="markdown")
id_gen = IdGenerator()
await coco.map(process_chunk, chunks, file.file_path.path, id_gen, table)




> **Match chunk_size to your model.** `chunk_size=2000` comes from the upstream example. With `all-MiniLM-L6-v2`, most of those chunks are longer than 256 tokens (the median is 433 tokens in the sample files), so the model reads only the start of each one. A smaller `chunk_size` keeps more of each chunk inside the limit: at 1,000 bytes about 3 in 4 chunks of the sample files fit, and at 600 bytes almost all do. A model with a longer input limit can use larger chunks.

`declare_row` does not write to the database at once. It declares a [target state](https://cocoindex.io/docs/programming_guide/target_state/), a row that should exist. CocoIndex compares the declared rows with the ones it declared on the last run and writes only the difference.

### Step 3: wire up the app

`app_main` sets up (mounts) the Postgres table, adds a vector index so Postgres can find the closest vectors quickly, walks the folder of Markdown files, and runs one `process_file` per file:



```python title=main.py
from cocoindex.connectors import localfs
from cocoindex.resources.file import PatternFilePathMatcher

@coco.fn
async def app_main(sourcedir: pathlib.Path) -> None:
    table = await postgres.mount_table_target(
        PG_DB,
        table_name="doc_embeddings",
        table_schema=await postgres.TableSchema.from_class(DocEmbedding, primary_key=["id"]),
        pg_schema_name="coco_examples",
    )
    table.declare_vector_index(column="embedding")

    files = localfs.walk_dir(
        sourcedir,
        recursive=True,
        path_matcher=PatternFilePathMatcher(included_patterns=["**/*.md"]),
        live=True,
    )
    await coco.mount_each(process_file, files.items(), table)

app = coco.App(
    coco.AppConfig(name="TextEmbeddingV1"),
    app_main,
    sourcedir=pathlib.Path("./markdown_files"),
)

coco.mount_each gives each file its own processing component: a unit of work that CocoIndex runs, tracks and cleans up on its own. That is what lets CocoIndex update one file without touching the others.

Step 4: build the index

cocoindex update main

The first run embeds all 106 chunks of the sample files. Add -L (cocoindex update -L main) to keep watching the folder and update the index as files change (live mode).

Query the index

Search mirrors indexing: embed the question with the same model, then ask Postgres for the nearest vectors. In main.py, the pool is created with asyncpg.create_pool(DATABASE_URL, init=register_vector) (from pgvector.asyncpg), so asyncpg can send a vector as a query parameter. A different model puts text in a different space, so its vectors cannot be compared with the stored ones.

```python title=main.py
async def query_once(pool, embedder, query: str, *, top_k: int = 5) -> None:
query_vec = await embedder.embed(query)
async with pool.acquire() as conn:
rows = await conn.fetch(
"""
SELECT filename, text, embedding <=> $1 AS distance
FROM "coco_examples"."doc_embeddings"
ORDER BY distance ASC
LIMIT $2
""",
query_vec,
top_k,
)
for r in rows:
print(f"[{1.0 - float(r['distance']):.3f}] {r['filename']}")
print(f" {r['text']}")




Run it with a question in plain language:



```sh
python main.py "what is self-attention?"

$1 LIMIT 5. The top 5 of 106 chunks by score, with the chunk's first line and where it is: 0.494, #### Attention Visualizations, 1706.03762v7.md 45126 to 46144; 0.445, ### 3.2.3 Applications of Attention in our Model, 13016 to 14977; 0.419, ### 4 Why Self-Attention, 17982 to 19559; 0.343, ## 2 Background, 4905 to 6513; 0.329, ### 3.2.1 Scaled Dot-Product Attention, 9743 to 11378. Score = 1 minus distance."} title={"A search"} width={960} height={452} detailScale={1.4} minWidth={840} />

All 5 hits come from the Transformer paper, and none of them comes from the BERT paper or the JSON spec. The section 4 row from Step 1 is the third hit. We computed this ranking with the same model; <=> gives the same order, because distance is 1 minus similarity. With the vector index, Postgres may return a slightly different top 5, because that index finds nearest neighbors approximately.

A re-run redoes only what changed

Run cocoindex update main again and CocoIndex does only the work that the change needs:

2 mechanisms make this work. memo=True on process_file skips files that did not change. The embedder's embed() is memoized too, so in an edited file only the chunks with new text go to the model. On the write side, each row's id comes from IdGenerator.next_id(chunk.text), so a chunk with the same text keeps the same id across runs. Rows for chunks that are gone are deleted, new chunks are inserted, and rows whose text moved get their new offsets without a new embedding.

Where to go next

You now have the core of semantic search: chunk, embed, store, and query by nearest neighbor. The same pipeline can read from Amazon S3, Azure Blob or Google Drive instead of a local folder, write to a vector database such as Qdrant instead of Postgres, or pass the passages it finds to an LLM for RAG (retrieval-augmented generation: the LLM answers from the passages you retrieved). Because the pipeline is declarative and incremental, none of that changes how the index stays fresh.

For code instead of Markdown, Index your codebase uses the same flow with syntax-aware chunking.

🎉 CocoIndex hits 1,000 stars – powering real-time data for AI

Linghua Jin — Tue, 13 May 2025 04:21:52 +0000

We’ve been building CocoIndex, an ultra performant real-time data transformation framework for AI, and the response from the developer and data engineering community has been incredible. This week, CocoIndex officially surpassed 1,000 stars on GitHub!

🎉 A huge thank you to everyone who has starred, forked, contributed, or shared the project — your support is helping shape the future of open-source AI data infrastructure. ❤️

CocoIndex is a high-performance, real-time data transformation framework designed for modern AI workflows. Built on a blazing-fast Rust engine, CocoIndex simplifies preparing fresh, continuously updated data for AI applications — including embedding generation, knowledge graph construction, and other complex data transformations. It enables developers to build real-time data pipelines that go far beyond the limitations of traditional SQL-based systems.

The philosophy is to have the framework handle the source updates, and having developers only worry about defining a series of data transformation, inspired by spreadsheet.

Here's a bit history about our open source journey:

Data Flow Programming

Unlike conventional workflow orchestration frameworks—where data is typically treated as an opaque payload—CocoIndex elevates data and data transformations to first-class citizens within its architecture. Built around the Dataflow programming model, CocoIndex enforces a design where each transformation is a pure, stateless function that produces a new output field exclusively from its input fields, without relying on hidden states or mutating existing data.

This approach guarantees deterministic and reproducible transformations that can be fully observed and audited. Every intermediate state of the data, both before and after each transformation, is completely visible and traceable, enabling automatic data lineage tracking. This lineage provides deep insights into how data flows through the pipeline, making it easier to debug, troubleshoot, and optimize real-time data pipelines.

By embracing this model, CocoIndex empowers developers to build transparent, incremental, and highly performant AI data pipelines where data freshness and accuracy are maintained effortlessly. This visibility is critical for AI applications that depend on reliable and up-to-date information, such as embedding generation, knowledge graph construction, and complex data enrichment workflows.

For example this is a kind of a data flow: Parse files -> Data Mapping -> Data Extraction -> Knowledge Graph

Particularly, users don't define data operations like creation, update, or deletion. Rather, they define something like -
for a set of source data, this is the transformation or formula. The framework takes care of the data operations such as when to create, update, or delete.

With operations like

# ingest
data['content'] = flow_builder.add_source(...) 

# transform
data['out'] = data['content'] 
    .transform(...)
    .transform(...)

# collect data
collector.collect(...)

# export to db, vector db, graph db ...
collector.export(...)

Data Freshness

As a cutting-edge data framework, CocoIndex significantly enhances data freshness by leveraging incremental processing—a core feature that ensures only changed data is processed and updated, enabling faster, more efficient real-time data pipelines.

The framework takes care of:

Change data capture
Figuring out what exactly needs to be updated, and only updating that without having to recompute everything throughout.

This makes it fast to reflect any source updates to the target store. If you have concerns with surfacing stale data to AI agents and are spending lots of effort working on infrastructure pieces to optimize latency, the framework actually handles it for you.

Built-in modules and custom modules

At CocoIndex, we recognize that data preparation is highly use-case specific, and there is no one-size-fits-all solution for every AI workflow. With rapid advancements in AI models and data technologies, selecting the best tools involves balancing multiple factors such as data quality, performance, and cost efficiency.

The data infrastructure ecosystem offers a wide variety of specialized tools — including parsers, embedding generation models, vector databases, graph databases, and more — each optimized for different goals and scenarios.

Instead of building everything from scratch, CocoIndex adopts a modular, composition-based approach. We provide native plugin support that enables seamless integration with your preferred data tools and services. By standardizing plugin interfaces, CocoIndex allows developers to easily plug in, swap, or upgrade components like building blocks, just like LEGO. This flexibility empowers you to tailor your AI data pipelines for maximum efficiency and adaptability.

And we can focus on what we're best at - building the best incremental pipeline and compute engine in the AI era.

Tooling

In addition to standard statistics and reports with CocoIndex, we are building a product called CocoInsight. It has zero pipeline data retention and connects to your on-premise CocoIndex server for pipeline insights. This makes data directly visible and easy to develop ETL pipelines.

Especially, as we mentioned earlier, CocoIndex is heavily influenced by spreadsheets. This is directly reflected in CocoInsight, as you can see all the data in the spreadsheet and how it looks after each transformation.

This is an example of mapping:

In addition to the data insights for data transformation, we also aim to provide tooling to make it directly visible to understand the LLM transformations, for example:

Understand and trouble shooting the chunking, if there's any, and debug why a chunk is showing from search results

Understand the relation extraction for Graphs

and more

Data should be transparent, especially for ETL frameworks.

Support us

We are constantly improving, and more features and examples are coming soon. If you love this article, please drop us a star ⭐ at GitHub repo to help us grow. You can also find us on Discord, I try to be there 24/7 😊 Looking forward to discussing CocoIndex or any topics in Data/AI infra with you!

Real-Time Knowledge Graph for Product recommendation with LLM taxonomy extraction

Linghua Jin — Thu, 08 May 2025 05:22:57 +0000

This post is also on cocoindex.io, with interactive figures. The code uses CocoIndex v1.

A pile of product listings has recommendations hiding in plain sight: a pen pairs with ink refills and a notebook; a monitor pairs with a stand and an HDMI cable. That knowledge is locked in the prose of each listing. The recommendation engine uses these relationships: an LLM pulls the knowledge out, and a Neo4j graph connects it.

For each product, CocoIndex v1 asks an LLM two things: what the product is (its taxonomy), and what a buyer might also need (its complementary taxonomy). Those labels become nodes and edges in a Neo4j graph, and the recommendation falls out of the graph: a product whose "also need" matches another product's "is" is the thing to cross-sell.

The full code is in the CocoIndex v1 repo.

The data model

Two node types and two edge types are enough:

Product nodes, one per listing, keyed by id (the filename), carrying title and price.
Taxonomy nodes, one per distinct label (gel pen, notebook, ink refill), keyed by value and shared across products.
PRODUCT_TAXONOMY edges, Product → Taxonomy: what the product is.
PRODUCT_COMPLEMENTARY_TAXONOMY edges, Product → Taxonomy: what pairs with it.

The nodes are plain dataclasses:

```python title=main.py
from dataclasses import dataclass

@dataclass
class Product:
id: str # primary key (the filename stem)
title: str
price: float

@dataclass
class Taxonomy:
value: str # primary key (the taxonomy label)




## The flow

Because taxonomy labels are shared across products, the pipeline runs in two phases: per-product extraction declares each `Product` node and carries its labels forward, then a single graph-building pass declares the deduplicated `Taxonomy` nodes and all the edges.

## Connect Neo4j

The Neo4j connection is provided once in the app lifespan and read anywhere with a [`ContextKey`](https://cocoindex.io/docs/programming_guide/core_concepts/). The LLM model name is a context value too, declared with `detect_change=True` so swapping models re-extracts everything against the new one:



```python title=main.py
import cocoindex as coco
from cocoindex.connectors import neo4j

KG_DB = coco.ContextKey[neo4j.ConnectionFactory]("kg_db")
LLM_MODEL = coco.ContextKey[str]("llm_model", detect_change=True)

@coco.lifespan
async def coco_lifespan(builder: coco.EnvironmentBuilder) -> AsyncIterator[None]:
    builder.provide(
        KG_DB,
        neo4j.ConnectionFactory(
            uri=os.environ.get("NEO4J_URI", "bolt://localhost:7687"),
            auth=(os.environ["NEO4J_USER"], os.environ["NEO4J_PASSWORD"]),
        ),
    )
    builder.provide(LLM_MODEL, os.environ.get("LLM_MODEL", "openai/gpt-4.1"))
    yield

Extract the taxonomy

The extraction schema is the whole prompt. A Pydantic model asks for two lists, and the field descriptions tell the model exactly what a good label looks like: a concise noun, US English, specific rather than broad (gel pen, not office supplies):

```python title=main.py
import pydantic

class ProductTaxonomy(pydantic.BaseModel):
name: str = pydantic.Field(
description="A concise noun for the product's core function, no branding "
"or style. Common US English, lowercase. Prefer specific ('pen') over "
"broad ('office supplies')."
)

class ProductTaxonomyInfo(pydantic.BaseModel):
taxonomies: list[ProductTaxonomy] = pydantic.Field(
description="Taxonomies describing what this product is."
)
complementary_taxonomies: list[ProductTaxonomy] = pydantic.Field(
description="Taxonomies for complementary products a buyer might also need."
)




Extraction is one CocoIndex function that calls the model through [instructor](https://github.com/instructor-ai/instructor) over [LiteLLM](https://cocoindex.io/docs/ops/litellm/), forcing the response to match the schema. `memo=True` caches the result by content, so an unchanged product is never sent to the model twice:



```python title=main.py
@coco.fn(memo=True)
async def extract_taxonomy(detail: str) -> ProductTaxonomyInfo:
    client = instructor.from_litellm(litellm.acompletion, mode=instructor.Mode.JSON)
    result = await client.chat.completions.create(
        model=coco.use_context(LLM_MODEL),
        response_model=ProductTaxonomyInfo,
        messages=[
            {"role": "system", "content": TAXONOMY_PROMPT},
            {"role": "user", "content": detail},
        ],
    )
    return ProductTaxonomyInfo.model_validate(result.model_dump())

For the gel pen, the model returns gel pen as the taxonomy, and notebook, pen refill, and the like as complementary taxonomies:

Phase 1: one Product node per file

process_file handles a single product: read the JSON, declare its Product node into the graph, extract the taxonomy, and return the labels for phase two. declare_record writes one node; CocoIndex reconciles it into Neo4j by the primary key:

```python title=main.py
@coco.fn(memo=True)
async def process_file(
file: FileLike,
product_table: neo4j.TableTarget[Product],
) -> ProductTaxonomies:
raw = json.loads(await file.read_text())
product_id = file.file_path.path.name.removesuffix(".json")
price = float(str(raw["price"]).lstrip("$").replace(",", ""))

product_table.declare_record(row=Product(id=product_id, title=raw["title"], price=price))

info = await extract_taxonomy(PRODUCT_TEMPLATE.render(**raw))
return ProductTaxonomies(
    product_id=product_id,
    taxonomies=[t.name for t in info.taxonomies],
    complementary=[t.name for t in info.complementary_taxonomies],
)




Each declared `Product` becomes a node with `id` as its key and every field as a property:

## Phase 2: shared Taxonomy nodes and the edges

Taxonomy labels repeat across products, so one pass owns them. It gathers every label, declares the `Taxonomy` nodes, and declares the edges. Because a node is keyed by `value`, declaring `gel pen` from ten different products still produces exactly one node: CocoIndex deduplicates by the primary key.

`declare_relation` connects a product to a taxonomy by their keys. Two relation targets, one per edge type, share the same `Product` and `Taxonomy` node tables:



```python title=main.py
@coco.fn
async def build_graph(
    products: list[ProductTaxonomies],
    taxonomy_table: neo4j.TableTarget[Taxonomy],
    product_taxonomy_rel: neo4j.RelationTarget[Any],
    complementary_rel: neo4j.RelationTarget[Any],
) -> None:
    labels = {t for p in products for t in (*p.taxonomies, *p.complementary)}
    for value in labels:
        taxonomy_table.declare_record(row=Taxonomy(value=value))

    for p in products:
        for t in set(p.taxonomies):
            product_taxonomy_rel.declare_relation(from_id=p.product_id, to_id=t)
        for t in set(p.complementary):
            complementary_rel.declare_relation(from_id=p.product_id, to_id=t)

The PRODUCT_TAXONOMY edges link each product to what it is:

The PRODUCT_COMPLEMENTARY_TAXONOMY edges add what pairs with it. These are the edges that cross between products and make recommendations possible:

Wire up the app

app_main mounts the two node tables and the two relation targets, walks the product folder, fans out one process_file per product, then runs the single build_graph pass over the collected labels:

```python title=main.py
@coco.fn
async def app_main(sourcedir: pathlib.Path) -> None:
product_table = await neo4j.mount_table_target(
KG_DB, "Product",
await neo4j.TableSchema.from_class(Product, primary_key="id"),
primary_key="id",
)
taxonomy_table = await neo4j.mount_table_target(
KG_DB, "Taxonomy",
await neo4j.TableSchema.from_class(Taxonomy, primary_key="value"),
primary_key="value",
)
product_taxonomy_rel = await neo4j.mount_relation_target(
KG_DB, "PRODUCT_TAXONOMY", product_table, taxonomy_table
)
complementary_rel = await neo4j.mount_relation_target(
KG_DB, "PRODUCT_COMPLEMENTARY_TAXONOMY", product_table, taxonomy_table
)

files = localfs.walk_dir(
    sourcedir, recursive=True,
    path_matcher=PatternFilePathMatcher(included_patterns=["**/*.json"]),
)
products = list(await asyncio.gather(*[
    coco.use_mount(coco.component_subpath("file", key), process_file, file, product_table)
    async for key, file in files.items()
]))

await coco.mount(
    coco.component_subpath("build_graph"),
    build_graph, products, taxonomy_table, product_taxonomy_rel, complementary_rel,
)

app = coco.App(coco.AppConfig(name="ProductRecommendation"), app_main,
sourcedir=pathlib.Path("./products"))




## Run it

Start Neo4j, install the example, and build the graph. The example ships a `products/` folder of sample listings:



```sh
docker run -d -p 7474:7474 -p 7687:7687 -e NEO4J_AUTH=neo4j/cocoindex neo4j:5.26-community
pip install "cocoindex[neo4j]" instructor litellm pydantic jinja2
cocoindex update main

On the nine sample products that is nine Product nodes, around forty Taxonomy nodes, and both edge types wired up. Editing one product re-extracts only that product, then the graph diffs: new nodes and edges are added, and ones no longer supported anywhere are removed.

Query the recommendations

Open Neo4j Browser (neo4j / cocoindex) and let the graph do the recommending. To find products to pair with a gel pen, walk from the pen's complementary taxonomy to any product that is that taxonomy:

MATCH (:Taxonomy {value: "gel pen"})<-[:PRODUCT_TAXONOMY]-(:Product)
      -[:PRODUCT_COMPLEMENTARY_TAXONOMY]->(need:Taxonomy)
MATCH (rec:Product)-[:PRODUCT_TAXONOMY]->(need)
RETURN DISTINCT rec.title

On the sample data, recommending for a pen surfaces the notepad and the multipurpose paper, the cross-sell you would want. To see the whole graph, MATCH p=()-->() RETURN p:

Build Real-Time Knowledge Graph For Documents with LLM

Linghua Jin — Thu, 01 May 2025 01:36:50 +0000

This post is also on cocoindex.io, with interactive figures. The code uses CocoIndex v1.

Documentation is a web of concepts pretending to be a list of files. Every page asserts relationships ("CocoIndex supports incremental processing"), but they are locked in prose. This post turns a folder of Markdown documents (CocoIndex's own docs) into a Neo4j knowledge graph: an LLM reads each document and extracts the relationships between the concepts it talks about, and CocoIndex reconciles them into the graph incrementally as documents change.

We generate two kinds of relationships:

Relationships between subjects and objects. E.g., "CocoIndex supports Incremental Processing".
Mentions of entities in a document. E.g., basics.md mentions CocoIndex and Incremental Processing.

The end result is a small, well-defined graph schema: two node labels (Document and Entity) and two relationship types (RELATIONSHIP between entities, and MENTION connecting a document to the entities it references). Everything the LLM extracts lands in one of those four shapes, which is what keeps the graph queryable rather than a loose pile of triples.

The source code is available at CocoIndex Examples - docs_to_knowledge_graph.

Prerequisites

Install Neo4j, a graph database. A docker one-liner is in the Run it section below.
An LLM API key. The example defaults to OpenAI, and the model is a LiteLLM id, so you can set LLM_MODEL=ollama/llama3.2 to run the extraction locally instead.

The graph schema

All nodes for Neo4j need two things:

Label: the type of the node. E.g., Document, Entity.
Primary key: the field that uniquely identifies the node. E.g., filename for Document nodes.

CocoIndex uses the primary key to match and deduplicate nodes: declaring the same key twice produces one node. In CocoIndex v1 the schema is plain dataclasses:

```python title=main.py
from dataclasses import dataclass

@dataclass
class Document:
filename: str # primary key
title: str
summary: str

@dataclass
class Entity:
value: str # primary key — the concept name

@dataclass
class Relationship:
"""RELATIONSHIP edge payload. id is a stable hash of the triple so the
same (subject, predicate, object) always maps to a single edge; the
predicate is stored as an edge property."""
id: int
predicate: str




`MENTION` carries no payload, so it needs no dataclass at all: the Neo4j connector derives its identity from the `(document, entity)` endpoints, one edge per pair.

## The flow

Entities are shared across documents: the same concept, say `CocoIndex`, shows up in triples from many files, and every one of those references must collapse onto a single `Entity` node. So the pipeline runs in two phases. Phase 1 processes each document independently: it declares the `Document` node and extracts the relationship triples. Phase 2 is a single pass over all the triples: it owns the deduplicated `Entity` nodes and declares the `RELATIONSHIP` and `MENTION` edges.

Because phase 1 runs one [component](https://cocoindex.io/docs/programming_guide/processing_component) per file, editing one document re-extracts only that document. The graph pass then diffs: nodes and edges that are no longer supported by any document are removed automatically.

## Connect Neo4j

The Neo4j connection is provided once in the app lifespan and read anywhere with a [`ContextKey`](https://cocoindex.io/docs/programming_guide/core_concepts/). The LLM model name is a context value too, declared with `detect_change=True` so swapping models re-extracts everything against the new one:



```python title=main.py
import cocoindex as coco
from cocoindex.connectors import localfs, neo4j

KG_DB = coco.ContextKey[neo4j.ConnectionFactory]("kg_db")
LLM_MODEL = coco.ContextKey[str]("llm_model", detect_change=True)

@coco.lifespan
async def coco_lifespan(builder: coco.EnvironmentBuilder) -> AsyncIterator[None]:
    builder.provide(
        KG_DB,
        neo4j.ConnectionFactory(
            uri=os.environ.get("NEO4J_URI", "bolt://localhost:7687"),
            auth=(
                os.environ.get("NEO4J_USER", "neo4j"),
                os.environ.get("NEO4J_PASSWORD", "cocoindex"),
            ),
            database=os.environ.get("NEO4J_DATABASE", "neo4j"),
        ),
    )
    builder.provide(LLM_MODEL, os.environ.get("LLM_MODEL", "openai/gpt-5-mini"))
    yield

Summarize each document

The extraction schemas are Pydantic models, and the field descriptions are part of the prompt. For the summary we ask for a title and one paragraph:

```python title=main.py
import pydantic

class DocumentSummary(pydantic.BaseModel):
title: str = pydantic.Field(description="A concise title for the document.")
summary: str = pydantic.Field(
description="A one-paragraph summary of what the document covers."
)




Extraction is one CocoIndex function that calls the model through [instructor](https://github.com/instructor-ai/instructor) over [LiteLLM](https://cocoindex.io/docs/ops/litellm/), forcing the response to match the schema. `memo=True` [memoizes](https://cocoindex.io/docs/advanced_topics/memoization_keys) the result by content, so an unchanged document is never sent to the model twice:



```python title=main.py
@coco.fn(memo=True)
async def extract_summary(content: str) -> DocumentSummary:
    client = instructor.from_litellm(litellm.acompletion, mode=instructor.Mode.JSON)
    result = await client.chat.completions.create(
        model=coco.use_context(LLM_MODEL),
        response_model=DocumentSummary,
        messages=[
            {"role": "system", "content": SUMMARY_PROMPT},
            {"role": "user", "content": content},
        ],
    )
    return DocumentSummary.model_validate(result.model_dump())

For each document the model returns a typed summary:

Extract relationship triples

A relationship is a (subject, predicate, object) triple. The subject and object become Entity nodes, and the predicate is the label on the edge between them:

The Pydantic model mirrors that shape, and the system prompt steers the model toward concepts rather than code:

```python title=main.py
class ExtractedRelationship(pydantic.BaseModel):
subject: str = pydantic.Field(
description="The concept the statement is about, e.g. 'CocoIndex'."
)
predicate: str = pydantic.Field(
description="How subject relates to object, e.g. 'supports'."
)
object: str = pydantic.Field(
description="The related concept, e.g. 'Incremental Processing'."
)

class RelationshipList(pydantic.BaseModel):
relationships: list[ExtractedRelationship] = pydantic.Field(default_factory=list)

RELATIONSHIP_PROMPT = """\
You extract a concept knowledge graph from technical documentation.
List the salient (subject, predicate, object) relationships between concepts.
Focus on concepts and ignore code examples and implementation details.
Use concise noun phrases for subjects and objects and a short verb phrase for
the predicate. Return only relationships supported by the text.
"""




The extraction function has the same shape as the summary one, memoized the same way:



```python title=main.py
@coco.fn(memo=True)
async def extract_relationships(content: str) -> list[Triple]:
    client = instructor.from_litellm(litellm.acompletion, mode=instructor.Mode.JSON)
    result = await client.chat.completions.create(
        model=coco.use_context(LLM_MODEL),
        response_model=RelationshipList,
        messages=[
            {"role": "system", "content": RELATIONSHIP_PROMPT},
            {"role": "user", "content": content},
        ],
    )
    validated = RelationshipList.model_validate(result.model_dump())
    return [Triple(r.subject, r.predicate, r.object) for r in validated.relationships]

On the CocoIndex docs, one file yields a few dozen triples:

Phase 1: one Document node per file

process_file handles a single document: read the content, extract the summary, declare the Document node into the graph, and return the triples for phase 2. declare_record writes one node; CocoIndex reconciles it into Neo4j by the primary key:

```python title=main.py
@coco.fn(memo=True)
async def process_file(
file: localfs.File,
document_table: neo4j.TableTarget[Document],
) -> DocTriples:
content = await file.read_text()
filename = file.file_path.path.as_posix()

summary = await extract_summary(content)
document_table.declare_record(
    row=Document(filename=filename, title=summary.title, summary=summary.summary)
)

triples = await extract_relationships(content)
return DocTriples(filename=filename, triples=triples)




Each declared record becomes a `Document` node keyed by `filename`, with the title and summary carried across as node properties:

## Phase 2: build the concept graph

Phase 2 receives the triples from every document. Because `Entity` nodes are keyed by `value`, declaring `CocoIndex` from ten different documents still produces exactly one node: CocoIndex deduplicates by the primary key. The edges then fan out from that one node, which is exactly what you want when you later query "what mentions this concept?".

One pass gathers the distinct entities and mentions, then declares everything. `declare_relation` connects two nodes by their primary keys, and `generate_id` hashes each triple so the same fact always maps to the same edge, whether it is re-extracted or asserted again by another document:



```python title=main.py
@coco.fn
async def build_graph(
    docs: list[DocTriples],
    entity_table: neo4j.TableTarget[Entity],
    relationship_rel: neo4j.RelationTarget[Relationship],
    mention_rel: neo4j.RelationTarget[Any],
) -> None:
    entities: set[str] = set()
    mentions: set[tuple[str, str]] = set()  # (filename, entity value)

    for doc in docs:
        for t in doc.triples:
            entities.add(t.subject)
            entities.add(t.object)
            mentions.add((doc.filename, t.subject))
            mentions.add((doc.filename, t.object))

            rel_id = await generate_id((t.subject, t.predicate, t.object))
            relationship_rel.declare_relation(
                from_id=t.subject,
                to_id=t.object,
                record=Relationship(id=rel_id, predicate=t.predicate),
            )

    for value in entities:
        entity_table.declare_record(row=Entity(value=value))

    for filename, entity in mentions:
        mention_rel.declare_relation(from_id=filename, to_id=entity)

The RELATIONSHIP edges connect Entity nodes, with the predicate stored on the edge:

The MENTION edges record which document named which concept. For example, basics.md has a sentence "CocoIndex supports Incremental Processing", so it mentions both CocoIndex and Incremental Processing:

Wire up the app

app_main mounts the two node tables and the two relation targets, walks the docs folder, fans out one process_file per document, then runs the single build_graph pass over the collected triples:

```python title=main.py
@coco.fn
async def app_main(sourcedir: pathlib.Path) -> None:
document_table = await neo4j.mount_table_target(
KG_DB,
"Document",
await neo4j.TableSchema.from_class(Document, primary_key="filename"),
primary_key="filename",
)
entity_table = await neo4j.mount_table_target(
KG_DB,
"Entity",
await neo4j.TableSchema.from_class(Entity, primary_key="value"),
primary_key="value",
)
relationship_rel = await neo4j.mount_relation_target(
KG_DB,
"RELATIONSHIP",
entity_table,
entity_table,
await neo4j.TableSchema.from_class(Relationship, primary_key="id"),
primary_key="id",
)
mention_rel = await neo4j.mount_relation_target(
KG_DB, "MENTION", document_table, entity_table
)

files = localfs.walk_dir(
    sourcedir,
    recursive=True,
    path_matcher=PatternFilePathMatcher(included_patterns=["**/*.md", "**/*.mdx"]),
)
file_coros = []
async for path_key, file in files.items():
    file_coros.append(
        coco.use_mount(
            coco.component_subpath("file", path_key),
            process_file,
            file,
            document_table,
        )
    )
docs: list[DocTriples] = list(await asyncio.gather(*file_coros))

await coco.mount(
    coco.component_subpath("build_graph"),
    build_graph,
    docs,
    entity_table,
    relationship_rel,
    mention_rel,
)

app = coco.App(
coco.AppConfig(name="DocsToKnowledgeGraph"),
app_main,
sourcedir=pathlib.Path("./markdown_files"),
)




`RELATIONSHIP` is mounted with the `Relationship` schema so each distinct triple, keyed by the hashed `id`, becomes its own edge. `MENTION` is mounted without a schema, so the connector keys it by the endpoints.

## Run it

Start Neo4j, install the example, and build the graph. The example ships a `markdown_files/` folder of sample docs so it runs out of the box:



```sh
docker run -d -p 7474:7474 -p 7687:7687 -e NEO4J_AUTH=neo4j/cocoindex --name cocoindex-neo4j neo4j:5.26-community
cp .env.example .env     # set OPENAI_API_KEY (or LLM_MODEL=ollama/llama3.2 to run locally)
pip install -e .
cocoindex update main

To graph your own docs, drop .md / .mdx files into markdown_files/ (or point sourcedir at your real docs folder) and re-run. Only changed documents are re-extracted; a no-change re-run makes zero LLM calls.

Browse the knowledge graph

Open Neo4j Browser (username neo4j, password cocoindex) and run a Cypher query. To see how concepts relate:

MATCH (a:Entity)-[r:RELATIONSHIP]->(b:Entity)
RETURN a.value, r.predicate, b.value

To find the concepts mentioned in the most documents:

MATCH (d:Document)-[:MENTION]->(e:Entity)
RETURN e.value, count(DISTINCT d) AS docs
ORDER BY docs DESC LIMIT 10

Or MATCH p=()-->() RETURN p to see everything:

Near-duplicate entities. The LLM will sometimes name the same concept two ways ("CocoIndex" vs "Cocoindex"). The meeting notes graph example adds an embedding plus LLM entity-resolution pass that collapses near-duplicates; it drops into this pipeline between the two phases.

How to do incremental processing for ETL - by examples

Linghua Jin — Tue, 08 Apr 2025 17:37:46 +0000

This post is also on cocoindex.io, with interactive figures. The code uses CocoIndex v1.

Incremental processing is one of the core values provided by CocoIndex.
In CocoIndex, users declare the transformation, and don't need to worry about the work to keep the index and source in sync.

CocoIndex creates & maintains an index, and keeps the derived index up to date based on source updates, with minimal computation and changes. That makes it suitable for ETL/RAG or any transformation tasks that stay low latency between source and index updates, and also minimizes the computation cost.

What is incremental processing?

Figuring out what exactly needs to be updated, and only updating that without having to recompute everything throughout.

How does it work?

You don't really need to do anything special, just focus on defining the transformation needed.

CocoIndex automatically tracks the lineage of your data and maintains a cache of computation results. When you update your source data, CocoIndex will:

Identify which parts of the data have changed
Only recompute transformations for the changed data
Reuse cached results for unchanged data
Update the index with minimal changes

And CocoIndex will handle the incremental processing for you.

CocoIndex provides two modes to run the pipeline:

One-time update: Once triggered, CocoIndex updates the target data to reflect the version of source data up to the current moment.
Live update: CocoIndex continuously reacts to changes of source data and updates the target data accordingly, based on various change capture mechanisms for the source.

Both modes run with incremental processing. You can view more details in Core Concepts.

Who needs incremental processing?

Many people may think incremental processing is only beneficial for large scale data. Thinking carefully,
it really depends on the cost and requirement for data freshness.

Google processes huge scale data, backed by a huge amount of resources.
Your data scale is much less, but your resource provision is also much more limited.

Incremental processing is needed under the following conditions:

High freshness requirement

For most user-facing applications this is needed. e.g. users update their documents,
and it's unexpected if they see stale information in search results.

If the search result is fed into an AI agent, it may mean an unexpected response to users (i.e. an LLM generates output based on inaccurate information).
It's more dangerous and users may even take the unexpected response without noticing.
Transformation cost is significantly higher than retrieval itself

Overall, say T is your most acceptable staleness. If you don't want to recompute the whole thing repeatedly every cycle of T,
you will need incremental processing more or less.

How CocoIndex does incremental processing, by examples

We could take a look at a few examples to understand what CocoIndex handles behind the scenes for incremental processing.

Example 1: Update a document

Consider this scenario:

I have a document. Initially, it's split into 5 chunks, resulting in 5 rows with their embeddings in the index.
After it's updated, 3 of them are exactly the same as previous. Others are changed.

So we need to keep 3 rows, remove 2 previously existing rows, and add 2 new rows. These need to happen behind the scenes:

Ideally, we only recompute embeddings for the 4 new rows, and reuse them for 3 unchanged chunks. This can save computation cost especially when the embedding API charges by usage. CocoIndex achieves this by maintaining a cache for heavy steps like embedding, and when the input for a transformation step isn't changed, the output will be reused.
Besides, we also maintain a lineage tracking in the internal storage. It keeps track of which rows in the index are derived from the previous version of this document, to make sure stale versions are properly removed.

CocoIndex takes care of this.

Example 2: Delete a document

Continuing with the same example. If we delete the document later, we need to delete all 7 rows derived from the document.
Again, this needs to be based on the lineage tracking maintained by CocoIndex.

Example 3: Change of the transformation flow

The transformation flow may also be changed, for example, the chunking logic is upgraded, or a parameter passed to the chunker is adjusted.
This may result in the following scenario:

Before the change, the document is split into 5 chunks, resulting in 5 rows with their embeddings in the index.
After the change, they become 6 chunks: 4 of the previous chunks remain unchanged, the remaining 1 is split into 2 smaller chunks.

This falls into a similar situation as document update (example 1), and CocoIndex will take care of it. The approach is similar, while this involves some additional considerations:

We can still safely reuse embeddings for 4 unchanged chunks by the caching mechanism: this needs a prerequisite that the logic and parameters for embedding are unchanged.
If the changed part is the embedding logic itself, we will recompute the embeddings for everything.
CocoIndex is able to see if the logic or parameters for a step changed from the cached version, by putting this additional information in the cache key.
To remove stale rows in the target index, the lineage tracking works well again.
Note that some other systems handle stale output deletions on source update/deletion by replaying the transformation logic on the previous version of input:
this only works well when transformation is fully deterministic and never upgraded.
CocoIndex's lineage tracking based approach doesn't have this limitation: it tracks derived outputs even when transformation logic changes or produces different results.

Example 4: Multiple inputs involved: merge / lookup / clustering

All examples above are simple cases: each single input row (e.g. a document) is involved independently during each specific transformation.

CocoIndex is a highly customizable framework, not only limited to simple chunking and embedding.
It allows users to perform more complex advanced transformations, such as:

Merge.
For example, you're building an index for "all AI products", and you want to combine information from multiple sources, some products exist in one source and some in multiple.
For each product, you want to combine information from different sources.
Lookup.
For example, you also have a data source about company information.
During your transformation for each product, you want to enrich it with information of the company building the product, so a lookup for the company information is needed.
Clustering.
For example, you want to cluster different products into scenarios, and create cluster-level summaries based on information of products in the cluster.

The common theme is that during transformation, multiple input rows (coming from single or multiple sources) need to be involved at the same time.
Once a single input row is updated or deleted, CocoIndex will need to fetch other related rows from the same or other sources.
Here which other rows are needed is based on which are involved in the transformations.
CocoIndex keeps track of such relationships, and will fetch related rows and trigger necessary reprocessings incrementally.

Change data capture (CDC)

1. When source supports push change

Some source connectors support push change.
For example, Google Drive supports drive-level changelog and sends change notifications to your public URL,
which is applicable for team drive and personal drive (only by OAuth, service account not supported).
When a file is created, updated, or deleted, CocoIndex could compute based on the diff.

Google Drive Change Log. - Google Drive change log is only supported on the drive level, not the file/folder level.

With service accounts, you can only enable changelog for team drives, by sharing the team drive with it.

2. Metadata-based, fullscan

All source connectors in CocoIndex provide a basic list API capability, which enables a generic change detection mechanism applicable to any data source.

For example, with local file systems, we can traverse all directories and subdirectories recursively to get the full list of entries and their metadata (like modification time). By comparing the current state with the previous state, CocoIndex can detect changes even without source-specific change notifications.

This approach works universally across all data sources, though when the number of entries is large, performing a complete traversal can be resource-intensive.

This is a generic mechanism applicable to all data sources, so all data sources can use this even if there's no source-specific change capture.

3. Optimizations for specific sources

When the source connector provides advanced features to list entries, for example, listing most recently changed entries,
CocoIndex could take advantage of that to provide more efficient change detection.

For example, when changelog is not available for Google Drive (see the conditions in the tip above),
CocoIndex could monitor the change based on last modified vs last poll time, with a periodic trigger to check modified entries.
However, this cannot capture full change, for example when an entry has been deleted.

Cache

In CocoIndex, every piece of the building block in the pipeline can be cached, keyed by memoization keys.
Currently, whether or not cache is enabled is decided by the implementation of the function.
For builtin functions, if it performs heavy transformation, cache is enabled.

The decision of whether a cached result can be reused comes down to a single question: did anything the function depends on change? CocoIndex answers it by reducing every input (the function arguments, declared dependencies, and context values) to a deterministic fingerprint. Each value is mapped to a canonical form (primitives and containers directly, dataclasses and Pydantic models field by field, and a pickle fallback for anything else), and those forms are combined into one key. If that key matches a previously stored entry, the cached output is a candidate for reuse. This is the mechanism behind every reuse decision described in the examples above: when the embedding step sees the same chunk text and the same logic, its fingerprint is unchanged, and the prior embedding is returned without recomputation.

The subtle part is separating identity from freshness. A file's path tells you which input you're looking at; its modification time and content tell you whether that input has changed. Folding everything into one key works, but it forces an expensive check on every run: for a file, that means reading and hashing the whole thing each time. CocoIndex lets you split the two: keep the stable identifier (path, URL, primary key) in the memoization key, and attach a memo state that validates freshness only after the fingerprint matches. A state function can do a cheap check first (compare the modification time) and fall back to the expensive one (hash the content) only when the cheap check fails. Even then, it can confirm the content is identical and reuse the cached result while quietly recording the new modification time. State validation can also run asynchronously, which is what makes freshness checks like an S3 HEAD request or an If-Modified-Since header practical. See memoization keys for the full model.

Custom functions can take a parameter memo.
When True, CocoIndex memoizes the result of the function for reuse during reprocessing.
We recommend setting this to True for any function that is computationally intensive.

Output will be reused if both of these are unchanged: the input data and the behavior of the function.
When you change the function's logic, bump version so the previously memoized results are invalidated.

For example, this enables memoization for a function:

import cocoindex as coco

@coco.fn(memo=True, version=1)
async def compute_something(arg1: str, arg2: int | None = None) -> str:
    ...

For a synchronous function you want to batch, force the async interface:

@coco.fn.as_async(memo=True, batching=True)
def compute_batch(args: list[str]) -> list[str]:
    ...

Thanks for reading and we love to hear from you!
Browse the CocoIndex source on GitHub.