Dynamic RAG: Real-Time Enterprise AI with Live Data Synchronization

Introduction & The Problem

The Solution Concept & Architecture

1. Source Knowledge Base: Your primary data stores (e.g., PostgreSQL, MongoDB, Notion, Confluence, internal APIs).
2. Change Data Capture (CDC) / Event Stream: A mechanism to detect changes in the source. This could be database triggers, logical replication, webhook notifications, or a dedicated CDC tool like Debezium or pg-listen for PostgreSQL.
3. Ingestion & Processing Service: A microservice responsible for consuming change events. It retrieves the affected data, chunks it appropriately, generates new embeddings (or updates existing ones), and manages the vector database.
4. Vector Database: Stores the document chunks and their corresponding embeddings, along with metadata (e.g., ChromaDB, Pinecone, Qdrant).
5. LLM Orchestration Layer: Uses frameworks like LangChain or LlamaIndex to perform retrieval from the vector database and augment LLM prompts.
6. Large Language Model (LLM): Provides the generative AI capabilities.

Step-by-Step Implementation

• Node.js installed
• PostgreSQL running with a knowledge_articles table
• ChromaDB running (e.g., via Docker: docker run -p 8000:8000 chromadb/chroma)

CREATE TABLE knowledge_articles ( 
    id SERIAL PRIMARY KEY, 
    title VARCHAR(255) NOT NULL, 
    content TEXT NOT NULL, 
    last_updated TIMESTAMP DEFAULT CURRENT_TIMESTAMP 
); 

-- Example for a trigger-based approach (simpler for tutorial, but less robust than full CDC) 
CREATE OR REPLACE FUNCTION notify_article_changes() 
RETURNS TRIGGER AS $ 
BEGIN 
    IF (TG_OP = 'DELETE') THEN 
        PERFORM pg_notify('article_changes', json_build_object('operation', TG_OP, 'id', OLD.id)::text); 
        RETURN OLD; 
    ELSE 
        PERFORM pg_notify('article_changes', json_build_object('operation', TG_OP, 'id', NEW.id, 'title', NEW.title, 'content', NEW.content)::text); 
        RETURN NEW; 
    END IF; 
END; 
$ LANGUAGE plpgsql; 

CREATE TRIGGER article_insert_trigger 
AFTER INSERT ON knowledge_articles 
FOR EACH ROW EXECUTE FUNCTION notify_article_changes(); 

CREATE TRIGGER article_update_trigger 
AFTER UPDATE ON knowledge_articles 
FOR EACH ROW EXECUTE FUNCTION notify_article_changes(); 

CREATE TRIGGER article_delete_trigger 
AFTER DELETE ON knowledge_articles 
FOR EACH ROW EXECUTE FUNCTION notify_article_changes();

mkdir dynamic-rag-agent && cd dynamic-rag-agent 
npm init -y 
npm install dotenv pg pg-listen langchain @langchain/chroma @langchain/openai 

# Create .env file 
# PG_CONNECTION_STRING="postgresql://user:password@host:port/database" 
# CHROMA_URL="http://localhost:8000" 
# OPENAI_API_KEY="your_openai_api_key"

// vectorStore.js 
import { ChromaClient } from "chromadb"; 
import { OpenAIEmbeddings } from "@langchain/openai"; 
import { CharacterTextSplitter } from "langchain/text_splitter"; 

const client = new ChromaClient({ path: process.env.CHROMA_URL }); 
const embeddings = new OpenAIEmbeddings({ 
    openAIApiKey: process.env.OPENAI_API_KEY, 
    modelName: "text-embedding-ada-002" 
}); 
const COLLECTION_NAME = "knowledge_base"; 

export async function getOrCreateCollection() { 
    try { 
        return await client.getOrCreateCollection({ name: COLLECTION_NAME, embeddingFunction: embeddings }); 
    } catch (error) { 
        console.error("Error getting/creating Chroma collection:", error); 
        throw error; 
    } 
} 

