back to topology
ai-agents
artykuł··zaktualizowano ·9 min czytania

Zasilanie RAG w n8n: wersje dokumentów, kasowanie i metadane świeżości

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

Chatbot z RAG odpowiada na podstawie tego, co zwróci baza wektorowa. Jeśli w bazie leżą dwie wersje tego samego cennika, retriever wybierze chunk z wyższym wynikiem podobieństwa, a ten wynik nic nie mówi o tym, która wersja obowiązuje. Model odpowiada zgodnie z kontekstem. Odpowiedź brzmi dobrze i powołuje się na prawdziwy dokument, tyle że jest błędna. W logach nie ma błędu, detektor halucynacji milczy, bo model niczego nie wymyślił.

Usterka siedzi w przepływie zasilania, nie w modelu ani w prompcie. Poniżej opisuję niezmienniki, których pipeline zasilania powinien pilnować, i sposób na ich wdrożenie w n8n z Pinecone. Z Qdrantem, Weaviate czy pgvectorem zasady są te same.

Skąd biorą się stare chunki

Baza wektorowa nie zna pojęcia wersji dokumentu. Trzyma rekordy pod identyfikatorami, a upsert podmienia rekord tylko przy zgodnym ID. Cała reszta się kumuluje.

Najczęściej duplikaty biorą się z ID wyliczanego z nazwy pliku albo z treści. Nowa wersja zapisana jako cennik-2026-v2.pdf albo wersja ze zmienionym pierwszym akapitem dostaje nowe ID, a stare rekordy zostają.

Krótsza nowa wersja robi to samo, tylko mniej widocznie. Wersja 1 dała 12 chunków, wersja 2 daje 9. Upsert po docId + indeks nadpisze chunki 0–8, a 9–11 ze starej wersji zostaną.

Do tego dochodzi usuwanie u źródła. Przeniesienie pliku do archiwum albo skasowanie go na Google Drive niczego nie zmienia w bazie wektorowej, więc bez jawnego delete chunki da się wyszukać w nieskończoność.

Typowa pierwsza wersja przepływu (trigger Drive, splitter tekstu, upsert do Pinecone) ma wszystkie te problemy naraz:

// n8n Code node: naiwne przygotowanie chunków
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 },
  },
}));

ID zależy od treści strony, jedyną metadaną jest nazwa pliku, a w całym przepływie nie ma kroku kasowania.

Niezmiennik

Dla każdego dokumentu w folderze źródłowym indeks zawiera dokładnie chunki jego bieżącej wersji i nic więcej. Z tego wynikają dalsze decyzje:

  • ID dokumentu pochodzi z systemu źródłowego (file ID z Drive), bo file ID nie zmienia się przy zmianie nazwy ani przy wgraniu nowej rewizji;
  • ID chunka zaczyna się od ID dokumentu i hasha treści, więc wszystkie chunki dokumentu da się znaleźć po prefiksie, a starą wersję odróżnić od nowej;
  • podmiana dokumentu to zapis nowych chunków, a potem usunięcie wszystkich chunków z tym samym prefiksem dokumentu i innym hashem;
  • uzgadnianie uruchamiane z harmonogramu usuwa dokumenty, których nie ma już u źródła;
  • świeżość jest zapisana jako liczby, bo na liczbach działają filtry zakresowe w metadanych.

Przygotowanie chunków

Poniższy węzeł działa w trybie „Run Once for All Items”. Oczekuje jednego itemu na plik z Drive z polami id, name, modifiedTime, mimeType, wyciągniętym text i docType (na przykład wyliczonym z podfolderu). Na self-hosted n8n require('crypto') działa tylko wtedy, gdy instancja ma ustawione NODE_FUNCTION_ALLOW_BUILTIN=crypto (przy zewnętrznych task runnerach zmienną ustawia się na runnerze, główna instancja jej nie czyta).

