Production Guide for RAG-Based FAQ Chatbots: From Vector DB Selection to Operational Optimization

FAQ Document Collection -> Preprocessing/Normalization -> Chunking -> Embedding Generation -> Vector DB Storage -> Metadata Indexing

User Question -> Query Preprocessing -> Embedding Conversion -> Vector Search + BM25 -> Reranking -> Prompt Construction -> LLM Answer Generation -> Post-processing/Guardrails

"""
FAQ-specific chunking strategy.
Maintains question-answer pairs as a single unit while using
Parent-Child structure for both search precision and context.
"""
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_core.documents import Document
from typing import List, Tuple
import re
import hashlib


def parse_faq_pairs(raw_text: str) -> List[Tuple[str, str, dict]]:
    """Extract question-answer pairs from raw FAQ text."""
    faq_pattern = re.compile(
        r"(?:Q|질문)\s*[.:]\s*(.+?)\n+"
        r"(?:A|답변)\s*[.:]\s*(.+?)(?=\n(?:Q|질문)\s*[.:]|\Z)",
        re.DOTALL
    )
    pairs = []
    for i, match in enumerate(faq_pattern.finditer(raw_text)):
        question = match.group(1).strip()
        answer = match.group(2).strip()
        metadata = {
            "faq_id": hashlib.md5(question.encode()).hexdigest()[:8],
            "source_type": "faq",
            "question": question,
            "pair_index": i,
        }
        pairs.append((question, answer, metadata))
    return pairs


def create_faq_chunks(
    faq_pairs: List[Tuple[str, str, dict]],
    child_chunk_size: int = 200,
    child_chunk_overlap: int = 50,
) -> Tuple[List[Document], List[Document]]:
    """
    Split FAQ documents using Parent-Child chunking strategy.
    - Parent: Question + full answer (for LLM context)
    - Child: Answer split into small chunks (for search precision)
    """
    parent_docs = []
    child_docs = []

    child_splitter = RecursiveCharacterTextSplitter(
        chunk_size=child_chunk_size,
        chunk_overlap=child_chunk_overlap,
        separators=["\n\n", "\n", ". ", " "],
    )

    for question, answer, metadata in faq_pairs:
        # Parent document: Question + full answer
        parent_content = f"Question: {question}\nAnswer: {answer}"
        parent_id = metadata["faq_id"]
        parent_doc = Document(
            page_content=parent_content,
            metadata={**metadata, "doc_type": "parent", "parent_id": parent_id},
        )
        parent_docs.append(parent_doc)

        # Child documents: subdivide answer for improved search precision
        answer_chunks = child_splitter.split_text(answer)
        for j, chunk in enumerate(answer_chunks):
            child_content = f"Question: {question}\nAnswer excerpt: {chunk}"
            child_doc = Document(
                page_content=child_content,
                metadata={
                    **metadata,
                    "doc_type": "child",
                    "parent_id": parent_id,
                    "chunk_index": j,
                },
            )
            child_docs.append(child_doc)

    return parent_docs, child_docs


# Usage example
raw_faq = """
Q: How long does the refund process take?
A: Refunds are processed within 3-5 business days from the date of request.
For credit card payments, an additional 2-3 days for card company processing may apply.
For bank transfers, refunds are directly deposited to the registered account.

Q: Is international shipping available?
A: International shipping is currently available to the US, Japan, China, and Southeast Asia.
International shipping costs vary by region and weight, and customs duties are the recipient's responsibility.
Delivery takes 7-14 business days depending on the region.
"""

pairs = parse_faq_pairs(raw_faq)
parents, children = create_faq_chunks(pairs)
print(f"Parent documents: {len(parents)}, Child documents: {len(children)}")

"""
Pinecone vector DB setup and FAQ document indexing.
Separates document types by namespace and leverages metadata filtering.
"""
from pinecone import Pinecone, ServerlessSpec
from langchain_openai import OpenAIEmbeddings
from langchain_pinecone import PineconeVectorStore
from langchain_core.documents import Document
from typing import List
import os
import time


