Production-Grade RAG: A Blueprint for Scalable, Real-Time Architecture

// src/ingestion-worker.ts
import { Kafka } from 'kafkajs';
import { EmbeddingService } from './embeddingService';
import { VectorDbClient } from './vectorDbClient';
 
interface DocumentChangeEvent {
  id: string;
  content: string;
  metadata: Record<string, any>;
  type: 'upsert' | 'delete';
  timestamp: string;
}
 
const kafka = new Kafka({ brokers: [process.env.KAFKA_BROKERS || 'localhost:9092'] });
const consumer = kafka.consumer({ groupId: 'rag-ingestion-group' });
 
const embeddingService = new EmbeddingService();
const vectorDbClient = new VectorDbClient();
 
async function startIngestionWorker() {
  await consumer.connect();
  await consumer.subscribe({ topic: 'document-changes', fromBeginning: false });
 
  await consumer.run({
    eachMessage: async ({ topic, partition, message, heartbeat }) => {
      if (!message.value) return;
 
      try {
        const event: DocumentChangeEvent = JSON.parse(message.value.toString());
 
        if (event.type === 'upsert') {
          const embedding = await embeddingService.getEmbedding(event.content);
 
          await vectorDbClient.upsert({
            id: event.id,
            vector: embedding,
            metadata: { ...event.metadata, ingestion_timestamp: event.timestamp },
            content: event.content
          });
        } else if (event.type === 'delete') {
          await vectorDbClient.delete(event.id);
        }
        await heartbeat();
      } catch (error) {
        console.error(`Ingestion error at offset ${message.offset}:`, error);
      }
    },
  });
}
 
startIngestionWorker().catch(err => {
  console.error("Ingestion worker startup failed:", err);
  process.exit(1);
});

// src/vectorDbClient.ts
import { QdrantClient } from '@qdrant/qdrant-sdk';
import { PointStruct, Filter } from '@qdrant/qdrant-sdk/dist/qdrant_client';
 
export class VectorDbClient {
  private client: QdrantClient;
  private collectionName: string;
 
  constructor(
    host: string = process.env.QDRANT_HOST || 'localhost',
    port: number = parseInt(process.env.QDRANT_PORT || '6333', 10),
    collectionName: string = process.env.QDRANT_COLLECTION || 'rag_documents'
  ) {
    this.client = new QdrantClient({ host, port });
    this.collectionName = collectionName;
  }
 
  async upsert(document: { id: string; vector: number[]; metadata: Record<string, any>; content?: string }) {
    const point: PointStruct = {
      id: document.id,
      vector: document.vector,
      payload: { ...document.metadata, content: document.content || null }
    };
    await this.client.upsert(this.collectionName, {
      wait: true,
      points: [point],
    });
  }
 
  async query(
    queryVector: number[],
    limit: number = 5,
    filters?: Filter
  ): Promise<Array<{ id: string | number; score: number; payload: Record<string, any> }>> {
    const searchResult = await this.client.search(this.collectionName, {
      vector: queryVector,
      limit: limit,
      filter: filters,
      with_payload: true,
      with_vectors: false,
    });
    return searchResult.map(hit => ({
      id: hit.id,
      score: hit.score,
      payload: hit.payload || {},
    }));
  }
 
  async delete(id: string) {
    await this.client.delete(this.collectionName, { pointsSelector: { points: [id] } });
  }
}

// src/retrievalService.ts
import { EmbeddingService } from './embeddingService';
import { VectorDbClient } from './vectorDbClient';
import { LLMClient } from './llmClient';
import { RedisClient } from './redisClient';
import crypto from 'crypto';
 
export class RetrievalService {
  private embeddingService: EmbeddingService;
  private vectorDbClient: VectorDbClient;
  private llmClient: LLMClient;
  private redis: RedisClient;
 
  constructor() {
    this.embeddingService = new EmbeddingService();
    this.vectorDbClient = new VectorDbClient();
    this.llmClient = new LLMClient();
    this.redis = new RedisClient();
  }
 
  private generateCacheKey(input: string): string {
    return crypto.createHash('sha256').update(input).digest('hex');
  }
 
  private async getCached<T>(key: string, fetchFn: () => Promise<T>, ttlSeconds: number): Promise<T> {
    const cached = await this.redis.get(key);
    if (cached) {
      return JSON.parse(cached) as T;
    }
    const result = await fetchFn();
    await this.redis.set(key, JSON.stringify(result), ttlSeconds);
    return result;
  }
 
  async retrieveAndGenerate(
    userQuery: string,
    filters?: Record<string, any>,
    numDocuments: number = 5
  ): Promise<string> {
    // 1. Cached Query Embedding
    const embeddingCacheKey = this.generateCacheKey(`embedding:${userQuery}`);
    const queryEmbedding = await this.getCached(
      embeddingCacheKey,
      () => this.embeddingService.getEmbedding(userQuery),
      86400 // 24 hours
    );
 
    // 2. Cached Vector Retrieval
    const filterString = filters ? JSON.stringify(filters) : '';
    const retrievalCacheKey = this.generateCacheKey(`retrieval:${userQuery}:${filterString}:${numDocuments}`);
    const retrievedDocs = await this.getCached(
      retrievalCacheKey,
      async () => {
        const hits = await this.vectorDbClient.query(queryEmbedding, numDocuments, filters);
        return hits.map(hit => ({
          content: hit.payload.content as string,
          metadata: hit.payload as Record<string, any>
        }));
      },
      300 // 5 minutes
    );
 
    if (retrievedDocs.length === 0) {
      return await this.llmClient.generateResponse([{ role: 'user', content: userQuery }]);
    }
 
    // 3. Prompt Assembly & Cached Generation
    const context = retrievedDocs.map(doc => doc.content).join('\n\n---\n\n');
    const promptMessages = [
      { role: 'system' as const, content: 'You are an accurate technical assistant. Use only the provided context.' },
      { role: 'user' as const, content: `Context:\n${context}\n\nQuestion: ${userQuery}` }
    ];
 
    const sortedDocIds = retrievedDocs.map(d => d.metadata.id || '').sort().join(',');
    const responseCacheKey = this.generateCacheKey(`response:${userQuery}:${sortedDocIds}:${filterString}`);
 
    return await this.getCached(
      responseCacheKey,
      () => this.llmClient.generateResponse(promptMessages),
      120 // 2 minutes
    );
  }
}

// src/telemetry.ts
type MetricType = 'counter' | 'gauge' | 'histogram';
 
export function logMetric(name: string, value: number, type: MetricType, labels?: Record<string, string>) {
  const labelStr = labels ? Object.entries(labels).map(([k, v]) => `${k}="${v}"`).join(',') : '';
  console.log(`METRIC: ${name}{${labelStr}} ${value} (type: ${type}, timestamp: ${Date.now()})`);
}