Self-hosted n8n ma więcej takich zachowań, które wychodzą dopiero w praktyce. Przy przenosinach mojego n8n na nowy serwer active = false wcale nie wyłączało workflowów. Prawdziwym wyłącznikiem było activeVersionId i dopiero po jego wyczyszczeniu w logu pojawiło się „Processed 8 draft workflows”. Telegram Trigger przejmuje webhooka, zamiast się do niego dołączyć: moja kopia testowa żyła 41 sekund i przez ten czas „wypisała” produkcję z Telegrama. Naprawił to dopiero restart produkcyjnej instancji. Test klucza szyfrowania poszedł czysto, odszyfrowało się 14 z 14 credentiali.

Po tych przenosinach goniłem też fałszywy alarm. Błąd „Error in sub-node Postgres Chat Memory” wyglądał na skutek migracji, a wziął się z mojego własnego testu, który wysyłał pole sessionId zamiast session_id. Teraz, zanim uznam coś za skutek migracji, sprawdzam, jak to działało przed nią.

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

const CHUNK_SIZE = 1500; // znaki
const OVERLAP = 200;
// Liczba dni od ostatniej modyfikacji, po której dokument przestaje być zwracany.
// null oznacza, że dany typ dokumentu nie wygasa.
const TTL_DAYS = { pricing: 90, changelog: 180, adr: null, general: null };
const NEVER = 4102444800; // 2100-01-01 jako Unix timestamp

function split(text) {
  const chunks = [];
  for (let start = 0; start < text.length; start += CHUNK_SIZE - OVERLAP) {
    chunks.push(text.slice(start, start + CHUNK_SIZE));
    if (start + CHUNK_SIZE >= text.length) break;
  }
  return chunks;
}

const out = [];
for (const item of $input.all()) {
  const { id: docId, name, modifiedTime, mimeType, text } = item.json;
  if (typeof text !== 'string' || text.trim().length < 50) continue;

  const docType = item.json.docType ?? 'general';
  const body = text.trim();
  const version = crypto.createHash('sha256').update(body).digest('hex').slice(0, 12);
  const modifiedTs = Math.floor(Date.parse(modifiedTime) / 1000);
  const ttl = TTL_DAYS[docType] ?? null;
  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;

Hash w ID przydaje się niezależnie od tego, czy dokument się zmienił. Jeśli nie, ID chunków są identyczne i przepływ może całkiem pominąć embedding (przed wywołaniem modelu embeddingów sprawdź, czy istnieje jakiekolwiek ID z prefiksem docId#version#). Jeśli się zmienił, nowe chunki współistnieją ze starymi tylko na czas przebiegu, a krok sprzątania dokładnie wie, które są stare.

Tekst chunka trafia też do metadanych, żeby wyniki zapytań dało się logować i audytować bez drugiego odczytu. Pinecone ogranicza rozmiar metadanych rekordu (w chwili pisania 40 KB), stąd ograniczony rozmiar chunka.

Podmiana wersji

Po udanym upsercie usuń poprzednie wersje każdego dokumentu. Przy indeksie serverless w Pinecone to dwa wywołania z węzła HTTP Request: listowanie ID rekordów po prefiksie (GET /vectors/list?prefix=<docId>%23, ze stronicowaniem) i kasowanie po ID (POST /vectors/delete z tablicą ids). Listowanie po prefiksie działa tylko w indeksach serverless. Pinecone przyjmuje też delete po filtrze metadanych (na przykład docId równe dokumentowi i version różne od bieżącej), co usuwa krok listowania, ale ma znacznie niższy limit zapytań niż delete po ID. Logika wyboru jest w obu przypadkach ta sama:

// n8n Code node: wybór ID nieaktualnych wersji
// Wejście: itemy z węzła "List IDs by prefix", każdy z polem { id }
// Bieżąca wersja każdego dokumentu pochodzi z węzła "Prepare chunks"
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 przyjmuje do 1000 ID w jednym wywołaniu delete
const batches = [];
for (let i = 0; i < stale.length; i += 1000) {
  batches.push({ json: { ids: stale.slice(i, i + 1000) } });
}
return batches;

Najpierw upsert, potem delete, i to celowo. Odwrotna kolejność zostawia okno, w którym dokument nie ma żadnych chunków, a nieudany upsert po udanym delete wycina dokument z wiedzy bota do następnego przebiegu.

Uzgadnianie usunięć

Osobny przepływ, uruchamiany z harmonogramu, porównuje ID dokumentów w indeksie z plikami w folderze źródłowym i kasuje sieroty.

// n8n Code node: dokumenty usunięte u źródła
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]));