export async function addDocument(id, title, content) { 
    const collection = await getOrCreateCollection(); 
    const splitter = new CharacterTextSplitter({ 
        separator: "\n\n", 
        chunkSize: 1000, 
        chunkOverlap: 200, 
    }); 
    const docs = await splitter.splitDocuments([ 
        { pageContent: `Title: ${title}\nContent: ${content}`, metadata: { id, title } } 
    ]); 

    const documentContents = docs.map(doc => doc.pageContent); 
    const documentIds = docs.map((_, index) => `${id}-${index}`); // Unique ID for each chunk 
    const documentMetadatas = docs.map(doc => doc.metadata); 

    await collection.add({ 
        documents: documentContents, 
        metadatas: documentMetadatas, 
        ids: documentIds, 
    }); 
    console.log(`Added/Updated document ${id} with ${docs.length} chunks.`); 
} 

export async function updateDocument(id, title, content) { 
    // For updates, we often delete existing chunks and re-add. 
    // ChromaDB doesn't have a direct 'update by document id' for all chunks, 
    // so we manage at a higher level. 
    await deleteDocument(id); 
    await addDocument(id, title, content); 
    console.log(`Updated document ID ${id}`); 
} 

export async function deleteDocument(id) { 
    const collection = await getOrCreateCollection(); 
    // Delete all chunks associated with this original document ID 
    await collection.delete({ where: { id: id } }); 
    console.log(`Deleted document ID ${id} and its chunks.`); 
} 

export async function queryCollection(query) { 
    const collection = await getOrCreateCollection(); 
    const results = await collection.query({ 
        queryTexts: [query], 
        nResults: 5, 
        include: ['documents', 'metadatas'] 
    }); 
    return results.documents[0].map((doc, index) => ({ 
        content: doc, 
        metadata: results.metadatas[0][index] 
    })); 
}

// cdcListener.js 
import "dotenv/config"; 
import pgListen from "pg-listen"; 
import { addDocument, updateDocument, deleteDocument } from "./vectorStore.js"; 

const listener = pgListen({ 
    connectionString: process.env.PG_CONNECTION_STRING, 
    // Add ssl: true if using a cloud PG instance 
}); 

listener.on("error", (error) => { 
    console.error("PostgreSQL listen error:", error); 
    process.exit(1); 
}); 

async function startCdcListener() { 
    await listener.connect(); 
    await listener.listenTo("article_changes"); 
    console.log("Listening for 'article_changes' on PostgreSQL..."); 

    listener.notifications.on("article_changes", async (payload) => { 
        console.log("Received change notification:", payload); 
        const data = JSON.parse(payload); 

        switch (data.operation) { 
            case "INSERT": 
                await addDocument(data.id, data.title, data.content); 
                break; 
            case "UPDATE": 
                await updateDocument(data.id, data.title, data.content); 
                break; 
            case "DELETE": 
                await deleteDocument(data.id); 
                break; 
            default: 
                console.warn("Unknown operation:", data.operation); 
        } 
    }); 

    process.on("beforeExit", async () => { 
        await listener.close(); 
        console.log("PostgreSQL listener closed."); 
    }); 
} 

export { startCdcListener };

// ragAgent.js 
import "dotenv/config"; 
import { OpenAI } from "@langchain/openai"; 
import { PromptTemplate } from "@langchain/core/prompts"; 
import { queryCollection } from "./vectorStore.js"; 

const model = new OpenAI({ 
    openAIApiKey: process.env.OPENAI_API_KEY, 
    temperature: 0.7, 
    modelName: "gpt-3.5-turbo" 
}); 

const QA_PROMPT_TEMPLATE = `You are an AI assistant for a knowledge base. 
Use the following context to answer the question. 
If you don't know the answer, just say that you don't know, don't try to make up an answer. 

Context: 
{context} 

Question: {question} 
Answer:`; 

const qaPrompt = PromptTemplate.fromTemplate(QA_PROMPT_TEMPLATE); 

export async function getAnswer(question) { 
    // 1. Retrieve relevant documents from ChromaDB 
    const retrievedDocs = await queryCollection(question); 
    const context = retrievedDocs.map(doc => doc.content).join("\n\n"); 

    // 2. Format prompt with context 
    const formattedPrompt = await qaPrompt.format({ 
        context: context, 
        question: question 
    }); 

    // 3. Invoke LLM 
    console.log("Invoking LLM with formatted prompt..."); 
    const response = await model.invoke(formattedPrompt); 
    return response; 
}

