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

CocoIndex supports multi-modal processing natively -
it could process both text and image with the same programming model and observe in the same user flow (in CocoInsight).

In this blog, we’ll walk through a comprehensive example of building a scalable face recognition pipeline using CocoIndex.
We’ll show how to extract and embed faces from images, structure the data relationally,
and export everything into a vector database for real-time querying.

CocoInsight can now visualize identified sections of an image based on the bounding boxes and makes it easier to understand and
evaluate AI extractions - seamlessly attaching computed features in the context of unstructured visual data.

If you find this tutorial helpful, we’d greatly appreciate it if you could ⭐ star
CocoIndex on GitHub.

Use Cases

Photo search
Face-based access control and surveillance
Visual deduplication and identity detection
Multimodal search involving people or facial identity
Social graph analysis from photos

What We Will Achieve

The photo taken of this conference's participants is sometimes entitled
"The Most Intelligent Picture Ever Taken", for its depiction of the world's leading physicists gathered together in one shot
(Wikipedia)

Here’s what we want to accomplish:

Detect all faces in the image and extract their bounding boxes
Crop and encode each face image into a 128-dimensional face embedding
Store metadata and vectors in a structured index to support queries like: “Find all similar faces to this one” or “Search images that include this person”

⭐ You can find the full code here.

Indexing Flow

We ingest a list of images.
For each image, we:
- Extract faces from the image.
- Compute embeddings for each face.
We export the following fields to a table in Postgres with PGVector:
- Filename, rect, embedding for each face.

Core Components

Image Ingestion

We monitor an images/ directory using the built-in LocalFile source. All newly added files are automatically processed and indexed.

python
CopyEdit
@cocoindex.flow_def(name="FaceRecognition")
def face_recognition_flow(flow_builder, data_scope):
    data_scope["images"] = flow_builder.add_source(
        cocoindex.sources.LocalFile(path="images", binary=True),
        refresh_interval=datetime.timedelta(seconds=10),
    )

This creates a table with filename and content fields. 📂

You can connect it to your S3 Buckets (with SQS integration, example)
or Azure Blob store.

Detect and Extract Faces

We use the face_recognition library under the hood, powered by dlib’s CNN-based face detector. Since the model is slow on large images, we downscale wide images before detection.

@cocoindex.op.function(
    cache=True,
    behavior_version=1,
    gpu=True,
    arg_relationship=(cocoindex.op.ArgRelationship.RECTS_BASE_IMAGE, "content"),
)
def extract_faces(content: bytes) -> list[FaceBase]:
    orig_img = Image.open(io.BytesIO(content)).convert("RGB")

    # The model is too slow on large images, so we resize them if too large.
    if orig_img.width > MAX_IMAGE_WIDTH:
        ratio = orig_img.width * 1.0 / MAX_IMAGE_WIDTH
        img = orig_img.resize(
            (MAX_IMAGE_WIDTH, int(orig_img.height / ratio)),
            resample=Image.Resampling.BICUBIC,
        )
    else:
        ratio = 1.0
        img = orig_img

    # Extract face locations.
    locs = face_recognition.face_locations(np.array(img), model="cnn")

    faces: list[FaceBase] = []
    for min_y, max_x, max_y, min_x in locs:
        rect = ImageRect(
            min_x=int(min_x * ratio),
            min_y=int(min_y * ratio),
            max_x=int(max_x * ratio),
            max_y=int(max_y * ratio),
        )

        # Crop the face and save it as a PNG.
        buf = io.BytesIO()
        orig_img.crop((rect.min_x, rect.min_y, rect.max_x, rect.max_y)).save(
            buf, format="PNG"
        )
        face = buf.getvalue()
        faces.append(FaceBase(rect, face))

    return faces

We transform the image content:

with data_scope["images"].row() as image:
    image["faces"] = image["content"].transform(extract_faces)

After this step, each image has a list of detected faces and bounding boxes.

Each detected face is cropped from the original image and stored as a PNG.

Sample Extraction:

Compute Face Embeddings

We encode each cropped face using the same library. This generates a 128-dimensional vector representation per face.

@cocoindex.op.function(cache=True, behavior_version=1, gpu=True)
def extract_face_embedding(
    face: bytes,
) -> cocoindex.Vector[cocoindex.Float32, typing.Literal[128]]:
    """Extract the embedding of a face."""
    img = Image.open(io.BytesIO(face)).convert("RGB")
    embedding = face_recognition.face_encodings(
        np.array(img),
        known_face_locations=[(0, img.width - 1, img.height - 1, 0)],
    )[0]
    return embedding

We plug the embedding function into the flow:

with image["faces"].row() as face:
    face["embedding"] = face["image"].transform(extract_face_embedding)

After this step, we have embeddings ready to be indexed!

Collect and Export Embeddings

We now collect structured data for each face: filename, bounding box, and embedding.

face_embeddings = data_scope.add_collector()

face_embeddings.collect(
    id=cocoindex.GeneratedField.UUID,
    filename=image["filename"],
    rect=face["rect"],
    embedding=face["embedding"],
)

We export to a Qdrant collection:

face_embeddings.export(
    QDRANT_COLLECTION,
    cocoindex.targets.Qdrant(
        collection_name=QDRANT_COLLECTION
    ),
    primary_key_fields=["id"],
)

Now you can run cosine similarity queries over facial vectors.

CocoIndex supports 1-line switch with other vector databases like Postgres.

Query the Index

You can now build facial search apps or dashboards. For example:

Given a new face embedding, find the most similar faces
Find all face images that appear in a set of photos
Cluster embeddings to group visually similar people

For querying embeddings, check out Image Search project.

If you’d like to see a full example on the query path with image match, give it a shout at
our group.

Support Us

We’re constantly adding more examples and improving our runtime. If you found this helpful, please ⭐ star CocoIndex on GitHub and share it with others.

Thanks for reading!

Let us know what pipelines you’re building—we’d love to feature them.

Build Real-Time Codebase Indexing for AI Coding agents

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

In this blog, we will show you how to index a codebase for RAG with CocoIndex.
CocoIndex provides built-in support for codebase chunking, with native Tree-sitter support and real-time update.

Because we use incremental processing (only reprocessing what has changed) under the hood, the index is updated in near real-time.
It can be real-time if we integrate with any source that directly pushes change notifications, e.g., a code editor.
This can be built for code search applications, or for providing relevant context to AI code generation systems where low latency is critical for a seamless developer experience.

It'd mean a lot to us if you could ⭐ star CocoIndex on Github, if this tutorial helps you.

Tree-sitter

CocoIndex leverages Tree-sitter's capabilities to intelligently chunk code based on the actual syntax structure rather than arbitrary line breaks.
These syntactically coherent chunks are then used to build a more effective index for RAG systems, enabling more precise code retrieval and better context preservation.

Tree-sitter is a parser generator tool and an incremental parsing library,
it is available in Rust 🦀.

You can find the full code of this project here.
Only ~ 50 lines of Python on the indexing path, check it out 🚀!

Prerequisites

If you don't have Postgres installed, please follow installation guide.
CocoIndex uses Postgres to keep track of data lineage for incremental processing.

Define Indexing Flow

Flow Design

The flow diagram illustrates how we'll process our codebase:

Load code files from the local file system.
Determine the programming language by extracting file extensions.
Use Tree-sitter to parse and split code into semantically meaningful chunks.
Generate embeddings for each code chunk.
Store the embeddings in a vector database for efficient retrieval.

1. Add the codebase as a source.

@cocoindex.flow_def(name="CodeEmbedding")
def code_embedding_flow(flow_builder: cocoindex.FlowBuilder, data_scope: cocoindex.DataScope):
    """
    Define an example flow that embeds files into a vector database.
    """
    data_scope["files"] = flow_builder.add_source(
        cocoindex.sources.LocalFile(path="../..",
                                    included_patterns=["*.py", "*.rs", "*.toml", "*.md", "*.mdx"],
                                    excluded_patterns=[".*", "target", "**/node_modules"]))
    code_embeddings = data_scope.add_collector()

