End-to-end patterns — vector DB ingestion, LangChain and LlamaIndex adapters, incremental re-indexing, de-duplication, and choosing a mode.

Practical patterns you can paste. Every snippet is language-aware — use the tabs or the switcher below to see Python, JavaScript, or Rust.

Showing examples for Python

— your choice follows you across the docs.

1. Choose a mode first

Everything downstream depends on this, and it is the cheapest thing to get right. chunk-engine has already parsed the document's structure; the mode decides how much of that structure survives into a chunk.

Your retrieval unit	Mode	Why
"A coherent idea" — the usual RAG answer	`semantic`	Merges adjacent blocks while a lexical ladder says they continue each other; capped at 1,500 chars.
"A whole documented topic"	`section`	One chunk per heading body. Headings come back as separate `heading` chunks — pair them by `heading_path`.
"A paragraph / a table / a code block"	`default` / `structural`	Finest grain; each chunk is one element with its own `content_type`.
"A fixed token budget"	`sentence` + `fit_tokens`	Modes size in characters (the parity-safe unit); `fit_tokens` / `fitTokens` then guarantee no chunk exceeds N tokens under your own tokenizer.
"Recall above all"	`sliding_window`	Overlapping windows; nothing falls in a crack. Costs index size.
"A citable page"	`page_aware`	Only DOCX / DOC carry a real `page_number` — check the table before you build citations.
A spreadsheet or CSV	`row` (the default)	Row-per-chunk; `table` / `sheet` / `semantic` group rows instead.

Do not re-split chunk-engine output

The most common mistake in a LangChain or LlamaIndex pipeline is to run RecursiveCharacterTextSplitter (or a node parser) over these chunks. That cuts the structure the engine just preserved, mid-table and mid-sentence. The adapters below deliberately emit finished Document / TextNode objects with no further splitting.

Start with semantic and change it only when a measurement says to. The full comparison is in Chunking Modes.

2. RAG ingestion (the minimal shape)

Chunk, embed, store. Ninety percent of pipelines are this plus error handling.

ingest.py

from py_chunks import get_chunks

def ingest(path):
    # Semantic chunks make the best embeddings.
    for chunk in get_chunks(path, mode="semantic"):
        vector = embed(chunk["content"])          # your embedding model
        store(vector, text=chunk["content"], meta=chunk["metadata"])

Each chunk carries a typed content_type and a metadata object (section headings, page numbers, merge reasons, …), so you can filter results or show provenance at query time — see the Output Schema.

3. Ingest into a real vector database

The same shape against an actual client: pgvector via psycopg 3 in Python, Qdrant in JavaScript and Rust.

ingest_pgvector.py

# pip install py-chunks psycopg[binary] pgvector
import psycopg
from pgvector.psycopg import register_vector
from py_chunks import get_chunks

conn = psycopg.connect("postgresql://localhost/rag")
register_vector(conn)
conn.execute("CREATE EXTENSION IF NOT EXISTS vector")
conn.execute("""
    CREATE TABLE IF NOT EXISTS chunks (
        id          bigserial PRIMARY KEY,
        source      text NOT NULL,
        chunk_index int  NOT NULL,
        content     text NOT NULL,
        content_type text NOT NULL,
        metadata    jsonb NOT NULL,
        embedding   vector(1536),
        UNIQUE (source, chunk_index)
    )
""")

def ingest(path: str) -> int:
    chunks = get_chunks(path, mode="semantic")
    with conn.cursor() as cur:
        for i, chunk in enumerate(chunks):
            cur.execute(
                """INSERT INTO chunks
                       (source, chunk_index, content, content_type, metadata, embedding)
                   VALUES (%s, %s, %s, %s, %s, %s)
                   ON CONFLICT (source, chunk_index) DO UPDATE
                       SET content   = EXCLUDED.content,
                           metadata  = EXCLUDED.metadata,
                           embedding = EXCLUDED.embedding""",
                (path, i, chunk["content"], chunk["content_type"],
                 psycopg.types.json.Jsonb(chunk["metadata"]),
                 embed(chunk["content"])),          # your embedding model
            )
    conn.commit()
    return len(chunks)

# chunk_index is positional here because not every format writes one into
# metadata. See /docs/output-schema for which ones do.

Two things worth copying regardless of which store you use:

Key on (source, chunk_index) and upsert, so a re-run of the same document replaces rather than duplicates. chunk_index here is the loop position — not every format writes one into metadata (which ones do).
Store the whole metadata object. It costs almost nothing in a JSONB / payload column and it is what lets you answer "which page was that on?" later without re-chunking.

4. LangChain

A loader that is already chunked — no TextSplitter in the pipeline.

langchain_loader.py

# pip install py-chunks langchain-core
from langchain_core.documents import Document
from langchain_core.document_loaders import BaseLoader
from py_chunks import get_chunks


class ChunkEngineLoader(BaseLoader):
    """A LangChain loader that is already chunked — no TextSplitter needed."""

    def __init__(self, path: str, mode: str = "semantic"):
        self.path, self.mode = path, mode

    def lazy_load(self):
        for i, chunk in enumerate(get_chunks(self.path, mode=self.mode)):
            yield Document(
                page_content=chunk["content"],
                metadata={
                    "source": self.path,
                    "chunk_index": i,
                    "content_type": chunk["content_type"],
                    **chunk["metadata"],
                },
            )


