back to topology
ai-agents
article··updated ·10 min read

RAG ingestion in n8n: document versions, deletions and freshness metadata

#n8n #rag #vector-db #pinecone #data-quality

A retrieval-augmented chatbot answers from whatever the vector store returns. If the store holds two versions of the same price list, the retriever picks the chunk that scores best on the index metric, and nothing in that score reflects which version is current. The model can answer correctly with respect to its context, even though that context is out of date. The answer reads well and cites a real document, but the price in it is wrong. Nothing shows up in the error log and no hallucination detector fires, because the model didn't invent anything.

The problem is in the ingestion workflow. Changing the prompt or the model won't help much. Below are the invariants an ingestion pipeline should hold and a way to implement them in n8n with Pinecone. Qdrant, Weaviate or pgvector follow the same rules.

Why stale chunks survive

A vector store has no notion of a document version. It stores records under IDs, and an upsert replaces a record only when the ID matches, while records with other IDs stay in the index.

Duplicates usually come from IDs derived from the filename or from the content. A new version saved as pricing-2026-v2.pdf, or one whose first paragraph changed, gets new IDs, and the old records stay where they were.

A shorter new version does the same thing more quietly. Say version 1 produced 12 chunks and version 2 produces 9. An upsert by docId + index overwrites chunks 0 to 8 and leaves 9 to 11 from the old version.

Then there are deletions at the source. Moving a file to an archive folder or deleting it in Google Drive doesn't trigger anything in the vector store, so without an explicit delete the chunks stay retrievable forever.

A typical first version of the workflow (Drive trigger, text splitter, Pinecone upsert) has all of these problems at once:

// n8n Code node: naive chunk preparation
return $input.all().map(chunk => ({
  json: {
    id: chunk.json.metadata.loc.pageNumber + '_' + chunk.json.pageContent.slice(0, 32),
    text: chunk.json.pageContent,
    metadata: { source: chunk.json.metadata.source },
  },
}));

The ID depends on page content, the only metadata is the filename, and there is no delete step anywhere in the flow.

The invariant

For every document in the source folder, the index contains exactly the chunks of its current version, and nothing else. The code below deliberately departs from this in one case: a document whose text extraction failed is skipped and keeps its previous chunks, so those files need their own alert. Between runs (a failed cleanup, a deletion waiting for reconciliation) the index can also briefly hold more. The design decisions below come from that:

  • the document ID comes from the source system (the Drive file ID), because the file ID doesn't change when someone renames the file or uploads a new revision (a new version saved as a separate file does get a new file ID, so two price-list files have to be sorted out at the source, for example by keeping old versions in an archive outside the source folder);
  • chunk IDs are prefixed with the document ID and a content hash, so all chunks of one document can be found by prefix and old versions can be told apart from the new one;
  • replacing a document means writing the new chunks and then deleting every chunk with the same document prefix and a different hash;
  • a scheduled reconciliation removes documents that no longer exist at the source;
  • freshness is stored as numbers, because that is what metadata range filters operate on.

Preparing chunks

The node below runs in "Run Once for All Items" mode. It expects one item per Drive file with id, name, modifiedTime, mimeType, the extracted text and a docType (for example derived from the subfolder). On self-hosted n8n, require('crypto') works only if the instance runs with NODE_FUNCTION_ALLOW_BUILTIN=crypto (in external mode, with the runner in a separate container, the variable goes on the runner; in the default internal mode setting it on the main instance is enough, because the main instance passes it to the runner process).

// n8n Code node, mode: Run Once for All Items
const crypto = require('crypto');

const CHUNK_SIZE = 1500; // Unicode code points
const OVERLAP = 200;
// Days after the last modification when a document stops being served.
// null means the document type does not expire.
const TTL_DAYS = { pricing: 90, changelog: 180, adr: null, general: null };
const NEVER = 4102444800; // 2100-01-01 as a Unix timestamp

function split(text) {
  const chars = Array.from(text); // code points, so a surrogate pair is never cut in half
  const chunks = [];
  for (let start = 0; start < chars.length; start += CHUNK_SIZE - OVERLAP) {
    chunks.push(chars.slice(start, start + CHUNK_SIZE).join(''));
    if (start + CHUNK_SIZE >= chars.length) break;
  }
  return chunks;
}