We ingest the CocoIndex codebase from the root directory.
We can change the path to the codebase you want to index.
We index all the files with the extensions of .py, .rs, .toml, .md, .mdx, and skip directories starting with ., target (in the root) and node_modules (under any directory).

flow_builder.add_source will create a table with sub fields (filename, content), we can refer to the documentation for more details.

2. Process each file and collect the information.

2.1 Extract the extension of a filename

We need to pass the language (or extension) to Tree-sitter to parse the code.
Let's define a function to extract the extension of a filename while processing each file.
You can find the documentation for custom function here.

@cocoindex.op.function()
def extract_extension(filename: str) -> str:
    """Extract the extension of a filename."""
    return os.path.splitext(filename)[1]

Then we are going to process each file and collect the information.

with data_scope["files"].row() as file:
    file["extension"] = file["filename"].transform(extract_extension)

Here we extract the extension of the filename and store it in the extension field.

2.2 Split the file into chunks

We will chunk the code with Tree-sitter.
We use the SplitRecursively function to split the file into chunks.
It is integrated with Tree-sitter, so you can pass in the language to the language parameter.
To see all supported language names and extensions, see the documentation here. All the major languages are supported, e.g., Python, Rust, JavaScript, TypeScript, Java, C++, etc. If it's unspecified or the specified language is not supported, it will be treated as plain text.

with data_scope["files"].row() as file:
    file["chunks"] = file["content"].transform(
          cocoindex.functions.SplitRecursively(),
          language=file["extension"], chunk_size=1000, chunk_overlap=300)

2.3 Embed the chunks

We use SentenceTransformerEmbed to embed the chunks.
You can refer to the documentation here.
There are 12k models supported by 🤗 Hugging Face.
You can just pick your favorite model.

@cocoindex.transform_flow()
def code_to_embedding(text: cocoindex.DataSlice[str]) -> cocoindex.DataSlice[list[float]]:
    """
    Embed the text using a SentenceTransformer model.
    """
    return text.transform(
        cocoindex.functions.SentenceTransformerEmbed(
            model="sentence-transformers/all-MiniLM-L6-v2"))

Then for each chunk, we will embed it using the code_to_embedding function. and collect the embeddings to the code_embeddings collector.

@cocoindex.transform_flow() is needed to share the transformation across indexing and query. We build a vector index and query against it,
the embedding computation needs to be consistent between indexing and querying. See documentation for more details.

with data_scope["files"].row() as file:
    with file["chunks"].row() as chunk:
        chunk["embedding"] = chunk["text"].call(code_to_embedding)
        code_embeddings.collect(filename=file["filename"], location=chunk["location"],
                                code=chunk["text"], embedding=chunk["embedding"])

2.4 Collect the embeddings

Export the embeddings to a table.

code_embeddings.export(
    "code_embeddings",
    cocoindex.storages.Postgres(),
    primary_key_fields=["filename", "location"],
    vector_indexes=[cocoindex.VectorIndex("embedding", cocoindex.VectorSimilarityMetric.COSINE_SIMILARITY)])

We use Consine Similarity to measure the similarity between the query and the indexed data.
To learn more about Consine Similarity, see Wiki.

3. Query the index

We match against user-provided text by a SQL query, reusing the embedding operation in the indexing flow.

def search(pool: ConnectionPool, query: str, top_k: int = 5):
    # Get the table name, for the export target in the code_embedding_flow above.
    table_name = cocoindex.utils.get_target_storage_default_name(code_embedding_flow, "code_embeddings")
    # Evaluate the transform flow defined above with the input query, to get the embedding.
    query_vector = code_to_embedding.eval(query)
    # Run the query and get the results.
    with pool.connection() as conn:
        with conn.cursor() as cur:
            cur.execute(f"""
                SELECT filename, code, embedding <=> %s::vector AS distance
                FROM {table_name} ORDER BY distance LIMIT %s
            """, (query_vector, top_k))
            return [
                {"filename": row[0], "code": row[1], "score": 1.0 - row[2]}
                for row in cur.fetchall()
            ]

Define a main function to run the query in terminal.

@cocoindex.main_fn()
def _run():
    # Initialize the database connection pool.
    pool = ConnectionPool(os.getenv("COCOINDEX_DATABASE_URL"))
    # Run queries in a loop to demonstrate the query capabilities.
    while True:
        try:
            query = input("Enter search query (or Enter to quit): ")
            if query == '':
                break
            # Run the query function with the database connection pool and the query.
            results = search(pool, query)
            print("\nSearch results:")
            for result in results:
                print(f"[{result['score']:.3f}] {result['filename']}")
                print(f"    {result['code']}")
                print("---")
            print()
        except KeyboardInterrupt:
            break

The @cocoindex.main_fn() decorator initializes the library with settings loaded from environment variables.
See documentation for initialization for more details.

Run the index setup & update

🎉 Now you are all set!

Run following commands to setup and update the index.

python main.py cocoindex setup
python main.py cocoindex update

You'll see the index updates state in the terminal

Test the query

At this point, you can start the CocoIndex server and develop your RAG runtime against the data. To test your index, you could

python main.py

When you see the prompt, you can enter your search query. for example: spec.

You can find the search results in the terminal

The returned results - each entry contains score (Cosine Similarity), filename, and the code snippet that get matched.

Support us

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

Thanks for reading!

Beyond Embeddings: Building Metadata-Rich Indexes from Academic PDFs

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

In this blog we will walk through a comprehensive example of indexing research papers with extracting different metadata —
beyond full text chunking and embedding — and build semantic embeddings for indexing and querying.

We would greatly appreciate it if you could ⭐ star CocoIndex on GitHub if you find this tutorial helpful.

Use Cases

Academic search and retrieval, as well as research-based AI agents
Paper recommendation systems
Research knowledge graphs
Semantic analysis of scientific literature

What we will achieve

Let's take a look at this PDF as an example.

Here’s what we want to accomplish:

Extract the paper metadata, including file name, title, author information, abstract, and number of pages.

Build vector embeddings for the metadata, such as the title and abstract, for semantic search. This enables better metadata-driven semantic search results. For example, you can match text queries against titles and abstracts.

Build an index of authors and all the file names associated with each author to answer questions like "Give me all the papers by Jeff Dean."

If you want to perform full PDF embedding for the paper, you can also refer to this article.

You can find the full code here.

Core Components

PDF Preprocessing
- Reads PDFs using pypdf and extracts:
  - Total number of pages
  - First page content (used as a proxy for metadata-rich information)
Markdown Conversion
- Converts the first page to Markdown using Marker.
LLM-Powered Metadata Extraction
- Sends the first-page Markdown to GPT-4o using CocoIndex's ExtractByLlm function.
- Extracted metadata includes and more.
  - title (string)
  - authors (with name, email, and affiliation)
  - abstract (string)
Semantic Embedding
- The title is embedded directly using the all-MiniLM-L6-v2 model by the SentenceTransformer.
- Abstracts are chunked based on semantic punctuation and token count, then each chunk is embedded individually.
Relational Data Collection
- Authors are unrolled and collected into an author_papers relation, enabling queries like:
  - Show all papers by X
  - Which co-authors worked with Y?

Prerequisites

Install PostgreSQL. CocoIndex uses PostgreSQL internally for incremental processing.
Configure your OpenAI API key. Alternatively, we have native support for Gemini, Ollama, LiteLLM, checkout the guide. You could choose your favorite LLM provider and can work completely on-premises.

Define Indexing Flow

This project demonstrates a slightly more comprehensive example of metadata understanding closer to real-world use cases.
You’ll see how easy it is to achieve this design by CocoIndex within 100 lines of indexing logic -
code.

To better help you navigate what we will walk through, here is a flow diagram.

Import a list of papers in PDF.
For each file:
- Extract the first page of the paper.
- Convert the first page to Markdown.
- Extract metadata (title, authors, abstract) from the first page.
- Split the abstract into chunks, and compute embeddings for each chunk.
Export to the following tables in Postgres with PGVector:
- Metadata (title, authors, abstract) for each paper.
- Author-to-paper mapping, for author-based query.
- Embeddings for titles and abstract chunks, for semantic search.

Let’s zoom in the steps.

Import the Papers

