RAG with ElasticSearch

In previous article, The Business Challenge: The “Knowledge Gap” in AI, we explored why Large Language Models (LLMs) alone aren’t enough for enterprise applications. While they possess incredible general knowledge, they remain ignorant of your company’s specific data – your internal documentation, latest reports, customer information, and proprietary knowledge.

We also introduced RAG (Retrieval-Augmented Generation) as the solution to this problem. Now, let’s dive into the practical implementation: how to actually build a RAG system using Elasticsearch as the powerful retrieval engine that bridges this knowledge gap.

Why ElasticSearch for RAG?

ElasticSearch isn’t just a search engine anymore – it’s evolved into a complete AI-ready platform. According to Elastic’s documentation, it provides native support for the entire RAG workflow:

The Perfect RAG Architecture with Elasticsearch

1. User Question 
2. Elasticsearch (Semantic Search) 
3. Retrieved Context 
4. LLM 
5. Informed Answer

Elasticsearch serves as the intelligent retrieval backbone that finds the most relevant information from your corporate knowledge base and feeds it to the LLM for response generation.

Implementation Guide: Building RAG with Elasticsearch

Step 1: Data Ingestion and Preparation

Elasticsearch can ingest data from virtually any source:

Document repositories (PDFs, Word, PowerPoint, Excel)
Internal wikis and knowledge bases
CRM and ERP systems
Real-time data streams
Database exports

from elasticsearch import Elasticsearch
from sentence_transformers import SentenceTransformer

# Initialize embedding model and Elasticsearch client
model = SentenceTransformer('all-MiniLM-L6-v2')
es = Elasticsearch(
    "https://your-elastic-cluster:9200",
    api_key="your-api-key"
)

# Process and index documents
def index_document(content, metadata=None):
    # Generate vector embedding
    embedding = model.encode(content).tolist()
    
    document = {
        "content": content,
        "embedding": embedding,
        "metadata": metadata or {},
        "timestamp": "2024-01-15T10:30:00Z"
    }
    
    es.index(index="enterprise-knowledge", document=document)

from elasticsearch import Elasticsearch
from sentence_transformers import SentenceTransformer

# Initialize embedding model and Elasticsearch client
model = SentenceTransformer('all-MiniLM-L6-v2')
es = Elasticsearch(
    "https://your-elastic-cluster:9200",
    api_key="your-api-key"
)

# Process and index documents
def index_document(content, metadata=None):
    # Generate vector embedding
    embedding = model.encode(content).tolist()
    
    document = {
        "content": content,
        "embedding": embedding,
        "metadata": metadata or {},
        "timestamp": "2024-01-15T10:30:00Z"
    }
    
    es.index(index="enterprise-knowledge", document=document)

Step 2: Semantic Search Implementation

Elasticsearch’s dense_vector field type enables powerful semantic search:

# Create index with vector mapping
index_mapping = {
    "mappings": {
        "properties": {
            "content": {"type": "text"},
            "embedding": {
                "type": "dense_vector",
                "dims": 384,
                "similarity": "cosine"
            },
            "metadata": {"type": "object"},
            "timestamp": {"type": "date"}
        }
    }
}

es.indices.create(index="enterprise-knowledge", body=index_mapping)

# Create index with vector mapping
index_mapping = {
    "mappings": {
        "properties": {
            "content": {"type": "text"},
            "embedding": {
                "type": "dense_vector",
                "dims": 384,
                "similarity": "cosine"
            },
            "metadata": {"type": "object"},
            "timestamp": {"type": "date"}
        }
    }
}

es.indices.create(index="enterprise-knowledge", body=index_mapping)

Step 3: Intelligent Retrieval

def retrieve_context(user_query, index="enterprise-knowledge", top_k=5):
    # Generate query embedding
    query_embedding = model.encode(user_query).tolist()
    
    # Semantic search query
    search_body = {
        "query": {
            "script_score": {
                "query": {"match_all": {}},
                "script": {
                    "source": """
                    cosineSimilarity(params.query_vector, 'embedding') + 1.0
                    """,
                    "params": {"query_vector": query_embedding}
                }
            }
        },
        "size": top_k,
        "_source": ["content", "metadata"]
    }
    
    response = es.search(index=index, body=search_body)
    
    # Return relevant contexts with scores
    return [
        {
            "content": hit["_source"]["content"],
            "score": hit["_score"],
            "metadata": hit["_source"].get("metadata", {})
        }
        for hit in response["hits"]["hits"]
    ]

def retrieve_context(user_query, index="enterprise-knowledge", top_k=5):
    # Generate query embedding
    query_embedding = model.encode(user_query).tolist()
    
    # Semantic search query
    search_body = {
        "query": {
            "script_score": {
                "query": {"match_all": {}},
                "script": {
                    "source": """
                    cosineSimilarity(params.query_vector, 'embedding') + 1.0
                    """,
                    "params": {"query_vector": query_embedding}
                }
            }
        },
        "size": top_k,
        "_source": ["content", "metadata"]
    }
    
    response = es.search(index=index, body=search_body)
    
    # Return relevant contexts with scores
    return [
        {
            "content": hit["_source"]["content"],
            "score": hit["_score"],
            "metadata": hit["_source"].get("metadata", {})
        }
        for hit in response["hits"]["hits"]
    ]

Step 4: Integration with LLM

