Elasticsearch Onboarding

User Question
     │
     ▼
┌─────────────┐     ┌──────────────────┐     ┌─────────────┐
│  Embed query │────▶│  Elasticsearch   │────▶│  LLM (GPT,  │
│              │     │  kNN retrieval   │     │  Claude, etc)│
└─────────────┘     └──────────────────┘     └─────────────┘
                           │                        │
                     Top-k chunks              Answer + sources

def chunk_documents(documents: list[dict], chunk_size: int = 500, overlap: int = 100) -> list[dict]:
    """Split documents into overlapping chunks with metadata."""
    chunks = []
    for doc in documents:
        text = doc["content"]
        words = text.split()
        for i in range(0, len(words), chunk_size - overlap):
            chunk_text = " ".join(words[i:i + chunk_size])
            if not chunk_text.strip():
                continue
            chunks.append({
                "content": chunk_text,
                "source_title": doc.get("title", ""),
                "source_url": doc.get("url", ""),
                "chunk_index": len(chunks),
                "parent_doc_id": doc.get("id", ""),
            })
    return chunks

from langchain.text_splitter import RecursiveCharacterTextSplitter

splitter = RecursiveCharacterTextSplitter(
    chunk_size=1000,
    chunk_overlap=200,
    separators=["\n\n", "\n", ". ", " "]
)
chunks = splitter.split_documents(documents)

PUT /knowledge-base
{
  "mappings": {
    "properties": {
      "content": { "type": "text" },
      "embedding": {
        "type": "dense_vector",
        "dims": 1024,
        "index": true,
        "similarity": "cosine"
      },
      "source_title": { "type": "text", "fields": { "keyword": { "type": "keyword" } } },
      "source_url": { "type": "keyword" },
      "section_header": { "type": "text" },
      "parent_doc_id": { "type": "keyword" },
      "chunk_index": { "type": "integer" },
      "created_at": { "type": "date" }
    }
  }
}

PUT _ingest/pipeline/embed-knowledge-base
{
  "processors": [
    {
      "inference": {
        "model_id": "<YOUR_EMBEDDING_MODEL>",
        "input_output": [
          {
            "input_field": "content",
            "output_field": "embedding"
          }
        ]
      }
    }
  ]
}

from elasticsearch import Elasticsearch, helpers

es = Elasticsearch(cloud_id="...", api_key="...")

def index_chunks(chunks: list[dict]) -> tuple[int, list]:
    actions = [
        {"_index": "knowledge-base", "_source": chunk, "pipeline": "embed-knowledge-base"}
        for chunk in chunks
    ]
    return helpers.bulk(es, actions, raise_on_error=False, raise_on_exception=False)

GET /knowledge-base/_search
{
  "knn": {
    "field": "embedding",
    "query_vector_builder": {
      "text_embedding": {
        "model_id": "<YOUR_EMBEDDING_MODEL>",
        "model_text": "How do I configure index mappings?"
      }
    },
    "k": 5,
    "num_candidates": 50
  },
  "_source": ["content", "source_title", "source_url", "section_header"]
}

POST /knowledge-base/_search
{
  "size": 5,
  "query": {
    "bool": {
      "should": [
        { "match": { "content": "configure index mappings" } },
        {
          "knn": {
            "field": "embedding",
            "query_vector_builder": {
              "text_embedding": {
                "model_id": "<YOUR_EMBEDDING_MODEL>",
                "model_text": "How do I configure index mappings?"
              }
            },
            "k": 5,
            "num_candidates": 50
          }
        }
      ]
    }
  },
  "rank": { "rrf": {} },
  "_source": ["content", "source_title", "source_url"]
}

GET /knowledge-base/_search
{
  "knn": {
    "field": "embedding",
    "query_vector_builder": {
      "text_embedding": {
        "model_id": "<YOUR_EMBEDDING_MODEL>",
        "model_text": "How do I configure mappings?"
      }
    },
    "k": 5,
    "num_candidates": 50,
    "filter": {
      "term": { "source_title.keyword": "Elasticsearch Guide" }
    }
  }
}

from openai import OpenAI  # or the SDK for the developer's chosen LLM
from elasticsearch import Elasticsearch

es = Elasticsearch(cloud_id="...", api_key="...")
llm = OpenAI()  # replace with the developer's LLM provider

def ask(question: str, k: int = 5) -> dict:
    # 1. Retrieve relevant chunks
    resp = es.search(
        index="knowledge-base",
        knn={
            "field": "embedding",
            "query_vector_builder": {
                "text_embedding": {
                    "model_id": "<YOUR_EMBEDDING_MODEL>",
                    "model_text": question
                }
            },
            "k": k,
            "num_candidates": k * 10
        },
        source=["content", "source_title", "source_url"]
    )
    chunks = resp["hits"]["hits"]

    # 2. Build context from retrieved chunks
    context_parts = []
    sources = []
    for i, hit in enumerate(chunks):
        src = hit["_source"]
        context_parts.append(f"[{i+1}] {src['content']}")
        sources.append({"title": src.get("source_title", ""), "url": src.get("source_url", "")})

    context = "\n\n".join(context_parts)

    # 3. Generate answer — replace model with the developer's choice from Step 2
    completion = llm.chat.completions.create(
        model="<LLM_MODEL>",
        messages=[
            {"role": "system", "content": (
                "Answer the user's question using ONLY the provided context. "
                "Cite sources using [1], [2], etc. "
                "If the context doesn't contain enough information, say so."
            )},
            {"role": "user", "content": f"Context:\n{context}\n\nQuestion: {question}"}
        ],
        temperature=0.2
    )

    return {
        "answer": completion.choices[0].message.content,
        "sources": sources
    }

def ask_with_history(question: str, history: list[dict], k: int = 5) -> dict:
    # Reformulate question using chat history for better retrieval
    if history:
        reformulation = llm.chat.completions.create(
            model="<LLM_MODEL>",
            messages=[
                {"role": "system", "content": (
                    "Rewrite the user's question as a standalone search query, "
                    "incorporating context from the conversation history."
                )},
                *history,
                {"role": "user", "content": question}
            ],
            temperature=0
        )
        search_query = reformulation.choices[0].message.content
    else:
        search_query = question

    # Retrieve using reformulated query
    result = ask(search_query, k=k)

    # Generate with full conversation context
    messages = [
        {"role": "system", "content": (
            "Answer the user's question using the provided context. "
            "Cite sources using [1], [2], etc. "
            "Consider the conversation history for context."
        )},
        *history,
        {"role": "user", "content": f"Context:\n{result['context']}\n\nQuestion: {question}"}
    ]

    completion = llm.chat.completions.create(
        model="<LLM_MODEL>",
        messages=messages,
        temperature=0.2
    )

    return {
        "answer": completion.choices[0].message.content,
        "sources": result["sources"]
    }

from flask import Flask, request, jsonify

app = Flask(__name__)

@app.route("/chat", methods=["POST"])
def chat():
    body = request.json
    question = body.get("question", "")
    history = body.get("history", [])

    if not question:
        return jsonify({"error": "Missing question"}), 400

    result = ask_with_history(question, history)
    return jsonify(result)