def setup_pinecone_index(
    index_name: str = "faq-chatbot",
    dimension: int = 1536,
    metric: str = "cosine",
) -> None:
    """Create a Pinecone index. Skip if it already exists."""
    pc = Pinecone(api_key=os.environ["PINECONE_API_KEY"])

    existing_indexes = [idx.name for idx in pc.list_indexes()]
    if index_name not in existing_indexes:
        pc.create_index(
            name=index_name,
            dimension=dimension,
            metric=metric,
            spec=ServerlessSpec(cloud="aws", region="us-east-1"),
        )
        # Wait until index is ready
        while not pc.describe_index(index_name).status["ready"]:
            time.sleep(1)
        print(f"Index '{index_name}' created successfully")
    else:
        print(f"Index '{index_name}' already exists")


def index_faq_documents(
    parent_docs: List[Document],
    child_docs: List[Document],
    index_name: str = "faq-chatbot",
) -> PineconeVectorStore:
    """
    Index Parent-Child structured FAQ documents in Pinecone.
    - Child documents: 'search' namespace (for retrieval)
    - Parent documents: 'context' namespace (for LLM context)
    """
    embeddings = OpenAIEmbeddings(model="text-embedding-3-small")

    # Index child documents (search targets)
    child_store = PineconeVectorStore.from_documents(
        documents=child_docs,
        embedding=embeddings,
        index_name=index_name,
        namespace="search",
    )
    print(f"Indexed {len(child_docs)} child documents (namespace: search)")

    # Index parent documents (for context provision)
    parent_store = PineconeVectorStore.from_documents(
        documents=parent_docs,
        embedding=embeddings,
        index_name=index_name,
        namespace="context",
    )
    print(f"Indexed {len(parent_docs)} parent documents (namespace: context)")

    return child_store


# Execute
setup_pinecone_index()
child_vectorstore = index_faq_documents(parents, children)

"""
LangChain-based FAQ chatbot implementation.
Integrates Parent-Child retrieval + custom prompts + conversation history management.
"""
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_pinecone import PineconeVectorStore
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.runnables import RunnablePassthrough, RunnableLambda
from langchain_core.output_parsers import StrOutputParser
from langchain_core.documents import Document
from typing import List, Dict
import os


# 1. Component Initialization
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
llm = ChatOpenAI(model="gpt-4o", temperature=0.1)

child_store = PineconeVectorStore(
    index_name="faq-chatbot",
    embedding=embeddings,
    namespace="search",
)
parent_store = PineconeVectorStore(
    index_name="faq-chatbot",
    embedding=embeddings,
    namespace="context",
)


# 2. Parent-Child Retriever Implementation
def retrieve_with_parent_lookup(query: str, k: int = 3) -> List[Document]:
    """
    Search with Child chunks, then return matched Parent documents.
    This ensures search precision at Child level, context at Parent level.
    """
    # Step 1: Similarity search on Child chunks
    child_results = child_store.similarity_search(query, k=k * 2)

    # Step 2: Deduplicate to extract unique parent_ids
    seen_parent_ids = set()
    unique_parent_ids = []
    for doc in child_results:
        pid = doc.metadata.get("parent_id")
        if pid and pid not in seen_parent_ids:
            seen_parent_ids.add(pid)
            unique_parent_ids.append(pid)
        if len(unique_parent_ids) >= k:
            break

    # Step 3: Retrieve Parent documents
    parent_results = parent_store.similarity_search(
        query,
        k=k,
        filter={"parent_id": {"$in": unique_parent_ids}},
    )

    return parent_results