def generate_rag_response(user_question):
    # Retrieve relevant context from Elasticsearch
    contexts = retrieve_context(user_question)
    
    # Build augmented prompt
    context_text = "\n\n".join([
        f"Document {i+1} (Relevance: {ctx['score']:.2f}):\n{ctx['content']}"
        for i, ctx in enumerate(contexts)
    ])
    
    augmented_prompt = f"""Based on the following enterprise documents, answer the user's question.

Relevant Documents:
{context_text}

User Question: {user_question}

Instructions:
- Use only the information from the provided documents
- If the documents don't contain relevant information, say so
- Cite which document(s) you used for your answer
- Be precise and business-focused

Answer:"""
    
    # Call LLM (example with OpenAI, but works with any LLM)
    from openai import OpenAI
    client = OpenAI()
    
    response = client.chat.completions.create(
        model="gpt-4",
        messages=[{"role": "user", "content": augmented_prompt}],
        temperature=0.1
    )
    
    return response.choices[0].message.content

def generate_rag_response(user_question):
    # Retrieve relevant context from Elasticsearch
    contexts = retrieve_context(user_question)
    
    # Build augmented prompt
    context_text = "\n\n".join([
        f"Document {i+1} (Relevance: {ctx['score']:.2f}):\n{ctx['content']}"
        for i, ctx in enumerate(contexts)
    ])
    
    augmented_prompt = f"""Based on the following enterprise documents, answer the user's question.

Relevant Documents:
{context_text}

User Question: {user_question}

Instructions:
- Use only the information from the provided documents
- If the documents don't contain relevant information, say so
- Cite which document(s) you used for your answer
- Be precise and business-focused

Answer:"""
    
    # Call LLM (example with OpenAI, but works with any LLM)
    from openai import OpenAI
    client = OpenAI()
    
    response = client.chat.completions.create(
        model="gpt-4",
        messages=[{"role": "user", "content": augmented_prompt}],
        temperature=0.1
    )
    
    return response.choices[0].message.content

Advanced Elasticsearch Features for RAG

Hybrid Search (BM25 + Semantic)

# Combine traditional keyword search with vector search
hybrid_query = {
    "query": {
        "bool": {
            "should": [
                # Semantic search
                {
                    "script_score": {
                        "query": {"match_all": {}},
                        "script": {
                            "source": "cosineSimilarity(params.query_vector, 'embedding') + 1.0",
                            "params": {"query_vector": query_embedding}
                        }
                    }
                },
                # Keyword search
                {
                    "match": {
                        "content": {
                            "query": user_query,
                            "boost": 0.5
                        }
                    }
                }
            ]
        }
    }
}

# Combine traditional keyword search with vector search
hybrid_query = {
    "query": {
        "bool": {
            "should": [
                # Semantic search
                {
                    "script_score": {
                        "query": {"match_all": {}},
                        "script": {
                            "source": "cosineSimilarity(params.query_vector, 'embedding') + 1.0",
                            "params": {"query_vector": query_embedding}
                        }
                    }
                },
                # Keyword search
                {
                    "match": {
                        "content": {
                            "query": user_query,
                            "boost": 0.5
                        }
                    }
                }
            ]
        }
    }
}

Filtering and Security

# Add access control filters
secure_query = {
    "query": {
        "bool": {
            "must": [
                {
                    "script_score": {
                        "query": {"match_all": {}},
                        "script": {
                            "source": "cosineSimilarity(params.query_vector, 'embedding') + 1.0",
                            "params": {"query_vector": query_embedding}
                        }
                    }
                }
            ],
            "filter": [
                {
                    "terms": {
                        "metadata.department": ["engineering", "hr"]  # User's access rights
                    }
                }
            ]
        }
    }
}

# Add access control filters
secure_query = {
    "query": {
        "bool": {
            "must": [
                {
                    "script_score": {
                        "query": {"match_all": {}},
                        "script": {
                            "source": "cosineSimilarity(params.query_vector, 'embedding') + 1.0",
                            "params": {"query_vector": query_embedding}
                        }
                    }
                }
            ],
            "filter": [
                {
                    "terms": {
                        "metadata.department": ["engineering", "hr"]  # User's access rights
                    }
                }
            ]
        }
    }
}

Real-World Business Benefits

Immediate Value

Faster employee onboarding – New hires can query internal knowledge naturally
Reduced support costs – Automated, accurate responses based on documentation
Better decision making – AI assistants with access to latest company data

Enterprise Ready

Scalability – Handle billions of documents across multiple departments
Security – Document-level access control and audit trails
Real-time updates – New information immediately available to the AI

Best Practices for Production

Chunking Strategy
- Split large documents into logical chunks (500-1000 words)
- Maintain context between chunks with overlap
Metadata Enrichment
- Add department, security level, document type metadata
- Include timestamps for temporal relevance
Performance Optimization
- Use approximate nearest neighbor for large datasets
- Implement caching for frequent queries
- Monitor query latency and accuracy
Quality Assurance
- Implement feedback loops for result quality
- Regularly update and prune the knowledge base
- A/B test different retrieval strategies

Getting Started

Ready to implement RAG with Elasticsearch? Start with:

Identify your key knowledge sources – What documents would most benefit from AI accessibility?
Set up Elasticsearch – Cloud or self-managed, with machine learning nodes enabled
Implement the ingestion pipeline – Automate document processing and embedding generation
Integrate with your LLM of choice – OpenAI, Anthropic, open-source models, or Azure OpenAI

The combination of Elasticsearch’s powerful retrieval capabilities with modern LLMs creates an AI system that truly understands your business context. It’s not just about having smart AI – it’s about having informed AI that can leverage your organization’s collective knowledge.