Event-Driven RAG Pipeline

When to use it

• Source documents change continuously and answers must reflect changes within minutes.
• Ingest volume is bursty enough that synchronous indexing would stall writers.
• More than one agent retrieves from the same corpus, so the index is a shared asset rather than an implementation detail of a single agent.

Diagram

 sources                ingest                       serving
┌──────────┐        ┌──────────────┐            ┌────────────────────┐
│ CMS      │──┐     │ change queue │            │ Velerion agent     │
│ Drive    │──┼────►│ (at-least-   │            │  ┌──────────────┐  │
│ Tickets  │──┘     │  once)       │            │  │ retrieve MCP │  │
└──────────┘        └──────┬───────┘            │  └──────┬───────┘  │
                           │                    └─────────┼──────────┘
                    ┌──────▼───────┐                      │
                    │ chunk + embed│                      │
                    │ worker pool  │                      │
                    └──────┬───────┘                      │
                           │  upsert by stable doc id     │
                    ┌──────▼───────────────────┐          │
                    │ vector index + metadata  │◄─────────┘
                    │ store (ACL-tagged)       │   filtered by caller ACL
                    └──────┬───────────────────┘
                           │
                    ┌──────▼───────┐
                    │ dead letter  │──► operator review
                    └──────────────┘

Components

Component	Responsibility
Change queue	Decouples source systems from indexing. At-least-once delivery, so workers must be idempotent.
Chunk + embed workers	Normalise, chunk, embed and upsert. Keyed by a stable document ID so replays overwrite rather than duplicate.
Vector index	Nearest-neighbour search plus metadata filters. Every chunk carries the ACL of its source document.
Retrieve MCP server	The only path an agent has to the index. Applies the caller’s ACL as a hard filter before ranking.
Dead letter queue	Documents that failed repeatedly. Needs an owner and an alert, or it silently becomes a data-loss channel.

Idempotent upsert

workers/index-document.ts

import { createHash } from 'node:crypto';

export async function indexDocument(event: ChangeEvent) {
  const docId = event.sourceId; // stable across revisions
  const chunks = chunk(await fetchDocument(event));

  await index.transaction(async (tx) => {
    // Replace the whole document atomically: partial updates leave orphans
    // that keep answering queries with stale text.
    await tx.deleteByFilter({ docId });
    await tx.upsert(
      chunks.map((c, i) => ({
        id: `${docId}:${i}:${createHash('sha256').update(c.text).digest('hex').slice(0, 12)}`,
        docId,
        vector: c.embedding,
        text: c.text,
        acl: event.acl,
        sourceRevision: event.revision,
      })),
    );
  });
}

Retrieval with ACL enforcement

import { defineAgent, mcp } from '@velerion/sdk';

export default defineAgent({
  name: 'knowledge-assistant',
  model: 'claude-opus-5',
  instructions: `
    Answer only from retrieved passages. Cite the document title for every claim.
    If retrieval returns nothing relevant, say that you do not know.
  `,
  tools: [
    mcp('velerion/vector-retrieve', {
      connection: 'kb-index',
      tools: ['search'],
      // The caller's identity is forwarded to the server, which applies it as a
      // pre-filter. Never filter after ranking: the model would already have
      // seen the passages it is not allowed to see.
      forwardIdentity: true,
    }),
  ],
});

Danger

ACL filtering must happen inside the retrieval server, before results reach the model. Post-filtering in agent code is a data leak: the text has already entered the context window.

Failure modes

Failure	Symptom	Mitigation
Embedding provider outage	Queue depth climbs, answers go stale	Alert on queue age, not queue length. Retry with backoff; the queue is the buffer.
Poison document	One document retries forever	Bounded retries, then dead letter with an alert.
Duplicate delivery	Duplicate chunks skew ranking	Content-hash chunk IDs plus delete-then-upsert per document.
ACL change not propagated	User sees a document they lost access to	Treat ACL changes as change events in their own right; do not wait for a content edit.
Index and source diverge silently	Confident answers from deleted text	Nightly reconciliation job comparing source IDs against indexed doc IDs.

Trade-offs

• Eventual consistency. There is a window, typically seconds to minutes, in which the index disagrees with the source. If your use case cannot tolerate that, read through to the source instead of indexing.
• Operational surface. A queue, a worker pool, an index and a dead letter queue are four things to monitor. A single agent over a small static corpus does not justify them.
• Chunking is a lasting commitment. Changing the strategy means a full re-embed of the corpus, so validate it on a representative sample before the first bulk load.
• ACL-tagged chunks couple the index to your permission model. A permission-model change becomes an index migration.