# Recipes

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.

## 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`](/docs/chunking-modes/section). |
| "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`](/docs/api-reference/python#fit_tokens--fit-chunks-to-a-token-budget) | 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](/docs/chunking-modes/page-aware#which-formats-have-real-pages) before you build citations. |
| A spreadsheet or CSV | `row` (the default) | Row-per-chunk; `table` / `sheet` / `semantic` group rows instead. |

  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](/docs/chunking-modes).

## 2. RAG ingestion (the minimal shape)

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

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

```ts
import { getChunks } from "js-chunks";

async function ingest(path: string) {
  // Semantic chunks make the best embeddings.
  for (const chunk of await getChunks(path, { mode: "semantic" })) {
    const vector = await embed(chunk.content);    // your embedding model
    await store(vector, { text: chunk.content, meta: chunk.metadata });
  }
}
```

```rust
use chunks_rs::get_chunks;

fn ingest(path: &str) -> Result<(), Box<dyn std::error::Error>> {
    // Semantic chunks make the best embeddings.
    for c in &get_chunks(path, "semantic", 3, 1, 3, 15)? {
        let vector = embed(&c.content);           // your embedding model
        store(vector, &c.content, &c.metadata);
    }
    Ok(())
}
```

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](/docs/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.

```python
# 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.
```

```ts
// npm i js-chunks @qdrant/js-client-rest
import { QdrantClient } from "@qdrant/js-client-rest";
import { getChunks } from "js-chunks";

const qdrant = new QdrantClient({ url: "http://localhost:6333" });
const COLLECTION = "docs";

await qdrant.createCollection(COLLECTION, {
  vectors: { size: 1536, distance: "Cosine" },
}).catch(() => {});   // already exists

export async function ingest(path: string) {
  const chunks = await getChunks(path, { mode: "semantic" });

  await qdrant.upsert(COLLECTION, {
    wait: true,
    points: await Promise.all(
      chunks.map(async (chunk, i) => ({
        id: `${path}#${i}`,
        vector: await embed(chunk.content),        // your embedding model
        payload: {
          source: path,
          chunkIndex: i,
          content: chunk.content,
          contentType: chunk.contentType,
          ...chunk.metadata,
        },
      })),
    ),
  });

  return chunks.length;
}
```

```rust
// qdrant-client = "1"
use chunks_rs::get_chunks;
use qdrant_client::qdrant::{PointStruct, UpsertPointsBuilder};
use qdrant_client::Qdrant;

async fn ingest(client: &Qdrant, path: &str) -> Result<usize, Box<dyn std::error::Error>> {
    let chunks = get_chunks(path, "semantic", 3, 1, 3, 15)?;

    let points: Vec<PointStruct> = chunks
        .iter()
        .enumerate()
        .map(|(i, c)| {
            PointStruct::new(
                i as u64,
                embed(&c.content),                   // your embedding model
                [
                    ("source", path.into()),
                    ("chunk_index", (i as i64).into()),
                    ("content", c.content.clone().into()),
                    ("content_type", c.content_type.clone().into()),
                ],
            )
        })
        .collect();

    client
        .upsert_points(UpsertPointsBuilder::new("docs", points).wait(true))
        .await?;
    Ok(chunks.len())
}
```

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](/docs/output-schema)).
- **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.

```python
# 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.
```

```ts
// npm i js-chunks @langchain/core
import { BaseDocumentLoader } from "@langchain/core/document_loaders/base";
import { Document } from "@langchain/core/documents";
import { getChunks, type ChunkMode } from "js-chunks";

export class ChunkEngineLoader extends BaseDocumentLoader {
  constructor(
    private path: string,
    private mode: ChunkMode = "semantic",
  ) {
    super();
  }

  async load(): Promise<Document[]> {
    const chunks = await getChunks(this.path, { mode: this.mode });
    return chunks.map(
      (chunk, i) =>
        new Document({
          pageContent: chunk.content,
          metadata: {
            source: this.path,
            chunkIndex: i,
            contentType: chunk.contentType,
            ...chunk.metadata,
          },
        }),
    );
  }
}

const docs = await new ChunkEngineLoader("./handbook.docx").load();
```

**Rust**

There is no LangChain binding for Rust. Use the vector-DB recipe above directly,
or expose an HTTP endpoint from your Rust service and call it from a Python
LangChain loader — see [Framework Integration → Rust](/docs/framework-integration/rust).

## 5. LlamaIndex

**Python**

Emit `TextNode`s straight from chunks and build the index with no
transformations.

```python
# 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
```

**JavaScript** / **Rust**

This adapter is Python-only — there is no js-chunks or rs-chunks LlamaIndex
integration. Switch to Python above to read it, or use the LangChain.js loader
in the previous recipe.

## 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.

```python
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
```

```ts
import { createHash } from "node:crypto";
import { getChunks } from "js-chunks";

const contentHash = (text: string) =>
  createHash("sha256").update(text, "utf8").digest("hex");

