Frontend Beauty
ESC

Type to search the knowledge base.

Document Ingestion Pipelines

A document ingestion pipeline extracts, cleans, chunks, embeds, upserts, and deletes. Version it so a parser bug can be rebuilt.

intermediate6 min read
  • rag
  • ingestion
  • pipelines

If you cannot rebuild the index from source files, you do not have a document ingestion pipeline. You have a snowflake: a laptop script, a half-updated namespace, and a chatbot that still quotes the PDF you deleted. Ingestion is extract, clean, chunk, embed, upsert, and delete, with a version number on the code that did it.

Parser bugs happen. The pipeline’s job is to make “run it again” a boring command, not a folk ritual.

Key takeaways

  • Source files are the source of truth. The index is a projection you can throw away.
  • Identity is sourceId plus a content hash, not “whatever row we inserted on Tuesday.”
  • Delete is a first-class stage. A removed page must not linger as a high-ranking chunk.
  • Observe extract failures as product metrics. Silent empty extracts become confident nonsense.
  • Queue for large or bursty corpora. Batch APIs for embeddings. Chat APIs for chat.

A document ingestion pipeline has six stages

Stage Input Output Typical failure
Extract File bytes or MDX Text + structure (headings, tables) Scanned PDF, wrong encoding
Clean Raw text Body without nav, cookie banners, repeated footers Left the sidebar in, retrieve “Buy now”
Chunk Clean text Sections with metadata Timer split through a table
Embed Chunk text Vectors from one model Mixed models in one index
Upsert Vectors + metadata Index rows keyed by chunk id Append-only duplicates
Delete Source id or hash set Tombstones / removed rows Orphans rank forever

This is not chunking strategies and not RAG on PDFs. Those pages pick splitters and parsers. Here you wire the job so those choices are replayable.

Content hashes as identity

Hash the bytes you actually embed (chunk text plus the metadata you send to the model). If the hash matches, skip the embedding call. If it changed, replace the rows for that source. If the source vanished, delete.

import { createHash } from 'node:crypto';
import { embedMany } from 'ai';

type Source = { sourceId: string; chunks: { heading: string; text: string }[] };
type Row = {
  id: string;
  sourceId: string;
  heading: string;
  text: string;
  hash: string;
  pipeline: string;
  embedding: number[];
};

export const PIPELINE = 'docs-mdx-v3';

function hashChunk(sourceId: string, heading: string, text: string) {
  return createHash('sha256')
    .update(PIPELINE)
    .update(sourceId)
    .update(heading)
    .update(text)
    .digest('hex');
}

export async function ingestSource(
  source: Source,
  index: {
    fetchBySource(id: string): Promise<Row[]>;
    upsert(rows: Row[]): Promise<void>;
    deleteIds(ids: string[]): Promise<void>;
  },
) {
  const existing = await index.fetchBySource(source.sourceId);
  const byHash = new Map(existing.map((row) => [row.hash, row]));

  const next: Row[] = [];
  const toEmbed: { heading: string; text: string; hash: string }[] = [];

  for (const [i, chunk] of source.chunks.entries()) {
    const hash = hashChunk(source.sourceId, chunk.heading, chunk.text);
    const reuse = byHash.get(hash);
    if (reuse) {
      next.push({ ...reuse, id: `${source.sourceId}:${i}` });
    } else {
      toEmbed.push({ ...chunk, hash });
      next.push({
        id: `${source.sourceId}:${i}`,
        sourceId: source.sourceId,
        heading: chunk.heading,
        text: chunk.text,
        hash,
        pipeline: PIPELINE,
        embedding: [],
      });
    }
  }

  if (toEmbed.length > 0) {
    const { embeddings } = await embedMany({
      model: 'openai/text-embedding-3-small',
      values: toEmbed.map((c) => `${c.heading}\n${c.text}`),
    });
    let j = 0;
    for (const row of next) {
      if (row.embedding.length === 0) {
        row.embedding = embeddings[j++];
      }
    }
  }

  const nextIds = new Set(next.map((r) => r.id));
  const stale = existing.map((r) => r.id).filter((id) => !nextIds.has(id));
  if (stale.length) await index.deleteIds(stale);
  await index.upsert(next);
}

Idempotent upserts mean a crash mid-file is safe. Re-run the source. Do not “insert again” without delete, or cosine search will return two near-identical chunks and waste the prompt.

PIPELINE is in the hash on purpose. When you change the cleaner, every chunk is new. That is a rebuild, not a surprise.

Delete is part of ingest

A job that only writes will rot. Sources of deletes:

  • File removed from the repo
  • draft: true flipped on
  • ACL says the tenant must not see it
  • Parser now returns empty (treat as failure, not as “empty doc”)
export async function reconcile(
  liveSourceIds: Set<string>,
  listed: { sourceId: string }[],
  deleteBySource: (id: string) => Promise<void>,
) {
  for (const row of listed) {
    if (!liveSourceIds.has(row.sourceId)) {
      await deleteBySource(row.sourceId);
    }
  }
}

If your vector store has no delete, you do not have a store you can operate. Soft-delete via deletedAt metadata and a filter is acceptable only if every query includes the filter. Forgetting the filter is how “deleted” docs still answer. Vector index updates covers freshness. Ingestion is what produces the events.

Queue vs batch

Work Mechanism
One MDX commit Synchronous CI job, or a queue with one source per task
10k PDFs Queue. Isolate parser crashes per file
Embedding those 10k Provider batch APIs, not the chat endpoint
User-facing Q&A Streaming chat, never this pipeline

A parser that throws must fail the file, record the error, and continue the corpus. One bad PDF is not a reason to skip the rest, and a swallowed exception is not a reason to index empty strings.

type IngestEvent =
  | { type: 'ok'; sourceId: string; chunks: number; ms: number }
  | { type: 'extract_failed'; sourceId: string; reason: string }
  | { type: 'empty_extract'; sourceId: string };

export function record(e: IngestEvent) {
  // metrics + log. empty_extract should page you after a threshold.
  console.info(JSON.stringify({ pipeline: PIPELINE, ...e }));
}

Watch extract_failed and empty_extract like you watch 5xx. Retrieval quality is downstream of this counter.

Version the pipeline

Store pipeline: 'docs-mdx-v3' on every row. When v4 ships (better HTML strip, new chunker), you can:

  1. Build v4 into a new namespace or table.
  2. Flip the query alias.
  3. Drop v3 after a soak.

Or rebuild in place if downtime is fine. What you must not do is mix v3 and v4 vectors in one unfiltered namespace. Different chunk boundaries plus the same source id is duplicate meaning.

Python loaders (LangChain document loaders) are fine for extract. The ops contract is language-agnostic: hash, upsert, delete, version, observe. TypeScript above is the same contract you can run next to a Next.js app.

Footguns

  • Indexing the built site HTML without stripping chrome. You will retrieve the global nav.
  • Embedding with model A, querying with model B. Similarity is undefined.
  • No sourceId. You cannot delete.
  • Transforming text after hashing. Then the hash never matches and you re-embed forever.
  • Running ingest inside the chat request. First-token latency becomes “we crawled Notion.”

Interview answer (out loud)

“A document ingestion pipeline is a replayable projection: extract, clean, chunk, embed, upsert, delete. I key rows by source id and a hash of the pipeline plus text so re-runs are idempotent. Deletes are part of the job. I version the pipeline so a parser fix can rebuild. Failures are metrics, not silent empty chunks. Embeddings go through batch or a queue, not the chat path.”

Further reading (primary sources)

Related guides