# 3. Prompt Design
FAQ_PROMPT = ChatPromptTemplate.from_messages([
    ("system", """You are an AI assistant specializing in customer FAQ responses.
You must strictly follow these rules:

1. Answer based only on the provided FAQ documents.
2. If no answer is found in the FAQ documents, say "I could not find an answer to that question. Please contact our customer service at 1234-5678."
3. Do not speculate or generate information not in the FAQ.
4. Include the source of the relevant FAQ document in your answer.
5. Be friendly and concise.

Reference FAQ documents:
{context}"""),
    MessagesPlaceholder(variable_name="chat_history"),
    ("human", "{question}"),
])


# 4. Chain Construction
def format_docs(docs: List[Document]) -> str:
    """Format retrieved documents for inclusion in the prompt."""
    formatted = []
    for i, doc in enumerate(docs, 1):
        source_info = doc.metadata.get("faq_id", "unknown")
        formatted.append(
            f"[FAQ-{source_info}]\n{doc.page_content}"
        )
    return "\n\n---\n\n".join(formatted)


faq_chain = (
    {
        "context": RunnableLambda(
            lambda x: format_docs(retrieve_with_parent_lookup(x["question"]))
        ),
        "question": RunnablePassthrough() | RunnableLambda(lambda x: x["question"]),
        "chat_history": RunnableLambda(lambda x: x.get("chat_history", [])),
    }
    | FAQ_PROMPT
    | llm
    | StrOutputParser()
)


# 5. Execute
response = faq_chain.invoke({
    "question": "How long does the refund process take?",
    "chat_history": [],
})
print(response)

"""
BM25 + Dense hybrid search implementation.
Combines two search results using Reciprocal Rank Fusion (RRF).
"""
from langchain_community.retrievers import BM25Retriever
from langchain_core.documents import Document
from typing import List, Dict, Tuple
import numpy as np


class HybridRetriever:
    """Hybrid retriever combining BM25 and vector search."""

    def __init__(
        self,
        vector_store,
        documents: List[Document],
        bm25_k: int = 10,
        vector_k: int = 10,
        rrf_k: int = 60,
        alpha: float = 0.5,
    ):
        self.vector_store = vector_store
        self.bm25_retriever = BM25Retriever.from_documents(
            documents, k=bm25_k
        )
        self.vector_k = vector_k
        self.rrf_k = rrf_k
        self.alpha = alpha  # 0=BM25 only, 1=Dense only

    def _reciprocal_rank_fusion(
        self,
        bm25_results: List[Document],
        vector_results: List[Document],
    ) -> List[Tuple[Document, float]]:
        """Combine two search results using the RRF algorithm."""
        doc_scores: Dict[str, Tuple[Document, float]] = {}

        # Assign RRF scores to BM25 results
        for rank, doc in enumerate(bm25_results):
            doc_key = doc.page_content[:100]
            score = (1 - self.alpha) / (self.rrf_k + rank + 1)
            if doc_key in doc_scores:
                doc_scores[doc_key] = (
                    doc,
                    doc_scores[doc_key][1] + score,
                )
            else:
                doc_scores[doc_key] = (doc, score)

        # Assign RRF scores to Dense results
        for rank, doc in enumerate(vector_results):
            doc_key = doc.page_content[:100]
            score = self.alpha / (self.rrf_k + rank + 1)
            if doc_key in doc_scores:
                doc_scores[doc_key] = (
                    doc,
                    doc_scores[doc_key][1] + score,
                )
            else:
                doc_scores[doc_key] = (doc, score)

        # Sort by RRF score in descending order
        sorted_results = sorted(
            doc_scores.values(), key=lambda x: x[1], reverse=True
        )
        return sorted_results

    def retrieve(self, query: str, top_k: int = 5) -> List[Document]:
        """Perform hybrid search."""
        # Parallel search (use asyncio in production)
        bm25_results = self.bm25_retriever.invoke(query)
        vector_results = self.vector_store.similarity_search(
            query, k=self.vector_k
        )

        # Combine with RRF
        fused = self._reciprocal_rank_fusion(bm25_results, vector_results)

        return [doc for doc, score in fused[:top_k]]