// index.js 
import { startCdcListener } from "./cdcListener.js"; 
import { getAnswer } from "./ragAgent.js"; 
import readline from "readline"; 

async function main() { 
    await startCdcListener(); 

    const rl = readline.createInterface({ 
        input: process.stdin, 
        output: process.stdout 
    }); 

    console.log("\nDynamic RAG Agent Ready. Type your questions (or 'exit' to quit):\n"); 

    rl.on('line', async (input) => { 
        if (input.toLowerCase() === 'exit') { 
            rl.close(); 
            return; 
        } 

        try { 
            const answer = await getAnswer(input); 
            console.log(`\nAI Answer: ${answer}\n`); 
        } catch (error) { 
            console.error("Error getting AI answer:", error); 
        } 
        console.log("Ask another question (or 'exit'):"); 
    }); 

    rl.on('close', () => { 
        console.log("Exiting Dynamic RAG Agent."); 
        process.exit(0); 
    }); 
} 

main().catch(console.error);

• Ensure ChromaDB and PostgreSQL are running.
• Populate .env with your connection strings and API key.
• Run node index.js.
• In a separate terminal, interact with your PostgreSQL database:
  • INSERT INTO knowledge_articles (title, content) VALUES ('New Feature XYZ', 'This feature allows real-time data streaming.');
  • UPDATE knowledge_articles SET content = 'This feature now supports enhanced encryption for real-time data streaming.' WHERE id = 1;
  • DELETE FROM knowledge_articles WHERE id = 1;
• Observe the cdcListener console output and then query the RAG agent in the index.js terminal to see how the answers reflect the latest data.

Optimization & Best Practices

• Robust CDC: While pg-listen with triggers is shown for simplicity, consider more robust CDC solutions like Debezium or native logical replication for production, ensuring transactional integrity and comprehensive change capture.
• Intelligent Chunking: Experiment with different chunk sizes and overlaps. Context-aware chunking (e.g., keeping sections of a document together) or even hierarchical chunking can improve retrieval quality.
• Hybrid Retrieval: Combine semantic search (vector similarity) with keyword search (e.g., BM25) for more comprehensive results, especially for specific terms or proper nouns.
• Metadata Filtering: Store rich metadata (author, date, department, access level) with your embeddings. Use this metadata to pre-filter search results before vector similarity, greatly improving relevance and enforcing access control.
• Batch Processing for Bursts: If data changes come in high volume, batch updates to the vector store to reduce I/O and API calls, while still maintaining near real-time freshness.
• Error Handling & Retries: Implement robust error handling and retry mechanisms for both the CDC listener and vector store operations to ensure data consistency in the face of transient failures.
• Scalability: For high-volume data, consider distributed vector databases (e.g., Pinecone, Qdrant) and horizontally scalable ingestion services (e.g., Kubernetes deployments).
• Security & Access Control: Integrate enterprise identity management with your RAG system to ensure users only retrieve information they are authorized to see, especially crucial for sensitive internal documents.
• Monitoring Data Freshness: Implement metrics to track the latency between a data change in the source and its reflection in the vector database.

Business Impact & ROI

• Enhanced Decision Making: CEOs and business leaders gain access to AI insights grounded in the most current operational data, leading to more informed and agile strategic decisions.
• Improved Customer Experience: Support agents and self-service portals provide accurate, up-to-date answers, reducing customer frustration, improving satisfaction, and potentially deflecting a significant percentage of support tickets.
• Increased Developer Productivity: Developers spend less time manually updating knowledge bases or debugging issues caused by stale AI responses. The automated synchronization frees them to focus on higher-value tasks.
• Faster Employee Onboarding & Training: New hires can quickly get up to speed with access to a living, breathing knowledge base, accelerating their time to productivity.
• Reduced Operational Costs: By ensuring data freshness, businesses avoid the cost associated with re-training LLMs (which is expensive and slow) and reduce the need for human intervention to correct AI inaccuracies.
• Competitive Advantage: Organizations that can leverage their internal knowledge dynamically with AI gain a significant edge in innovation, responsiveness, and operational efficiency compared to competitors relying on static systems.
• Optimized Resource Utilization: Instead of full re-indexing, the system only processes changed data, leading to more efficient use of computing and API resources.

Conclusion