export async function reindex(path: string, store: Store) {
  const chunks = await getChunks(path, { mode: "semantic" });
  const known = await store.hashesForSource(path);   // Map<number, string>
  const stats = { added: 0, updated: 0, deleted: 0, unchanged: 0 };

  for (const [i, chunk] of chunks.entries()) {
    const hash = contentHash(chunk.content);
    if (known.get(i) === hash) {
      stats.unchanged++;
      continue;
    }
    await store.upsert({
      key: `${path}#${i}`,
      vector: await embed(chunk.content),
      text: chunk.content,
      meta: { ...chunk.metadata, source: path, chunkIndex: i, contentHash: hash },
    });
    known.has(i) ? stats.updated++ : stats.added++;
  }

  for (const idx of known.keys()) {
    if (idx >= chunks.length) {
      await store.delete(`${path}#${idx}`);
      stats.deleted++;
    }
  }

  return stats;
}
```

```rust
use std::collections::HashMap;

use chunks_rs::get_chunks;
use sha2::{Digest, Sha256};

fn content_hash(text: &str) -> String {
    format!("{:x}", Sha256::digest(text.as_bytes()))
}

fn reindex(path: &str, store: &mut Store) -> Result<(), Box<dyn std::error::Error>> {
    let chunks = get_chunks(path, "semantic", 3, 1, 3, 15)?;
    let known: HashMap<usize, String> = store.hashes_for_source(path);

    for (i, chunk) in chunks.iter().enumerate() {
        let hash = content_hash(&chunk.content);
        if known.get(&i) == Some(&hash) {
            continue;                       // unchanged — skip the embedding call
        }
        store.upsert(path, i, embed(&chunk.content), &chunk.content, &hash);
    }

    for stale in known.keys().filter(|i| **i >= chunks.len()) {
        store.delete(path, *stale);
    }
    Ok(())
}
```

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.

  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.

```python
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.
```

```ts
import { createHash } from "node:crypto";
import { readdir } from "node:fs/promises";
import { getChunks, type Chunk } from "js-chunks";

const normalize = (t: string) => t.split(/s+/).join(" ").toLowerCase();
const key = (t: string) => createHash("sha256").update(normalize(t)).digest("hex");

const seen = new Map<string, string>();
const unique: Chunk[] = [];
let duplicates = 0;

for (const name of await readdir("corpus")) {
  let chunks: Chunk[];
  try {
    chunks = await getChunks(`corpus/${name}`, { mode: "semantic" });
  } catch (e) {
    console.warn("skip", name, e);
    continue;
  }
  for (const [i, chunk] of chunks.entries()) {
    const k = key(chunk.content);
    if (seen.has(k)) {
      duplicates++;
      continue;
    }
    seen.set(k, `${name}#${i}`);
    unique.push(chunk);
  }
}

console.log(`${unique.length} unique, ${duplicates} duplicate chunks`);
```

```rust
use std::collections::HashMap;
use std::fs;

use chunks_rs::get_chunks;
use sha2::{Digest, Sha256};

fn normalize(text: &str) -> String {
    text.split_whitespace().collect::<Vec<_>>().join(" ").to_lowercase()
}

fn main() -> Result<(), Box<dyn std::error::Error>> {
    let mut seen: HashMap<String, String> = HashMap::new();
    let (mut unique, mut duplicates) = (0usize, 0usize);

    for entry in fs::read_dir("corpus")? {
        let path = entry?.path();
        let p = path.to_string_lossy().to_string();
        let chunks = match get_chunks(&p, "semantic", 3, 1, 3, 15) {
            Ok(c) => c,
            Err(e) => {
                eprintln!("skip {p}: {e}");
                continue;
            }
        };
        for (i, c) in chunks.iter().enumerate() {
            let key = format!("{:x}", Sha256::digest(normalize(&c.content).as_bytes()));
            if seen.contains_key(&key) {
                duplicates += 1;
                continue;
            }
            seen.insert(key, format!("{p}#{i}"));
            unique += 1;
        }
    }

    println!("{unique} unique, {duplicates} duplicate chunks");
    Ok(())
}
```

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.

```python
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
```

```ts
import { readdir } from "node:fs/promises";
import { getChunks } from "js-chunks";

for (const name of await readdir("docs")) {
  try {
    index(name, await getChunks(`docs/${name}`));
  } catch (e) {
    console.warn("skip", name, e);   // one bad file shouldn't halt the batch
  }
}
```

```rust
use std::fs;
use chunks_rs::get_chunks;

fn main() -> Result<(), Box<dyn std::error::Error>> {
    for entry in fs::read_dir("docs")? {
        let path = entry?.path();
        let p = path.to_string_lossy().to_string();
        match get_chunks(&p, "default", 3, 1, 3, 15) {
            Ok(chunks) => index(&path, &chunks),
            Err(e) => eprintln!("skip {p}: {e}"),  // keep going
        }
    }
    Ok(())
}
```

See [Error Handling](/docs/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](/docs/streaming).
- **Convert to Markdown** — `get_markdown` / `getMarkdown` for previews,
  diffing, or feeding a Markdown-native tool: [Quick Start](/docs/quick-start).
- **Web handlers** — FastAPI, Flask, Django, Express, Next.js, Axum:
  [Framework Integration](/docs/framework-integration).