const out = [];
for (const item of $input.all()) {
  const { id: docId, name, modifiedTime, mimeType, text } = item.json;
  // Near-empty text is usually an extraction failure: the document is skipped
  // and its previous chunks stay in the index.
  if (typeof text !== 'string' || text.trim().length < 50) continue;

  const docType = item.json.docType ?? 'general';
  // A typo such as 'pricng' would otherwise give expiresTs = NaN, which Pinecone won't store.
  // This stops the whole run, so the mislabelled file gets fixed before anything is written.
  if (!Object.hasOwn(TTL_DAYS, docType)) throw new Error(`Unknown docType "${docType}" for ${name}`);
  const body = text.trim();
  // Split settings are part of the version: changing CHUNK_SIZE or OVERLAP re-chunks every document.
  const version = crypto.createHash('sha256').update(`${CHUNK_SIZE}:${OVERLAP}:${body}`).digest('hex').slice(0, 12);
  const modifiedTs = Math.floor(Date.parse(modifiedTime) / 1000);
  const ttl = TTL_DAYS[docType];
  const expiresTs = ttl === null ? NEVER : modifiedTs + ttl * 86400;

  split(body).forEach((chunk, i) => {
    out.push({
      json: {
        id: `${docId}#${version}#${String(i).padStart(4, '0')}`,
        docId,
        version,
        text: chunk,
        metadata: { docId, version, docType, source: name, mimeType, modifiedTs, expiresTs, text: chunk },
      },
    });
  });
}
return out;

The hash in the ID pays off whether the document changed or not. If it hasn't, the chunk IDs are identical and the workflow can skip embedding (before calling the embedding model, check that all IDs of this version already exist; the splitter knows how many there are, and an earlier interrupted upsert may have written only some of them). The metadata still needs an update. The hash comes from the content and the split settings only, while modifiedTs and expiresTs come from the modification date, so a price list the owner re-saved without changing the text will otherwise expire on the old date. In Pinecone that's POST /vectors/update with setMetadata for each chunk ID. If the document has changed, the new chunks coexist with the old ones until the cleanup step, which can identify the old ones precisely. If the cleanup fails, the old chunks stay until the next run.

The chunk text is copied into metadata so that query results can be logged and audited without a second lookup. Pinecone limits metadata size per record (40 KB at the time of writing), hence the bound on chunk size.

Replacing a version

After the upsert succeeds, delete the previous versions of each document. With a Pinecone serverless index this is two calls through an HTTP Request node: list record IDs by prefix (GET /vectors/list?prefix=<docId>%23, paginated) and delete by ID (POST /vectors/delete with an ids array). Listing by prefix is available on serverless indexes only. Pinecone also accepts delete by metadata filter (for example docId equal to the document and version different from the current one), which removes the listing step but has a much lower rate limit than delete by ID. The selection logic is the same in both cases:

// n8n Code node: select IDs of superseded versions
// Input: items from "List IDs by prefix", each with { id }
// Current version per document comes from the "Prepare chunks" node
const current = new Map(
  $('Prepare chunks').all().map(i => [i.json.docId, i.json.version])
);

const stale = $input.all()
  .map(i => i.json.id)
  .filter(id => {
    const [docId, version] = id.split('#');
    return current.has(docId) && current.get(docId) !== version;
  });

// Pinecone accepts up to 1000 IDs per delete call
const batches = [];
for (let i = 0; i < stale.length; i += 1000) {
  batches.push({ json: { ids: stale.slice(i, i + 1000) } });
}
return batches;

The order (upsert first, delete afterwards) is deliberate. The reverse order leaves a window in which the document has no chunks at all, and a failed upsert after a successful delete removes the document from the bot's knowledge until the next run. The step also assumes that two runs never process the same document at once. Two parallel runs with different versions can mark each other's chunks as superseded and delete all of them, so runs for the same document have to go one after another.

Reconciling deletions

A separate workflow, run on a schedule, compares the document IDs in the index with the files in the source folder and deletes the orphans.

// n8n Code node: find documents deleted at the source
const live = new Set($('List Drive files').all().map(i => i.json.id));
const stored = new Set($('List index IDs').all().map(i => i.json.id.split('#')[0]));

// Guard: an empty or partial Drive listing (expired credentials, wrong folder,
// missed pagination) would otherwise delete the whole index.
if (live.size === 0 || live.size < stored.size * 0.8) {
  throw new Error(`Drive returned ${live.size} files for ${stored.size} indexed documents, aborting`);
}

return [...stored]
  .filter(docId => !live.has(docId))
  .map(docId => ({ json: { docId } }));

The Drive listing must exclude trashed files and must follow pagination. The 80% threshold is arbitrary. Pick a value that matches how many documents your team realistically removes between runs. The guard compares counts only, so a wrong folder with a similar or larger number of files passes it and deletes everything that folder doesn't contain.