@cocoindex.flow_def(name="PaperMetadata")
def paper_metadata_flow(
    flow_builder: cocoindex.FlowBuilder, data_scope: cocoindex.DataScope
) -> None:
    data_scope["documents"] = flow_builder.add_source(
        cocoindex.sources.LocalFile(path="papers", binary=True),
        refresh_interval=datetime.timedelta(seconds=10),
    )

flow_builder.add_source will create a table with sub fields (filename, content),
we can refer to the documentation for more details.

Extract and collect metadata

Extract first page for basic info

Define a custom function to extract the first page and number of pages of the PDF.

@dataclasses.dataclass
class PaperBasicInfo:
    num_pages: int
    first_page: bytes

@cocoindex.op.function()
def extract_basic_info(content: bytes) -> PaperBasicInfo:
    """Extract the first pages of 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())

Now, plug this into your flow.
We extract metadata from the first page to minimize processing cost, since the entire PDF can be very large.

with data_scope["documents"].row() as doc:
    doc["basic_info"] = doc["content"].transform(extract_basic_info)

After this step, you should have the basic info of each paper.

Parse basic info

We will convert the first page to Markdown using Marker.
Alternatively, you can easily plug in your favorite PDF parser, such as Docling.

Define a marker converter function and cache it, since its initialization is resource-intensive.
This ensures that the same converter instance is reused for different input files.

@cache
def get_marker_converter() -> PdfConverter:
    config_parser = ConfigParser({})
    return PdfConverter(
        create_model_dict(), config=config_parser.generate_config_dict()
    )

Plug it into a custom function.

@cocoindex.op.function(gpu=True, cache=True, behavior_version=1)
def pdf_to_markdown(content: bytes) -> str:
    """Convert to Markdown."""

    with tempfile.NamedTemporaryFile(delete=True, suffix=".pdf") as temp_file:
        temp_file.write(content)
        temp_file.flush()
        text, _, _ = text_from_rendered(get_marker_converter()(temp_file.name))
        return text

Pass it to your transform

with data_scope["documents"].row() as doc:      
    doc["first_page_md"] = doc["basic_info"]["first_page"].transform(
            pdf_to_markdown
        )

After this step, you should have the first page of each paper in Markdown format.

Extract basic info with LLM

Define a schema for LLM extraction. CocoIndex natively supports LLM-structured extraction with complex and nested schemas.
If you are interested in learning more about nested schemas, refer to this article.

@dataclasses.dataclass
class PaperMetadata:
    """
    Metadata for a paper.
    """

    title: str
    authors: list[Author]
    abstract: str

Plug it into the ExtractByLlm function. With a dataclass defined, CocoIndex will automatically parse the LLM response into the dataclass.

doc["metadata"] = doc["first_page_md"].transform(
    cocoindex.functions.ExtractByLlm(
        llm_spec=cocoindex.LlmSpec(
            api_type=cocoindex.LlmApiType.OPENAI, model="gpt-4o"
        ),
        output_type=PaperMetadata,
        instruction="Please extract the metadata from the first page of the paper.",
    )
)

After this step, you should have the metadata of each paper.

Collect paper metadata

  paper_metadata = data_scope.add_collector()
  with data_scope["documents"].row() as doc:
    # ... process
    # Collect metadata
    paper_metadata.collect(
        filename=doc["filename"],
        title=doc["metadata"]["title"],
        authors=doc["metadata"]["authors"],
        abstract=doc["metadata"]["abstract"],
        num_pages=doc["basic_info"]["num_pages"],
    )

Just collect anything you need :)

Collect `author` to `filename` information

We’ve already extracted author list. Here we want to collect Author → Papers in a separate table to build a look up functionality.
Simply collect by author.

author_papers = data_scope.add_collector()

with data_scope["documents"].row() as doc:
    with doc["metadata"]["authors"].row() as author:
        author_papers.collect(
            author_name=author["name"],
            filename=doc["filename"],
        )

Compute and collect embeddings

Title

doc["title_embedding"] = doc["metadata"]["title"].transform(
    cocoindex.functions.SentenceTransformerEmbed(
        model="sentence-transformers/all-MiniLM-L6-v2"
    )
)

Abstract

Split abstract into chunks, embed each chunk and collect their embeddings.
Sometimes the abstract could be very long.

doc["abstract_chunks"] = doc["metadata"]["abstract"].transform(
    cocoindex.functions.SplitRecursively(
        custom_languages=[
            cocoindex.functions.CustomLanguageSpec(
                language_name="abstract",
                separators_regex=[r"[.?!]+\s+", r"[:;]\s+", r",\s+", r"\s+"],
            )
        ]
    ),
    language="abstract",
    chunk_size=500,
    min_chunk_size=200,
    chunk_overlap=150,
)

After this step, you should have the abstract chunks of each paper.

Embed each chunk and collect their embeddings.

with doc["abstract_chunks"].row() as chunk:
    chunk["embedding"] = chunk["text"].transform(
        cocoindex.functions.SentenceTransformerEmbed(
            model="sentence-transformers/all-MiniLM-L6-v2"
        )
    )

After this step, you should have the embeddings of the abstract chunks of each paper.

Collect embeddings

metadata_embeddings = data_scope.add_collector()

with data_scope["documents"].row() as doc:
    # ... process
    # collect title embedding
    metadata_embeddings.collect(
        id=cocoindex.GeneratedField.UUID,
        filename=doc["filename"],
        location="title",
        text=doc["metadata"]["title"],
        embedding=doc["title_embedding"],
    )
    with doc["abstract_chunks"].row() as chunk:
        # ... process
        # collect abstract chunks embeddings
        metadata_embeddings.collect(
            id=cocoindex.GeneratedField.UUID,
            filename=doc["filename"],
            location="abstract",
            text=chunk["text"],
            embedding=chunk["embedding"],
        )

Export

Finally, we export the data to Postgres.

paper_metadata.export(
    "paper_metadata",
    cocoindex.targets.Postgres(),
    primary_key_fields=["filename"],
)
author_papers.export(
    "author_papers",
    cocoindex.targets.Postgres(),
    primary_key_fields=["author_name", "filename"],
)    
metadata_embeddings.export(
    "metadata_embeddings",
    cocoindex.targets.Postgres(),
    primary_key_fields=["id"],
    vector_indexes=[
        cocoindex.VectorIndexDef(
            field_name="embedding",
            metric=cocoindex.VectorSimilarityMetric.COSINE_SIMILARITY,
        )
    ],
)

In this example we use PGVector as embedding stores/
With CocoIndex, you can do one line switch on other supported Vector databases like Qdrant, see this guide for more details.
We aim to standardize interfaces and make it like building LEGO.

View in CocoInsight step by step

You can walk through the project step by step in CocoInsight to see
exactly how each field is constructed and what happens behind the scenes.

Query the index

You can refer to this section of Text Embeddings about
how to build query against embeddings.
For now CocoIndex doesn't provide additional query interface. We can write SQL or rely on the query engine by the target storage.

Many databases already have optimized query implementations with their own best practices
The query space has excellent solutions for querying, reranking, and other search-related functionality.

If you need assist with writing the query, please feel free to reach out to us at Discord.

Support us

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

Thanks for reading!

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

🚀 CocoIndex https://github.com/cocoindex-io/cocoindex Now Supports Amazon S3 for Native, Real-Time Incremental Data Processing

We’re excited to announce a major update to CocoIndex:
You can now natively connect Amazon S3 as a data source—and, when combined with AWS Simple Queue Service (SQS), CocoIndex enables true real-time, incremental processing of your data as it’s updated.

This upgrade makes CocoIndex a seamless drop-in for cloud-native teams that rely on S3 for storage and want to power real-time, up-to-date, AI-driven experiences—without reprocessing the world every time a file changes.

We are constantly improving and more features and examples are coming! Stay tuned with latest updates from CocoIndex by star ⭐ our github repo: https://github.com/cocoindex-io/cocoindex

Why Incremental Processing?

Traditional data pipelines often rely on batch processing—recomputing everything at a fixed interval. That’s fine at small scales. But once your data grows—or freshness becomes critical—that model starts to break down.

Incremental processing means focusing only on what’s changed: the new or modified data since the last run. This shift becomes essential in a world where data volume is growing, compute costs are rising, and users (or AI systems) demand the most up-to-date information.

When Incremental Processing Truly Matters

Here are a few key scenarios where this shift isn’t just a nice-to-have—it’s a necessity:

Freshness is Critical

When your app is user-facing, data freshness directly affects trust and usability. If someone updates a document, they expect that update to be reflected instantly in search results or AI responses.

Stale data doesn’t just break UX—it can lead to misleading outputs from downstream systems, especially AI. The worst part? Users often don’t realize the data is outdated until it’s too late.

Compute is Expensive

AI-driven systems often require heavy transformations: generating embeddings, fine-tuning models, cleaning unstructured data, etc.

Rerunning those processes on unchanged data leads to wasted cycles and bloated bills. Incremental processing avoids this waste by doing only the necessary work.

Working at Scale

Reprocessing terabytes—or even petabytes—of data every time there's a small change is simply not sustainable. CocoIndex makes scale manageable by processing only what’s necessary.

What Makes CocoIndex’s Approach Unique?

With the new S3 and SQS support, CocoIndex brings you:

Optimized Resource Utilization

We scan for new or updated S3 files only—no need to poll or manually manage sync logic. Compute cycles are focused on just the data that changed.

True Real-Time Processing with AWS SQS

CocoIndex listens to S3 change events via SQS queues. This means that as soon as a file is uploaded or modified, it can trigger downstream transformation—without waiting for a batch job or cron schedule.

Built-In State & Lineage Tracking

CocoIndex maintains a persistent internal state. It remembers what’s already been processed and tracks how each file (or chunk) was transformed. That enables reproducibility and auditability across runs.

Fine-Grained Smart Caching

Instead of reprocessing entire files, CocoIndex dives deeper. Files are broken into logical chunks. If only a subset of those chunks change, we only recompute the affected parts.

This is especially powerful in workflows like vector indexing, where only modified sections need new embeddings.

Real-Time Data Processing for the AI Era

If your systems depend on accurate, timely data—and your datasets are too large to rebuild every few hours—incremental, event-driven processing isn’t a luxury. It’s the only path forward.

With this new S3 + SQS integration, CocoIndex makes it easy to bring your existing cloud storage into a real-time, intelligent, and cost-efficient processing workflow—out of the box.

What Is Amazon SQS (Simple Queue Service)?

Amazon SQS (Simple Queue Service) is a fully managed message queuing service from AWS that helps you decouple, scale, and streamline communication between microservices, distributed systems, and serverless applications.

With Amazon SQS, developers can build reliable, fault-tolerant systems by using scalable, secure message queues that temporarily store messages as they move between different components of a cloud-native application.

Designed for high availability and performance, SQS supports both standard queues (high throughput, at-least-once delivery) and FIFO queues (exactly-once processing, ordered messages), making it ideal for a wide range of use cases, from real-time data pipelines to asynchronous task processing.

When files are uploaded to or modified in S3, SQS receives notifications about these changes and queues them as messages. Each message contains metadata about the S3 event, such as:

The type of event (e.g., ObjectCreated, ObjectRemoved)

The S3 bucket name
The object key (file path)
Timestamp of the event
Other relevant metadata

These messages remain in the queue until they are successfully processed, ensuring no events are lost even if the processing system is temporarily unavailable.

Live update out of the box with SQS

CocoIndex provides two modes to run your pipeline, one time update and live update, both leverage the incremental processing. Particularly with AWS SQS, you could leverage the live update mode -
where CocoIndex continuously monitors and reacts to the events in SQS, updating the target data in real-time. This is ideal for use cases where data freshness is critical.

How does it work?

Let's take a look at simple example of how to build a real-time data transformation pipeline with S3 and CocoIndex. It builds a vector database of text embeddings from markdown files in S3. You could find a similar example to process local files in this blog.

S3 bucket and SQS setup

Please follow the documentation here to setup S3 bucket and SQS queue.

S3 bucket

Creating an AWS account.
Configuring IAM permissions.
Configure policies. You'll need at least the AmazonS3ReadOnlyAccess policy, and if you want to enable change notifications, you'll also need the AmazonSQSFullAccess policy.

SQS queue

For real-time change detection, you'll need to create an SQS queue and configure it to receive notifications from your S3 bucket.
Please follow the documentation to configure the S3 bucket to send event notifications to the SQS queue.

Particularly, the SQS queue needs a specific access policy that allows S3 to send messages to it.

{
  ...
  "Statement": [
    ...
    {
      "Sid": "__publish_statement",
      "Effect": "Allow",
      "Principal": {
        "Service": "s3.amazonaws.com"
      },
      "Resource": "${SQS_QUEUE_ARN}",
      "Action": "SQS:SendMessage",
      "Condition": {
        "ArnLike": {
          "aws:SourceArn": "${S3_BUCKET_ARN}"
        }
      }
    }
  ]
}

Then you can upload your files to the S3 bucket.

Define Indexing Flow

Flow Design

The flow diagram illustrates how we'll process our codebase:

Read text files from the Amazon S3 bucket
Chunk each document
For each chunk, embed it with a text embedding model
Store the embeddings in a vector database for retrieval

AWS File Ingestion

Define the AWS endpoint and the SQS queue name in .env file:

# Database Configuration
DATABASE_URL=postgresql://localhost:5432/cocoindex

# Amazon S3 Configuration
AMAZON_S3_BUCKET_NAME=your-bucket-name
AMAZON_S3-SQS_QUEUE_URL=https://sqs.us-west-2.amazonaws.com/123456789/S3ChangeNotifications

Define indexing flow and ingest from Amazon S3 SQS queue:

@cocoindex.flow_def(name="AmazonS3TextEmbedding")
def amazon_s3_text_embedding_flow(
    flow_builder: cocoindex.FlowBuilder, data_scope: cocoindex.DataScope
):
    bucket_name = os.environ["AMAZON_S3_BUCKET_NAME"]
    prefix = os.environ.get("AMAZON_S3_PREFIX", None)
    sqs_queue_url = os.environ.get("AMAZON_S3_SQS_QUEUE_URL", None)

    data_scope["documents"] = flow_builder.add_source(
        cocoindex.sources.AmazonS3(
            bucket_name=bucket_name,
            prefix=prefix,
            included_patterns=["*.md", "*.mdx", "*.txt", "*.docx"],
            binary=False,
            sqs_queue_url=sqs_queue_url,
        )
    )

This defines a flow that reads text files from the Amazon S3 bucket.

Rest of the flow

For the rest of the flow, we can follow the tutorial in this blog.
The entire project is available here.

Run the flow with live update

cocoindex update main.py -L

-L option means live update, see the documentation for more details.
And you will have a continuous long running process that will update the vector database with any updates in the S3 bucket.

If you need to run the server and indexing in the same process, without spawning two separate jobs, you can add the FlowLiveUpdater in the main() function, see the example here.
And you can run the server with python3 main.py.

Support us

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

Thanks for reading!

How to build image search with semantic understanding

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

In this blog, we will build live image search and query it with natural language.
For example, you can search for "an elephant", or a "cute animal" with a list of images as input.

We are going to use multimodal embedding model to understand and embed the image, and build a vector index for efficient retrieval. We are going use CocoIndex to build the indexing flow, it is a ultra performant real-time data transformation framework. During running,
you can add new files to the folder and it only process changed files and will be indexed within a minute.

It'd mean a lot to us if you could drop a star at CocoIndex on Github, if this tutorial is helpful.

🚀 You can find the full code of this project here.

Technologies

CocoIndex

CocoIndex is an ultra performant real-time data transformation framework for AI.

CLIP ViT-L/14

CLIP ViT-L/14 is a powerful vision-language model that can understand both images and texts.
It's trained to align visual and textual representations in a shared embedding space, making it perfect for our image search use case.

In our project, we use CLIP to:

Generate embeddings of the images directly
Convert natural language search queries into the same embedding space
Enable semantic search by comparing query embeddings with caption embeddings

Qdrant

Qdrant is a high performance vector database. We use it to store and query the embeddings.

FastAPI

FastAPI is a modern, fast (high-performance), web framework for building APIs with Python 3.7+ based on standard Python type hints.
We use it to build the web API for the image search.

Prerequisites

Install Postgres. CocoIndex uses Postgres to keep track of data lineage for incremental processing.
Install Qdrant.

Define Indexing Flow

Flow Design

The flow diagram illustrates how we'll process our codebase:

Read image files from the local filesystem
Use CLIP to understand and embed the image
Store the embeddings in a vector database for retrieval

1. Ingest the images.

@cocoindex.flow_def(name="ImageObjectEmbedding")
def image_object_embedding_flow(flow_builder: cocoindex.FlowBuilder, data_scope: cocoindex.DataScope):
    data_scope["images"] = flow_builder.add_source(
        cocoindex.sources.LocalFile(path="img", included_patterns=["*.jpg", "*.jpeg", "*.png"], binary=True),
        refresh_interval=datetime.timedelta(minutes=1)  # Poll for changes every 1 minute
    )
    img_embeddings = data_scope.add_collector()

flow_builder.add_source will create a table with sub fields (filename, content), we can refer to the documentation for more details.

2. Process each image and collect the information.

2.1 Embed the image with CLIP

@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

The @functools.cache decorator caches the results of a function call. In this case, it ensures that we only load the CLIP model and processor once.

@cocoindex.op.function(cache=True, behavior_version=1, gpu=True)
def embed_image(img_bytes: bytes) -> cocoindex.Vector[cocoindex.Float32, Literal[384]]:
    """
    Convert image to embedding using CLIP model.
    """
    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():
        features = model.get_image_features(**inputs)
    return features[0].tolist()

embed_image is a custom function that uses the CLIP model to convert an image into a vector embedding.
It accepts image data in bytes format and returns a list of floating-point numbers representing the image's embedding.

The function supports caching through the cache parameter.
When enabled, the executor will store the function's results for reuse during reprocessing,
which is particularly useful for computationally intensive operations.
For more information about custom function parameters, refer to the documentation.

Then we are going to process each image and collect the information.

with data_scope["images"].row() as img:
    img["embedding"] = img["content"].transform(embed_image)
    img_embeddings.collect(
        id=cocoindex.GeneratedField.UUID,
        filename=img["filename"],
        embedding=img["embedding"],
    )

2.3 Collect the embeddings

Export the embeddings to a table in Qdrant.

img_embeddings.export(
    "img_embeddings",
    cocoindex.storages.Qdrant(
        collection_name="image_search",
        grpc_url=QDRANT_GRPC_URL,
    ),
    primary_key_fields=["id"],
    setup_by_user=True,
)

3. Query the index

Embed the query with CLIP, which maps both text and images into the same embedding space, allowing for cross-modal similarity search.

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():
        features = model.get_text_features(**inputs)
    return features[0].tolist()

Defines a FastAPI endpoint /search that performs semantic image search.

@app.get("/search")
def search(q: str = Query(..., description="Search query"), limit: int = Query(5, description="Number of results")):
    # Get the embedding for the query
    query_embedding = embed_query(q)

    # Search in Qdrant
    search_results = app.state.qdrant_client.search(
        collection_name="image_search",
        query_vector=("embedding", query_embedding),
        limit=limit
    )

This searches the Qdrant vector database for similar embeddings. Returns the top limit results

# Format results
out = []
for result in search_results:
    out.append({
        "filename": result.payload["filename"],
        "score": result.score
    })
return {"results": out}

This endpoint enables semantic image search where users can find images by describing them in natural language,
rather than using exact keyword matches.

Application

Fast API

app = FastAPI()
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)
# Serve images from the 'img' directory at /img
app.mount("/img", StaticFiles(directory="img"), name="img")

FastAPI application setup with CORS middleware and static file serving
The app is configured to:

Allow cross-origin requests from any origin
Serve static image files from the 'img' directory
Handle API endpoints for image search functionality

@app.on_event("startup")
def startup_event():
    load_dotenv()
    cocoindex.init()
    # Initialize Qdrant client
    app.state.qdrant_client = QdrantClient(
        url=QDRANT_GRPC_URL,
        prefer_grpc=True
    )
    app.state.live_updater = cocoindex.FlowLiveUpdater(image_object_embedding_flow)
    app.state.live_updater.start()

The startup event handler initializes the application when it first starts up. Here's what each part does:

load_dotenv(): Loads environment variables from a .env file, which is useful for configuration like API keys and URLs
cocoindex.init(): Initializes the CocoIndex framework, setting up necessary components and configurations
Qdrant Client Setup:
- Creates a new QdrantClient instance
- Configures it to use the gRPC URL specified in environment variables
- Enables gRPC preference for better performance
- Stores the client in the FastAPI app state for access across requests
Live Updater Setup:
- Creates a FlowLiveUpdater instance for the image_object_embedding_flow
- This enables real-time updates to the image search index
- Starts the live updater to begin monitoring for changes

This initialization ensures that all necessary components are properly configured and running when the application starts.

Frontend

you can check the frontend code here.
We intentionally kept it simple and minimalistic to focus on the image search functionality.

Time to have fun!

Create a collection in Qdrant

curl -X PUT 'http://localhost:6333/collections/image_search' \
-H 'Content-Type: application/json' \
-d '{
    "vectors": {
    "embedding": {
        "size": 768,
        "distance": "Cosine"
    }
    }
}'

Setup indexing flow
```
cocoindex setup main.py
```
It is setup with a live updater, so you can add new files to the folder and it will be indexed within a minute.

Run backend

uvicorn main:app --reload --host 0.0.0.0 --port 8000

Run frontend
```
cd frontend
npm install
npm run dev
```

Go to http://localhost:5174 to search.

Now add another image in the img folder, for example, this cute squirrel, or any picture you like.
Wait a minute for the new image to be processed and indexed.

If you want to monitor the indexing progress, you can view it in CocoInsight cocoindex server -ci main.py.

Acknowledgement

This image search project is largely contributed by @par4m - Param Arora ❤️. He is currently a open source developer at Google Summer of Code.

Support us

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

Thanks for reading!

How to build index with text embeddings

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

In this blog, we will build index with text embeddings and query it with natural language. We try to keep it minimalistic and focus on the gist of the indexing flow.

It'd mean a lot to us if you could drop a star at CocoIndex on Github, if this tutorial is helpful.

🚀 You can find the full code of this project here or play it on Colab.

Prerequisites

Install Postgres. CocoIndex uses Postgres to keep track of data lineage for incremental processing.

Define Indexing Flow

Flow Design

The flow diagram illustrates how we'll process our codebase:

Read text files from the local filesystem
Chunk each document
For each chunk, embed it with a text embedding model
Store the embeddings in a vector database for retrieval

1. Ingest the files

@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()

flow_builder.add_source will create a table with sub fields (filename, content), we can refer to the documentation for more details.

2. Process each file and collect the embeddings

2.1 Chunk the file

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

2.2 Embed each chunk

@cocoindex.transform_flow()
def text_to_embedding(text: cocoindex.DataSlice[str]) -> cocoindex.DataSlice[list[float]]:
    """
    Embed the text using a SentenceTransformer model.
    This is a shared logic between indexing and querying, so extract it as a function.
    """
    return text.transform(
        cocoindex.functions.SentenceTransformerEmbed(
            model="sentence-transformers/all-MiniLM-L6-v2"))

This code defines a transformation function that converts text into vector embeddings using the SentenceTransformer model.
@cocoindex.transform_flow() is needed to share the transformation across indexing and query.
This decorator marks this as a reusable transformation flow that can be called on specific input data from user code using eval(), as shown in the search function below.

The function uses CocoIndex's built-in SentenceTransformerEmbed function to convert the input text into 384-dimensional embeddings
The MiniLM-L6-v2 model is a good balance of speed and quality for text embeddings, though you can swap in other SentenceTransformer models as needed.

with doc["chunks"].row() as chunk:
    chunk["embedding"] = text_to_embedding(chunk["text"])
    doc_embeddings.collect(filename=doc["filename"], location=chunk["location"],
                            text=chunk["text"], embedding=chunk["embedding"])

2.3 Export the embeddings

Export the embeddings to a table in Postgres.

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. Query the index

CocoIndex doesn't provide additional query interface. We can write SQL or rely on the query engine by the target storage, if any.

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

Setup main() for interactive query in terminal.

def _main():
    # Initialize the database connection pool.
    pool = ConnectionPool(os.getenv("COCOINDEX_DATABASE_URL"))
    # Run queries in a loop to demonstrate the query capabilities.
    while True:
        query = input("Enter search query (or Enter to quit): ")
        if query == '':
            break
        # Run the query function with the database connection pool and the query.
        results = search(pool, query)
        print("\nSearch results:")
        for result in results:
            print(f"[{result['score']:.3f}] {result['filename']}")
            print(f"    {result['text']}")
            print("---")
        print()

if __name__ == "__main__":
    load_dotenv()
    cocoindex.init()
    _main()

Time to have fun!

Run following commands to setup and update the index.

python main.py cocoindex setup
python main.py cocoindex update

Start the interactive query in terminal.
```
python main.py
```

Support us

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

Thanks for reading!

🎉 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

In this blog, we will build a knowledge graph for product recommendations using taxonomy and complementary taxonomy. CocoIndex makes it easy to build and maintain knowledge graphs with continuous source updates.

Product taxonomy is a way to organize product catalogs in a logical and hierarchical structure; a great detailed explanation can be found here. In practice, it is a complicated problem: a product can be part of multiple categories, and a category can have multiple parents.

We will use LLM to extract a list of taxonomies from product descriptions. We will also use LLM to generate a list of complementary taxonomies for each product - for example, when someone buys a notepad, they may also buy a pen as a complementary product. We'll use Knowledge Graph to explore the relationships between products that can be further used for product recommendations or labeling.

The source code is available at CocoIndex Examples - product_taxonomy.

We are constantly improving, and more features and examples are coming soon.
Stay tuned and follow our progress by starring our GitHub repo.

Prerequisites

Install PostgreSQL. CocoIndex uses PostgreSQL internally for incremental processing.
Install Neo4j, a graph database.
Configure your OpenAI API key. Alternatively, you can switch to Ollama, which runs LLM models locally - guide.

Documentation

You can read the official CocoIndex Documentation for Property Graph Targets here.

Data flow to build knowledge graph

Overview

The core flow is about ~100 lines of python code

We are going to declare a data flow

ingest products (in JSON)
for each product,
- parse JSON
- map & clean up data
- extract taxonomy from the mapped data
collect data
export data to neo4j

Add documents as source

@cocoindex.flow_def(name="StoreProduct")
def store_product_flow(flow_builder: cocoindex.FlowBuilder, data_scope: cocoindex.DataScope):
    data_scope["products"] = flow_builder.add_source(
        cocoindex.sources.LocalFile(path="products",
                                    included_patterns=["*.json"]),
        refresh_interval=datetime.timedelta(seconds=5))

Here flow_builder.add_source creates a KTable.
filename is the key of the KTable.

Add data collectors

Add collectors at the root scope to collect the product, taxonomy and complementary taxonomy.

product_node = data_scope.add_collector()
product_taxonomy = data_scope.add_collector()
product_complementary_taxonomy = data_scope.add_collector()

Process each product

We will parse the JSON file for each product, and transform the data to the format that we need for downstream processing.

Data Mapping

@cocoindex.op.function(behavior_version=2)
def extract_product_info(product: cocoindex.typing.Json, filename: str) -> ProductInfo:
    return ProductInfo(
        id=f"{filename.removesuffix('.json')}",
        url=product["source"],
        title=product["title"],
        price=float(product["price"].lstrip("$").replace(",", "")),
        detail=Template(PRODUCT_TEMPLATE).render(**product),
    )

Here we define a function for data mapping, e.g.,

clean up the id field
map title -> title
clean up the price field
generate a markdown string for the product detail based on all the fields (for LLM to extract taxonomy and complementary taxonomy, we find that markdown works best as context for LLM).

Flow

Within the flow, we plug in the data mapping transformation to process each product JSON.

with data_scope["products"].row() as product:
    data = (product["content"]
            .transform(cocoindex.functions.ParseJson(), language="json")
            .transform(extract_product_info, filename=product["filename"]))
    product_node.collect(id=data["id"], url=data["url"], title=data["title"], price=data["price"])

The first transform() parses the JSON file.
The second transform() performs the defined data mapping.
We collect the fields we need for the product node in Neo4j.

Extract taxonomy and complementary taxonomy using LLM

Product Taxonomy Definition

Since we are using LLM to extract product taxonomy, we need to provide a detailed instruction at the class-level docstring.

@dataclasses.dataclass
class ProductTaxonomy:
    """
    Taxonomy for the product.

    A taxonomy is a concise noun (or short noun phrase), based on its core functionality, without specific details such as branding, style, etc.

    Always use the most common words in US English.

    Use lowercase without punctuation, unless it's a proper noun or acronym.

    A product may have multiple taxonomies. Avoid large categories like "office supplies" or "electronics". Use specific ones, like "pen" or "printer".
    """
    name: str

Define Product Taxonomy Info

Basically we want to extract all possible taxonomies for a product, and think about what other products are likely to be bought together with the current product.

@dataclasses.dataclass
class ProductTaxonomyInfo:
    """
    Taxonomy information for the product.

    Fields:
    - taxonomies: Taxonomies for the current product.
    - complementary_taxonomies: Think about when customers buy this product, what else they might need as complementary products. Put labels for these complentary products.
    """
    taxonomies: list[ProductTaxonomy]
    complementary_taxonomies: list[ProductTaxonomy]

For each product, we want some insight about its taxonomy and complementary taxonomy and we could use that as bridge to find related product using knowledge graph.

LLM Extraction

Finally, we will use cocoindex.functions.ExtractByLlm to extract the taxonomy and complementary taxonomy from the product detail.

taxonomy = data["detail"].transform(cocoindex.functions.ExtractByLlm(
            llm_spec=cocoindex.LlmSpec(
                api_type=cocoindex.LlmApiType.OPENAI, model="gpt-4.1"),
                output_type=ProductTaxonomyInfo))

For example, LLM takes the description of the gel pen, and extracts taxonomy to be gel pen.
Meanwhile, it suggests that when people buy gel pen, they may also be interested in notebook etc as complimentary taxonomy.

And then we will collect the taxonomy and complementary taxonomy to the collector.

with taxonomy['taxonomies'].row() as t:
    product_taxonomy.collect(id=cocoindex.GeneratedField.UUID, product_id=data["id"], taxonomy=t["name"])
with taxonomy['complementary_taxonomies'].row() as t:
    product_complementary_taxonomy.collect(id=cocoindex.GeneratedField.UUID, product_id=data["id"], taxonomy=t["name"])

Build knowledge graph

Basic concepts

All nodes for Neo4j need two things:

Label: The type of the node. E.g., Product, Taxonomy.
Primary key field: The field that uniquely identifies the node. E.g., id for Product nodes.

CocoIndex uses the primary key field to match the nodes and deduplicate them. If you have multiple nodes with the same primary key, CocoIndex keeps only one of them.

There are two ways to map nodes:

When you have a collector just for the node, you can directly export it to Neo4j. For example Product. We've collected each product explicitly.
When you have a collector for relationships connecting to the node, you can map nodes from selected fields in the relationship collector. You must declare a node label and primary key field.

For example,

product_taxonomy.collect(id=cocoindex.GeneratedField.UUID, product_id=data["id"], taxonomy=t["name"])

Collects a relationship, and taxonomy node is created from the relationship.

Configure Neo4j connection:

conn_spec = cocoindex.add_auth_entry(
    "Neo4jConnection",
    cocoindex.storages.Neo4jConnection(
        uri="bolt://localhost:7687",
        user="neo4j",
        password="cocoindex",
))

Export `Product` nodes to Neo4j

product_node.export(
    "product_node",
    cocoindex.storages.Neo4j(
        connection=conn_spec,
        mapping=cocoindex.storages.Nodes(label="Product")
    ),
    primary_key_fields=["id"],
)

This exports Neo4j nodes with label Product from the product_node collector.

It declares Neo4j node label Product. It specifies id as the primary key field.
It carries all the fields from product_node collector to Neo4j nodes with label Product.

Export `Taxonomy` nodes to Neo4j

We don't have explicit collector for Taxonomy nodes.
They are part of the product_taxonomy and product_complementary_taxonomy collectors and fields are collected during the taxonomy extraction.

To export them as Neo4j nodes, we need to first declare Taxonomy nodes.

flow_builder.declare(
    cocoindex.storages.Neo4jDeclaration(
        connection=conn_spec,
        nodes_label="Taxonomy",
        primary_key_fields=["value"],
    )
)

Next, export the product_taxonomy as relationship to Neo4j.

product_taxonomy.export(
    "product_taxonomy",
    cocoindex.storages.Neo4j(
        connection=conn_spec,
        mapping=cocoindex.storages.Relationships(
            rel_type="PRODUCT_TAXONOMY",
            source=cocoindex.storages.NodeFromFields(
                label="Product",
                fields=[
                    cocoindex.storages.TargetFieldMapping(
                        source="product_id", target="id"),
                ]
            ),
            target=cocoindex.storages.NodeFromFields(
                label="Taxonomy",
                fields=[
                    cocoindex.storages.TargetFieldMapping(
                        source="taxonomy", target="value"),
                ]
            ),
        ),
    ),
    primary_key_fields=["id"],
)

Similarly, we can export the product_complementary_taxonomy as relationship to Neo4j.

    product_complementary_taxonomy.export(
        "product_complementary_taxonomy",
        cocoindex.storages.Neo4j(
            connection=conn_spec,
            mapping=cocoindex.storages.Relationships(
                rel_type="PRODUCT_COMPLEMENTARY_TAXONOMY",
                source=cocoindex.storages.NodeFromFields(
                    label="Product",
                    fields=[
                        cocoindex.storages.TargetFieldMapping(
                            source="product_id", target="id"),
                    ]
                ),
                target=cocoindex.storages.NodeFromFields(
                    label="Taxonomy",
                    fields=[
                        cocoindex.storages.TargetFieldMapping(
                            source="taxonomy", target="value"),
                    ]
                ),
            ),
        ),
        primary_key_fields=["id"],
    )

The cocoindex.storages.Relationships declares how to map relationships in Neo4j.

In a relationship, there's:

A source node and a target node.
A relationship connecting the source and target. Note that different relationships may share the same source and target nodes.

NodeFromFields takes the fields from the entity_relationship collector and creates Taxonomy nodes.

Main function

Finally, the main function for the flow initializes the CocoIndex flow and runs it.

 @cocoindex.main_fn()
def _run():
    pass

if __name__ == "__main__":
    load_dotenv(override=True)
    _run()

Query and test your index

🎉 Now you are all set!

Install the dependencies:
```
pip install -e .
```
Run following commands to setup and update the index.
```
python main.py cocoindex setup
python main.py cocoindex update
```
You'll see the index updates state in the terminal. For example, you'll see the following output:
```
documents: 9 added, 0 removed, 0 updated
```
(Optional) I used CocoInsight to troubleshoot the index generation and understand the data lineage of the pipeline.
It is in free beta now, you can give it a try. Run following command to start CocoInsight:
```
python3 main.py cocoindex server -c https://cocoindex.io
```
And then open the url https://cocoindex.io/cocoinsight. It just connects to your local CocoIndex server, with Zero pipeline data retention.

Browse the knowledge graph

After the knowledge graph is built, you can explore the knowledge graph you built in Neo4j Browser.

For the dev environment, you can connect to Neo4j browser using credentials:

username: Neo4j
password: cocoindex which is pre-configured in our docker compose config.yaml.

You can open it at http://localhost:7474, and run the following Cypher query to get all relationships:

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

Support us

We are constantly improving, and more features and examples are coming soon.
If you love this article, please give us a star ⭐ at GitHub repo to help us grow.

Thanks for reading!

Build Real-Time Knowledge Graph For Documents with LLM

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

CocoIndex makes it easy to build and maintain knowledge graphs with continuous source updates. In this blog, we will process a list of documents (using CocoIndex documentation as an example). We will use LLM to extract relationships between the concepts in each document. We will generate two kinds of relationships:

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 source code is available at CocoIndex Examples - docs_to_knowledge_graph.

We are constantly improving, and more features and examples are coming soon.

Stay tuned and follow our progress by starring our GitHub repo.

Prerequisites

Install PostgreSQL. CocoIndex uses PostgreSQL internally for incremental processing.
Install Neo4j, a graph database.
Configure your OpenAI API key. Alternatively, you can switch to Ollama, which runs LLM models locally - guide.

Documentation

You can read the official CocoIndex Documentation for Property Graph Targets here.

Data flow to build knowledge graph

Add documents as source

We will process CocoIndex documentation markdown files (.md, .mdx) from the docs/core directory (markdown files, deployed docs).

@cocoindex.flow_def(name="DocsToKG")
def docs_to_kg_flow(flow_builder: cocoindex.FlowBuilder, data_scope: cocoindex.DataScope):
    data_scope["documents"] = flow_builder.add_source(
        cocoindex.sources.LocalFile(path="../../docs/docs/core",
                                    included_patterns=["*.md", "*.mdx"]))

Here flow_builder.add_source creates a KTable.
filename is the key of the KTable.

Add data collectors

Add collectors at the root scope:

document_node = data_scope.add_collector()
entity_relationship = data_scope.add_collector()
entity_mention = data_scope.add_collector()

document_node collects documents. E.g., core/basics.mdx is a document.
entity_relationship collects relationships. E.g., "CocoIndex supports Incremental Processing" indicates a relationship between CocoIndex and Incremental Processing.
entity_mention collects mentions of entities in a document. E.g., core/basics.mdx mentions CocoIndex and Incremental Processing.

Process each document and extract summary

Define a DocumentSummary data class to extract the summary of a document.

@dataclasses.dataclass
class DocumentSummary:
    title: str
    summary: str

Within the flow, use cocoindex.functions.ExtractByLlm for structured output.

with data_scope["documents"].row() as doc:
    doc["summary"] = doc["content"].transform(
            cocoindex.functions.ExtractByLlm(
                llm_spec=cocoindex.LlmSpec(
                    api_type=cocoindex.LlmApiType.OPENAI, model="gpt-4o"),
                output_type=DocumentSummary,
                instruction="Please summarize the content of the document."))

    document_node.collect(
        filename=doc["filename"], title=doc["summary"]["title"],
        summary=doc["summary"]["summary"])

doc["summary"] adds a new column to the KTable data_scope["documents"].

Extract relationships from the document using LLM

Define a data class to represent relationship for the LLM 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

The Data class defines a knowledge graph relationship. We recommend putting detailed instructions in the class-level docstring to help the LLM extract relationships correctly.

subject: Represents the entity the statement is about (e.g., 'CocoIndex').
predicate: Describes the type of relationship or property connecting the subject and object (e.g., 'supports').
object: Represents the entity or value that the subject is related to via the predicate (e.g., 'Incremental Processing').

This structure represents facts like "CocoIndex supports Incremental Processing". Its graph representation is:

Next, we will use cocoindex.functions.ExtractByLlm to extract the relationships from the document.

doc["relationships"] = doc["content"].transform(
    cocoindex.functions.ExtractByLlm(
        llm_spec=cocoindex.LlmSpec(
            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. "
        )
    )
)

doc["relationships"] adds a new field relationships to each document. output_type=list[Relationship] specifies that the output of the transformation is a LTable.

Collect relationships

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"],
    )
    # mention of an entity in a document, for subject
    entity_mention.collect(
        id=cocoindex.GeneratedField.UUID, entity=relationship["subject"],
        filename=doc["filename"],
    )
    # mention of an entity in a document, for object
    entity_mention.collect(
        id=cocoindex.GeneratedField.UUID, entity=relationship["object"],
        filename=doc["filename"],
    )

entity_relationship collects relationships between subjects and objects.
entity_mention collects mentions of entities (as subjects or objects) in the document separately. For example, core/basics.mdx has a sentence CocoIndex supports Incremental Processing. We want to collect:
- core/basics.mdx mentions CocoIndex.
- core/basics.mdx mentions Incremental Processing.

Build knowledge graph

Basic concepts

All nodes for Neo4j need two things:

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

CocoIndex uses the primary key field to match the nodes and deduplicate them. If you have multiple nodes with the same primary key, CocoIndex keeps only one of them.

There are two ways to map nodes:

When you have a collector just for the node, you can directly export it to Neo4j.
When you have a collector for relationships connecting to the node, you can map nodes from selected fields in the relationship collector. You must declare a node label and primary key field.

Configure Neo4j connection:

conn_spec = cocoindex.add_auth_entry(
    "Neo4jConnection",
    cocoindex.storages.Neo4jConnection(
        uri="bolt://localhost:7687",
        user="neo4j",
        password="cocoindex",
))

Export `Document` nodes to Neo4j

document_node.export(
    "document_node",
    cocoindex.storages.Neo4j(
        connection=conn_spec,
        mapping=cocoindex.storages.Nodes(label="Document")),
    primary_key_fields=["filename"],
)

This exports Neo4j nodes with label Document from the document_node collector.

It declares Neo4j node label Document. It specifies filename as the primary key field.
It carries all the fields from document_node collector to Neo4j nodes with label Document.

Export `RELATIONSHIP` and `Entity` nodes to Neo4j

We don't have explicit collector for Entity nodes.
They are part of the entity_relationship collector and fields are collected during the relationship extraction.

To export them as Neo4j nodes, we need to first declare Entity nodes.

flow_builder.declare(
    cocoindex.storages.Neo4jDeclaration(
        connection=conn_spec,
        nodes_label="Entity",
        primary_key_fields=["value"],
    )
)

Next, export the entity_relationship to Neo4j.

entity_relationship.export(
    "entity_relationship",
    cocoindex.storages.Neo4j(
        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"],
)
)

The cocoindex.storages.Relationships declares how to map relationships in Neo4j.

In a relationship, there's:

A source node and a target node.
A relationship connecting the source and target. Note that different relationships may share the same source and target nodes.

NodeFromFields takes the fields from the entity_relationship collector and creates Entity nodes.

Export the `entity_mention` to Neo4j.

entity_mention.export(
    "entity_mention",
    cocoindex.storages.Neo4j(
        connection=conn_spec,
        mapping=cocoindex.storages.Relationships(
            rel_type="MENTION",
            source=cocoindex.storages.NodesFromFields(
                label="Document",
                fields=[cocoindex.storages.TargetFieldMapping("filename")],
            ),
            target=cocoindex.storages.NodesFromFields(
                label="Entity",
                fields=[cocoindex.storages.TargetFieldMapping(
                    source="entity", target="value")],
            ),
        ),
    ),
    primary_key_fields=["id"],
)

Similarly here, we export entity_mention to Neo4j Relationships using cocoindex.storages.Relationships.
It creates relationships by:

Creating Document nodes and Entity nodes from the entity_mention collector.
Connecting Document nodes and Entity nodes with relationship MENTION.

Main function

Finally, the main function for the flow initializes the CocoIndex flow and runs it.

 @cocoindex.main_fn()
def _run():
    pass

if __name__ == "__main__":
    load_dotenv(override=True)
    _run()

Query and test your index

🎉 Now you are all set!

Install the dependencies:
```
pip install -e .
```
Run following commands to setup and update the index.
```
python main.py cocoindex setup
python main.py cocoindex update
```
You'll see the index updates state in the terminal. For example, you'll see the following output:
```
documents: 7 added, 0 removed, 0 updated
```
(Optional) I used CocoInsight to troubleshoot the index generation and understand the data lineage of the pipeline.
It is in free beta now, you can give it a try. It just connects to your local CocoIndex server, with Zero pipeline data retention.
Run following command to start CocoInsight:
```
python3 main.py cocoindex server -c https://cocoindex.io
```
And then open the url https://cocoindex.io/cocoinsight.

Browse the knowledge graph

After the knowledge graph is built, you can explore the knowledge graph you built in Neo4j Browser.

For the dev environment, you can connect to Neo4j browser using credentials:

username: Neo4j
password: cocoindex which is pre-configured in our docker compose config.yaml.

You can open it at http://localhost:7474, and run the following Cypher query to get all relationships:

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

Support us

We are constantly improving, and more features and examples are coming soon.
If you love this article, please give us a star ⭐ at GitHub repo to help us grow.

Thanks for reading!

How to do incremental processing for ETL - by examples

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

We could take a look at a few examples to understand what CocoIndex handles behind the scene 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 scene:

Ideally, we only recompute embeddings for the 4 new rows, and reuse it 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 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 spec for embedding is unchanged.
If the changed part is the embedding logic or spec, we will recompute the embeddings for everything.
CocoIndex is able to see if the logic or spec for an operation step is 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's robust to transformation logic non-determinism and changes.

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 for 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 product exists 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.

:::tip Google Drive Change Log

Google drive change log is only supported on drive level, not 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 leverage this even if there's no source-specific change capture.

3. Optimizations for specific sources

When source connector provides advanced features to list entries, for example, list 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 condition here on when it is not available.

CocoIndex could monitor the change based on last modified vs last poll time, 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 lego block in the pipeline can be cached.
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.

Custom functions can take a parameter cache.
When True, the executor will cache the result of the function for reuse during reprocessing.
We recommend to set this to True for any function that is computationally intensive.

Output will be reused if all these are unchanged: spec (if exists), input data, behavior of the function.
For this purpose, a behavior_version needs to be provided, and should increase on behavior changes.

For example, this enables cache for a standalone function:

@cocoindex.op.function(cache=True, behavior_version=1)
def compute_something(arg1: str, arg2: int | None = None) -> str:
    ...

This enables cache for a function defined by a spec and an executor:

class ComputeSomething(cocoindex.op.FunctionSpec):
    ...

@cocoindex.op.executor_class(cache=True, behavior_version=1)
class ComputeSomethingExecutor:
    spec: ComputeSomething

    ...

Thanks for reading and we love to hear from you!
If you like our work, it would mean a lot to us if you could support Cocoindex on Github with a star.
Thank you so much with a warm coconut hug 🥥🤗.

DEV Community: CocoIndex

How I Built a Semantic Search Engine with CocoIndex

Introduction

What is CocoIndex?

Why I Chose CocoIndex

Getting Started

Building the Search Engine

1. Initialize CocoIndex

2. Add Documents

Index the documents

3. Perform Semantic Search

Key Features I Implemented

Fast Indexing

Semantic Understanding

Customizable Embeddings

Real-World Example

Performance Tips

Challenges I Faced

Challenge 1: Choosing Embedding Dimensions

Challenge 2: Handling Large Document Collections

Results

Conclusion

Resources

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

Use Cases

What We Will Achieve

Core Components

Image Ingestion

Detect and Extract Faces

Compute Face Embeddings

Collect and Export Embeddings

Query the Index

Support Us

Build Real-Time Codebase Indexing for AI Coding agents

Tree-sitter

Prerequisites

Define Indexing Flow

Flow Design

1. Add the codebase as a source.

2. Process each file and collect the information.

2.1 Extract the extension of a filename

2.2 Split the file into chunks

2.3 Embed the chunks

2.4 Collect the embeddings

3. Query the index

Run the index setup & update

Test the query

Support us

Beyond Embeddings: Building Metadata-Rich Indexes from Academic PDFs

Use Cases

What we will achieve

Core Components

Prerequisites

Define Indexing Flow

Import the Papers

Extract and collect metadata

Extract first page for basic info

Parse basic info

Extract basic info with LLM

Collect paper metadata

Collect author to filename information

Compute and collect embeddings

Title

Abstract

Collect embeddings

Export

View in CocoInsight step by step

Query the index

Support us

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

What You’ll Use

How to map to Kuzu in CocoIndex

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

Support us

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

Why Incremental Processing?

When Incremental Processing Truly Matters

Freshness is Critical

Compute is Expensive

Working at Scale

Collect `author` to `filename` information