# Usage example
hybrid_retriever = HybridRetriever(
    vector_store=child_store,
    documents=children,  # Original documents for BM25
    alpha=0.6,  # 60% Dense weight
)
results = hybrid_retriever.retrieve("How to resolve error code P4021")

                    +------------------+
                    |   Load Balancer  |
                    +--------+---------+
                             |
              +--------------+--------------+
              |                             |
    +---------v---------+    +---------v---------+
    |  API Server (1)   |    |  API Server (2)   |
    |  FastAPI + Uvicorn|    |  FastAPI + Uvicorn|
    +---------+---------+    +---------+---------+
              |                        |
    +---------v------------------------v---------+
    |              Redis Cache                    |
    |   (Query embedding cache, answer cache)     |
    +-----+-------------+-------------+----------+
          |             |             |
+---------v---+ +-------v-----+ +----v----------+
| Pinecone    | | BM25 Index  | | LLM API       |
| (Dense)     | | (Sparse)    | | (OpenAI/Azure)|
+-------------+ +-------------+ +---------------+

"""
RAGAS-based FAQ chatbot quality evaluation.
Automatically runs evaluation on a Golden Dataset and produces metrics.
"""
from ragas import evaluate
from ragas.metrics import (
    faithfulness,
    answer_relevancy,
    context_precision,
    context_recall,
)
from ragas.llms import LangchainLLMWrapper
from ragas.embeddings import LangchainEmbeddingsWrapper
from datasets import Dataset
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from typing import List, Dict
import json
from datetime import datetime


def prepare_evaluation_dataset(
    test_cases: List[Dict],
    retriever,
    chain,
) -> Dataset:
    """
    Convert test cases to RAGAS evaluation format.
    Performs actual retrieval and answer generation for each question.
    """
    eval_data = {
        "question": [],
        "answer": [],
        "contexts": [],
        "ground_truth": [],
    }

    for case in test_cases:
        question = case["question"]

        # Perform actual retrieval
        retrieved_docs = retriever.retrieve(question, top_k=5)
        contexts = [doc.page_content for doc in retrieved_docs]

        # Generate actual answer
        answer = chain.invoke({
            "question": question,
            "chat_history": [],
        })

        eval_data["question"].append(question)
        eval_data["answer"].append(answer)
        eval_data["contexts"].append(contexts)
        eval_data["ground_truth"].append(case["expected_answer"])

    return Dataset.from_dict(eval_data)


def run_ragas_evaluation(dataset: Dataset) -> Dict:
    """Run RAGAS evaluation and return results."""
    eval_llm = LangchainLLMWrapper(ChatOpenAI(model="gpt-4o", temperature=0))
    eval_embeddings = LangchainEmbeddingsWrapper(
        OpenAIEmbeddings(model="text-embedding-3-small")
    )

    result = evaluate(
        dataset=dataset,
        metrics=[
            faithfulness,
            answer_relevancy,
            context_precision,
            context_recall,
        ],
        llm=eval_llm,
        embeddings=eval_embeddings,
    )

    # Save results
    report = {
        "timestamp": datetime.now().isoformat(),
        "dataset_size": len(dataset),
        "metrics": {
            "faithfulness": float(result["faithfulness"]),
            "answer_relevancy": float(result["answer_relevancy"]),
            "context_precision": float(result["context_precision"]),
            "context_recall": float(result["context_recall"]),
        },
    }

    # Deployment gate: all metrics must exceed thresholds
    thresholds = {
        "faithfulness": 0.9,
        "answer_relevancy": 0.85,
        "context_precision": 0.8,
        "context_recall": 0.9,
    }
    report["deployment_gate"] = all(
        report["metrics"][k] >= v for k, v in thresholds.items()
    )

    with open(f"eval_report_{datetime.now().strftime('%Y%m%d_%H%M%S')}.json", "w") as f:
        json.dump(report, f, indent=2, ensure_ascii=False)

    return report