// Zabezpieczenie: pusta albo niepełna lista z Drive (wygasłe credentials,
// zły folder, pominięte stronicowanie) skasowałaby cały indeks.
if (live.size === 0 || live.size < stored.size * 0.8) {
  throw new Error(`Drive zwrócił ${live.size} plików przy ${stored.size} dokumentach w indeksie, przerywam`);
}

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

Listowanie z Drive musi pomijać pliki w koszu i obsługiwać stronicowanie. Próg 80% jest umowny. Dobierz go do tego, ile dokumentów zespół realnie usuwa między przebiegami.

Filtr świeżości przy zapytaniu

Operatory zakresowe w metadanych Pinecone ($gt, $gte, $lt, $lte) działają tylko na liczbach. Daty w formacie ISO odpadają, dlatego chunk ma expiresTs jako Unix timestamp. Treść zapytania w węźle HTTP Request z włączonymi expressions:

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

Gdy datę wygaśnięcia liczy się przy zasilaniu, polityka dla każdego typu dokumentu siedzi w jednym miejscu, a zapytanie zostaje proste. Cena za to jest taka, że poprawny cennik, którego nikt nie ruszał od 90 dni, też wypada. Przy danych zależnych od czasu lepiej, żeby system psuł się właśnie tak. Bot nie znajduje kontekstu i mówi, że nie wie, a to łatwiej wykryć niż złą cenę. Do TTL dodaj alert na zapytania bez wyników, żeby właściciel dokumentu go odświeżył albo potwierdził.

Logowanie wyszukiwań

Loguj każde wyszukiwanie (wszystkie, próbka tu nie wystarczy): zapytanie, zwrócone ID, wyniki podobieństwa, wersje dokumentów i daty modyfikacji. Każdą z opisanych wyżej awarii znajdziesz wtedy jednym zapytaniem do tego logu.

Jak cicho potrafią znikać dane, przekonałem się na własnym produkcie, CookieDomain, systemie zgód cookies. Przez 5 dni każda zmiana zgody kończyła się błędem 500 i około 88 wpisów przepadło bez śladu, bo widget połykał błąd w .catch.

// n8n Code node: ustrukturyzowany log wyszukiwania
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,
      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 większe od 1 znaczy, że podmiana wersji nie zadziałała dla co najmniej jednego dokumentu. Stale niski topScore dla jakiejś grupy pytań znaczy, że baza wiedzy ich nie pokrywa. Progi podobieństwa zależą od modelu embeddingów i korpusu. Wyznacz je z rozkładu we własnym logu, bo liczba z cudzej konfiguracji się nie przeniesie.

pgvector jako alternatywa

Jeśli baza wektorowa siedzi w Postgresie, podmiana wersji staje się transakcją, a sztuczki z prefiksem i hashem są zbędne:

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);  -- jeden wiersz na chunk
COMMIT;

Odczyt widzi starą wersję albo nową. Nie ma chwili, w której są obie albo żadna. Uzgadnianie sprowadza się do anti-joina z tabelą aktywnych ID dokumentów. Za to bazę i indeksy wektorowe utrzymujesz i stroisz sam.

Kiedy to zbędne

Korpus wgrany raz i nieedytowany, na przykład zamknięty zestaw instrukcji do wydanej wersji produktu, nie potrzebuje wersjonowania ani uzgadniania. Korpus, który mieści się w oknie kontekstu modelu, nie potrzebuje bazy wektorowej w ogóle. Opisane mechanizmy są potrzebne tam, gdzie dokumenty edytują, przemianowują i archiwizują ludzie, którzy o żadnym indeksie nie myślą. Na współdzielonym dysku dotyczy to większości plików.

Przy przeglądzie istniejącego pipeline'u większość problemów wychodzi po jednym pytaniu: jeśli ktoś w tej chwili zmieni dokument, co stanie się z chunkami poprzedniej wersji i który węzeł za to odpowiada?

koniec artykułu