If you move these workflows to a new n8n server (n8n 2.x), check the activeVersionId column. When I moved mine, active = false didn't disable the workflows: as long as activeVersionId isn't null, a workflow starts with the instance, so a test copy with the same credentials would also run the scheduled reconciliation against the production index.

Filtering by freshness at query time

Pinecone metadata range operators ($gt, $gte, $lt, $lte) only work on numbers. ISO date strings won't do, which is why the chunk carries expiresTs as a Unix timestamp. The query body in an HTTP Request node, with expressions enabled:

{
  "vector": {{ JSON.stringify($json.embedding) }},
  "topK": 6,
  "includeMetadata": true,
  "filter": { "expiresTs": { "$gte": {{ Math.floor(Date.now() / 1000) }} } }
}

Computing the expiry at ingestion keeps the policy per document type in one place and the query simple. The cost is that a valid pricing sheet nobody edited for 90 days also drops out. The expired price list disappears from the results, but the query still returns up to six other chunks, so the bot may answer from something that matches worse, for example a changelog with the old price. An empty result only happens once the whole namespace has expired. So alert on a low topScore, or on the absence of a pricing document in the results for a price question (both fields are in the log from the next section), so an owner refreshes or re-confirms the document.

Logging retrieval

Log every retrieval (all of them, sampling defeats the purpose): the query, the returned IDs, their scores, document versions and modification dates. Then you can check which document versions the bot actually received.

// n8n Code node: structured retrieval log
const query = $('User query').first().json.text;
const matches = $('Pinecone query').first().json.matches ?? [];
const now = Date.now() / 1000;

return [{
  json: {
    ts: new Date().toISOString(),
    query,
    matches: matches.map(m => ({
      id: m.id,
      score: m.score,
      docId: m.metadata.docId,
      docType: m.metadata.docType,
      version: m.metadata.version,
      ageDays: Math.round((now - m.metadata.modifiedTs) / 86400),
    })),
    topScore: matches.length ? Math.max(...matches.map(m => m.score)) : null,
    versionsPerDoc: Object.values(
      matches.reduce((acc, m) => {
        (acc[m.metadata.docId] ??= new Set()).add(m.metadata.version);
        return acc;
      }, {})
    ).reduce((max, s) => Math.max(max, s.size), 0),
  },
}];

versionsPerDoc above 1 that persists after the ingestion run has finished means version replacement failed for at least one document. A persistently low topScore for a class of questions means the knowledge base doesn't cover them (with the cosine or dotproduct metric; with euclidean the closest match has the lowest score, so swap Math.max for Math.min and alert on a high value). Score thresholds depend on the embedding model and the corpus. Derive them from the distribution in your own log.

In CookieDomain, my cookie-consent product, every consent change ended with a 500 for 5 days, and the widget swallowed the error in a .catch. The error was in the server log, but nobody was reading that log, and about 88 records were lost. A retrieval log without an alert is no different: the data is there, and nobody sees it.

pgvector as an alternative

If the vector store lives in Postgres, version replacement becomes a transaction and the prefix and hash tricks are unnecessary:

BEGIN;
DELETE FROM rag_chunks WHERE doc_id = $1;
INSERT INTO rag_chunks (doc_id, version, chunk_no, body, embedding, modified_at, expires_at)
VALUES ($1, $2, 0, $3, $4, $5, $6),
       ($1, $2, 1, $7, $8, $5, $6);  -- one row per chunk
COMMIT;

A single query sees the old version or the new one, with no state in between. A reader that runs several queries under READ COMMITTED can still see both across them. The transaction alone also doesn't serialize two parallel replacements of the same document: under READ COMMITTED, if the second DELETE starts before the first transaction commits, it doesn't see the rows that transaction inserted, and both versions stay in the table. A lock at the start of the transaction helps, for example pg_advisory_xact_lock with a key derived from the document ID. Reconciliation becomes an anti-join with a table of live document IDs. The trade-off is operational: you run and tune the database and the vector indexes yourself.

When this is unnecessary

A corpus that's loaded once and never edited, like a fixed set of manuals for a released product, doesn't need versioning or reconciliation. If it's small enough to fit into the model's context window, it can often skip the vector store altogether (cost and latency grow with the length of the context). The mechanisms above are for documents that people edit, rename and archive without thinking about any index.

When I review an existing pipeline, I ask one question: if someone edits a document right now, what happens to the chunks of the previous version, and which node is responsible for that?

end of node