docs = ChunkEngineLoader("handbook.docx").load()
# Feed straight to a vector store — do NOT run RecursiveCharacterTextSplitter
# over these; it would cut the structure chunk-engine just preserved.

5. LlamaIndex

Emit TextNodes straight from chunks and build the index with no transformations.

llamaindex_reader.py

# pip install py-chunks llama-index-core
from llama_index.core import VectorStoreIndex
from llama_index.core.readers.base import BaseReader
from llama_index.core.schema import TextNode
from py_chunks import get_chunks


class ChunkEngineReader(BaseReader):
    """Emit TextNodes directly — skip LlamaIndex's own node parser."""

    def load_data(self, path: str, mode: str = "semantic") -> list[TextNode]:
        nodes = []
        for i, chunk in enumerate(get_chunks(path, mode=mode)):
            nodes.append(
                TextNode(
                    text=chunk["content"],
                    id_=f"{path}#{i}",
                    metadata={
                        "source": path,
                        "chunk_index": i,
                        "content_type": chunk["content_type"],
                        **chunk["metadata"],
                    },
                )
            )
        return nodes


nodes = ChunkEngineReader().load_data("research.pdf")
index = VectorStoreIndex(nodes)   # nodes are pre-chunked; no transformations

6. Incremental re-indexing

Re-embedding an entire corpus because one file changed is the most expensive avoidable thing in a RAG pipeline. Give every chunk a stable identity — metadata.source + chunk_index — and a content hash, then only re-embed what moved.

reindex.py

import hashlib
from py_chunks import get_chunks


def content_hash(text: str) -> str:
    return hashlib.sha256(text.encode("utf-8")).hexdigest()


def reindex(path: str, store) -> dict:
    """Re-embed only the chunks whose text actually changed."""
    chunks = get_chunks(path, mode="semantic")

    # {chunk_index: hash} of what is already indexed for this source
    known = store.hashes_for_source(path)
    stats = {"added": 0, "updated": 0, "deleted": 0, "unchanged": 0}

    for i, chunk in enumerate(chunks):
        h = content_hash(chunk["content"])
        if known.get(i) == h:
            stats["unchanged"] += 1
            continue
        store.upsert(
            key=(path, i),
            vector=embed(chunk["content"]),
            text=chunk["content"],
            meta={**chunk["metadata"], "source": path,
                  "chunk_index": i, "content_hash": h},
        )
        stats["updated" if i in known else "added"] += 1

    # The document got shorter: drop the tail.
    for stale in (idx for idx in known if idx >= len(chunks)):
        store.delete((path, stale))
        stats["deleted"] += 1

    return stats

Three cases the code above covers, and you need all three:

Case	Detection	Action
Chunk unchanged	hash matches	skip — no embedding call
Chunk edited	hash differs, index known	re-embed and upsert
Document got shorter	index ≥ new chunk count	delete the tail

That last one is the one people miss: if a document loses a section, the old chunks stay in the index forever and keep being retrieved.

Chunk boundaries can move

Editing paragraph 2 of a document can shift every chunk after it, so a one-line edit may legitimately re-embed most of the file. That is correct behaviour, not a bug in the hash — semantic and section boundaries depend on the surrounding text. If you need edit-locality, structural produces the most stable chunk identities because each chunk is one element.

7. De-duplicate a corpus

Headers, footers, legal boilerplate and copy-pasted sections repeat verbatim across documents. In an index they crowd out real answers.

dedup.py

import hashlib
from pathlib import Path
from py_chunks import get_chunks


def normalize(text: str) -> str:
    # Collapse whitespace so "same paragraph, different wrapping" collides.
    return " ".join(text.split()).casefold()


seen: dict[str, tuple[str, int]] = {}
unique, duplicates = [], []

for path in Path("corpus").rglob("*"):
    if not path.is_file():
        continue
    try:
        chunks = get_chunks(str(path), mode="semantic")
    except Exception as e:
        print("skip", path, e)
        continue

    for i, chunk in enumerate(chunks):
        key = hashlib.sha256(normalize(chunk["content"]).encode()).hexdigest()
        if key in seen:
            duplicates.append((str(path), i, seen[key]))
            continue
        seen[key] = (str(path), i)
        unique.append(chunk)

print(f"{len(unique)} unique, {len(duplicates)} duplicate chunks")

# Tip: sliding_window is *designed* to overlap, so never dedup its output —
# you would delete the overlap that makes the mode work.

Normalize whitespace and case before hashing, or "same paragraph, different line wrapping" will read as two distinct chunks. And never de-duplicate sliding_window output — the overlap it removes is the entire point of the mode.

8. Batch a folder of mixed formats

The same call handles every format, so a mixed folder just works. Isolate failures per file so one bad document doesn't halt the batch.

batch.py

from pathlib import Path
from py_chunks import get_chunks

for path in Path("docs").rglob("*"):
    if not path.is_file():
        continue
    try:
        index(path.name, get_chunks(str(path)))
    except Exception as e:
        print("skip", path, e)   # one bad file shouldn't halt the batch

See Error Handling for what each failure actually raises — including an unreadable path, which in JavaScript is a ChunkError with kind: "io" carrying Node's ENOENT message.

Also see

Stream a large file — one chunk at a time, and what that does and does not buy you per runtime: Streaming.
Convert to Markdown — get_markdown / getMarkdown for previews, diffing, or feeding a Markdown-native tool: Quick Start.
Web handlers — FastAPI, Flask, Django, Express, Next.js, Axum: Framework Integration.

Recipes