# Execution example
test_cases = [
    {
        "question": "How long does the refund process take?",
        "expected_answer": "Refunds are processed within 3-5 business days from the date of request.",
    },
    {
        "question": "How much is international shipping?",
        "expected_answer": "International shipping costs vary by region and weight, and customs duties are the recipient's responsibility.",
    },
]

# eval_dataset = prepare_evaluation_dataset(test_cases, hybrid_retriever, faq_chain)
# report = run_ragas_evaluation(eval_dataset)
# print(f"Deployment gate passed: {report['deployment_gate']}")

"""
FAQ chatbot operational monitoring.
Handles per-request metric collection, anomaly detection, and alert delivery.
"""
import time
import logging
from dataclasses import dataclass, field
from typing import Optional
from datetime import datetime
from prometheus_client import (
    Counter,
    Histogram,
    Gauge,
    start_http_server,
)

logger = logging.getLogger(__name__)

# Prometheus Metric Definitions
REQUEST_COUNT = Counter(
    "faq_chatbot_requests_total",
    "Total FAQ chatbot requests",
    ["status", "category"],
)
RESPONSE_LATENCY = Histogram(
    "faq_chatbot_response_seconds",
    "Response latency in seconds",
    buckets=[0.5, 1.0, 2.0, 3.0, 5.0, 10.0],
)
RETRIEVAL_LATENCY = Histogram(
    "faq_chatbot_retrieval_seconds",
    "Retrieval latency in seconds",
    buckets=[0.05, 0.1, 0.2, 0.5, 1.0],
)
LLM_TOKENS_USED = Counter(
    "faq_chatbot_llm_tokens_total",
    "Total LLM tokens consumed",
    ["type"],  # prompt, completion
)
FALLBACK_RATE = Gauge(
    "faq_chatbot_fallback_rate",
    "Current fallback (no answer) rate",
)
ACTIVE_REQUESTS = Gauge(
    "faq_chatbot_active_requests",
    "Currently processing requests",
)


@dataclass
class RequestMetrics:
    """Context manager that collects metrics for a single request."""
    question: str
    start_time: float = field(default_factory=time.time)
    retrieval_time: Optional[float] = None
    llm_time: Optional[float] = None
    total_time: Optional[float] = None
    status: str = "success"
    is_fallback: bool = False
    prompt_tokens: int = 0
    completion_tokens: int = 0

    def record_retrieval(self):
        self.retrieval_time = time.time() - self.start_time

    def record_llm_start(self):
        self._llm_start = time.time()

    def record_llm_end(self, prompt_tokens: int, completion_tokens: int):
        self.llm_time = time.time() - self._llm_start
        self.prompt_tokens = prompt_tokens
        self.completion_tokens = completion_tokens

    def finalize(self):
        self.total_time = time.time() - self.start_time

        # Record Prometheus metrics
        REQUEST_COUNT.labels(
            status=self.status, category="faq"
        ).inc()
        RESPONSE_LATENCY.observe(self.total_time)

        if self.retrieval_time:
            RETRIEVAL_LATENCY.observe(self.retrieval_time)

        LLM_TOKENS_USED.labels(type="prompt").inc(self.prompt_tokens)
        LLM_TOKENS_USED.labels(type="completion").inc(
            self.completion_tokens
        )

        # Structured logging
        logger.info(
            "faq_request_completed",
            extra={
                "question_preview": self.question[:50],
                "total_time_ms": round(self.total_time * 1000),
                "retrieval_time_ms": round(
                    (self.retrieval_time or 0) * 1000
                ),
                "status": self.status,
                "is_fallback": self.is_fallback,
                "tokens": self.prompt_tokens + self.completion_tokens,
            },
        )


# Start Prometheus metrics server
# start_http_server(8001)  # Expose /metrics endpoint