Resilient RAG pipelines: queue-based ingestion, idempotent indexing and a resilient query path

#1. Ingestion architecture

document store (Blob/S3/SharePoint)
   │  event: created / updated / deleted
   ▼
queue (Service Bus / SQS / Kafka)  ──►  dead-letter queue
   ▼
workers (stateless, autoscaled): parse → chunk → embed → upsert
   ▼
vector index (+ metadata store with the document version)

• A queue decouples bursts (a 10,000-file upload) from embedding-API limits.
• Stateless workers can scale and fail freely. A message is only completed after a successful upsert.
• Dead-letter queue for documents that keep failing (corrupt PDFs, unsupported formats), with alerts and a re-drive tool.

#2. Idempotency

Use deterministic IDs, so reprocessing replaces the existing data instead of duplicating it:

import hashlib
def chunk_id(doc_id: str, chunk_no: int) -> str:
    return hashlib.sha256(f"{doc_id}:{chunk_no}".encode()).hexdigest()[:32]

• Upserts by chunk ID.
• On update: write the new chunks, then delete chunks with a higher chunk_no than the new count (or delete by doc_id and the old version).
• On delete events: delete by the doc_id filter. Run a nightly reconciliation that compares source documents with indexed documents to catch missed events.

#3. Embedding API resilience

• Batch requests (for example 16–64 chunks per call).
• Exponential backoff with jitter on 429 and 5xx, honouring Retry-After.
• Limit worker concurrency so you stay under the TPM quota (it's better to queue than to get throttled).
• Route through the AI gateway so ingestion has its own quota, separate from user traffic.

#4. Index versioning for re-indexing

Changing the chunking or embedding model means a full rebuild. Do it without downtime:

1. Build kb_docs_v2 alongside kb_docs_v1.
2. Run the evaluation set against v2.
3. Switch the alias (kb_docs → v2), via OpenSearch aliases or Qdrant collection aliases. For AI Search, change the index name in app config.
4. Keep v1 for quick rollback, then delete it after a week.

#5. Query path resilience

Never let the model answer without retrieved context when retrieval failed, unless the user is told.

#6. Monitoring

• Queue depth and age of the oldest message (data freshness).
• Ingestion success, failure and dead-letter counts per hour.
• The time from a document changing to it being searchable (an end-to-end freshness SLO, for example 15 minutes).
• Query-side: retrieval latency, empty-result rate and fallback rate.

Tip

Store the document version or hash with every chunk. Then "is this document indexed and current?" becomes one query, and the nightly reconciliation is easy.