Architecting Enterprise RAG: Semantic Search Across 10,000+ Internal Documents
Architectural breakdown of an enterprise retrieval-augmented generation pipeline built at Aretove Technologies, achieving 90%+ semantic accuracy across 10,000+ proprietary regulatory documents using LangChain, ChromaDB, and FastAPI.
The Breakdown of Naive Vector Search in Regulated Enterprises
Most Retrieval-Augmented Generation (RAG) tutorials follow a dangerously naive recipe: parse PDFs with a generic text extractor, split text into arbitrary 500-token chunks, compute dense embeddings, and query OpenAI. In a typical sandbox demo, this looks impressive. In a regulated enterprise environment handling clinical trials, pharmaceutical standard operating procedures (SOPs), and strict compliance documents, this approach fails catastrophically.
At Aretove Technologies, our clinical research teams and validation engineers were spending up to 40% of their onboarding time wading through SharePoint folders housing more than 10,000 legacy documents. Standard keyword search failed on synonyms and contextual questions (e.g., searching for 'adverse event escalation timeline' failed to surface documents titled 'Protocol Deviation Notification Procedures'). Conversely, naive RAG pipelines hallucinated answers by retrieving semantically similar paragraphs stripped of their governing clause hierarchies.
The Golden Rule of Enterprise RAG: Retrieval accuracy is not just cosine similarity. If the context passed into the language model lacks structural hierarchy and governing metadata, the generated answer is an ungrounded hallucination.
Document Parsing: Structure-Aware Hierarchical Chunking
The root cause of RAG hallucinations in dense documentation is context fracture. Standard recursive character splitters split text wherever a token count ceiling is hit, frequently severing a crucial sub-clause (e.g., 'Exception: Only applies to Phase II cohorts') from the parent heading that established the condition.
We engineered a custom document preprocessing pipeline. Documents were parsed into markdown trees preserving hierarchical headers (`# Title`, `## Section`, `### Subsection`) and tabular structures. When splitting text, chunks were created with sliding token windows (512 tokens with 64-token overlap) and enriched with a deterministic header breadcrumb prepended to the vector payload.
from typing import List
from langchain_core.documents import Document
from langchain_text_splitters import MarkdownHeaderTextSplitter, RecursiveCharacterTextSplitter
HEADERS_TO_SPLIT = [
("#", "header_1"),
("##", "header_2"),
("###", "header_3"),
]
def process_enterprise_document(raw_text: str, doc_metadata: dict) -> List[Document]:
# 1. Structural Markdown Splitting based on document headings
markdown_splitter = MarkdownHeaderTextSplitter(headers_to_split_on=HEADERS_TO_SPLIT)
header_splits = markdown_splitter.split_text(raw_text)
# 2. Token-bounded recursive splitting for chunks that exceed LLM context bounds
text_splitter = RecursiveCharacterTextSplitter(
chunk_size=512,
chunk_overlap=64,
separators=["\n\n", "\n", ". ", " "]
)
enriched_chunks = []
for split in header_splits:
sub_chunks = text_splitter.split_documents([split])
for chunk in sub_chunks:
# Construct human-readable breadcrumb for vector payload
breadcrumb = " > ".join(
filter(None, [
chunk.metadata.get("header_1"),
chunk.metadata.get("header_2"),
chunk.metadata.get("header_3")
])
)
# Inject context directly into content payload
chunk.page_content = f"Document: {doc_metadata['filename']}\nSection: {breadcrumb}\n\n{chunk.page_content}"
chunk.metadata.update(doc_metadata)
chunk.metadata["breadcrumb"] = breadcrumb
enriched_chunks.append(chunk)
return enriched_chunksHybrid Search: Merging BM25 Sparse Search with Dense Vectors
Pure dense vector search (bi-encoders) excels at capturing conceptual nuance, but exhibits a blind spot for exact alphanumeric identifiers: protocol codes (e.g., 'IRB-2024-884A'), serial numbers, chemical designations, and acronyms. An exact keyword search will nail 'IRB-2024-884A' in 1 millisecond, while a dense embedding model might retrieve unrelated protocols that share similar semantic prose.
We architected a hybrid retrieval pipeline using Reciprocal Rank Fusion (RRF). Queries are executed in parallel across two distinct indexes: an in-memory BM25 sparse index for exact term matches and a ChromaDB vector store running HNSW cosine similarity. The ranks from both pipelines are merged using a constant penalization formula (`1 / (k + rank)`), ensuring that documents performing well across both lexical and semantic criteria rise to the top.
from collections import defaultdict
from typing import List, Tuple
from langchain_chroma import Chroma
from rank_bm25 import BM25Okapi
class EnterpriseHybridRetriever:
def __init__(self, vectorstore: Chroma, documents: List[str], k_rrf: int = 60):
self.vectorstore = vectorstore
self.k_rrf = k_rrf
# Tokenize corpus for sparse lexical retrieval
self.tokenized_corpus = [doc.lower().split() for doc in documents]
self.bm25 = BM25Okapi(self.tokenized_corpus)
self.documents = documents
def retrieve(self, query: str, top_n: int = 5) -> List[Tuple[str, float]]:
# 1. Dense Vector Search (ChromaDB)
dense_results = self.vectorstore.similarity_search_with_relevance_scores(query, k=top_n * 2)
# 2. Sparse Lexical Search (BM25)
query_tokens = query.lower().split()
bm25_scores = self.bm25.get_scores(query_tokens)
bm25_top_indices = sorted(range(len(bm25_scores)), key=lambda i: bm25_scores[i], reverse=True)[:top_n * 2]
# 3. Reciprocal Rank Fusion (RRF)
rrf_scores = defaultdict(float)
doc_map = {}
for rank, (doc, _) in enumerate(dense_results):
rrf_scores[doc.page_content] += 1.0 / (self.k_rrf + rank + 1)
doc_map[doc.page_content] = doc
for rank, idx in enumerate(bm25_top_indices):
content = self.documents[idx]
rrf_scores[content] += 1.0 / (self.k_rrf + rank + 1)
# Sort combined results by aggregated RRF score
sorted_results = sorted(rrf_scores.items(), key=lambda item: item[1], reverse=True)[:top_n]
return sorted_resultsCross-Encoder Re-Ranking and Grounded Citation Guardrails
Feeding 15 retrieved chunks directly into an LLM context window causes severe cognitive dilution—a phenomenon known as 'Lost in the Middle'. Furthermore, each extraneous token increases API latency and financial cost.
We placed a lightweight cross-encoder model (`ms-marco-MiniLM-L-6-v2`) between the retriever and the generator. Unlike bi-encoders that compute vector similarity in isolation, a cross-encoder processes the query and candidate chunk simultaneously through self-attention layers, scoring the true contextual relevance. We filter the top 20 retrieved candidates down to the 4 most definitive paragraphs.
Finally, our prompt engineering enforces deterministic citation grounding. The temperature is locked at `0.0`. The system prompt instructs the model that any assertion unsupported by a specific bracketed chunk citation (`[DocID: Section]`) is considered an evaluation failure.
from fastapi import APIRouter, HTTPException
from fastapi.responses import StreamingResponse
from pydantic import BaseModel
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
router = APIRouter()
llm = ChatOpenAI(model="gpt-4o", temperature=0.0, streaming=True)
SYSTEM_PROMPT = """You are a Principal Regulatory Assistant for enterprise clinical operations.
Answer the user's question using ONLY the retrieved context fragments below.
For EVERY claim or statement made, append a bracketed citation indicating the source chunk (e.g., [Doc-402 > Sec 3.1]).
If the provided context does not contain sufficient facts to answer with 100% confidence, respond with:
'The indexed documentation does not contain sufficient regulatory evidence to answer this query.'"""
@router.post("/api/v1/search/stream")
async def stream_query(request: QueryRequest):
chunks = await hybrid_retriever_service.get_relevant_chunks(request.query)
if not chunks:
raise HTTPException(status_code=404, detail="No matching documentation found.")
formatted_context = "\n\n---\n\n".join([c.page_content for c in chunks])
prompt = ChatPromptTemplate.from_messages([
("system", SYSTEM_PROMPT),
("human", "Context:\n{context}\n\nQuestion: {question}")
])
async def event_generator():
# Stream tokens as Server-Sent Events (SSE)
async for chunk in (prompt | llm).astream({"context": formatted_context, "question": request.query}):
yield f"data: {chunk.content}\n\n"
yield "data: [DONE]\n\n"
return StreamingResponse(event_generator(), media_type="text/event-stream")Results and Operational Impact
On an evaluation test suite of 450 complex regulatory and clinical questions, the system achieved a 90%+ factual retrieval accuracy, reducing false-positive hallucination rates to less than 1.8%.
Most importantly, internal engineering and clinical assessment teams reduced onboarding research time by 40%. Rather than filing manual tickets or paging lead architects for internal SOP clarifications, team members received cited, traceable answers with verifiable links directly back to the source documents.
Want to explore our enterprise data and AI systems? Read the Enterprise RAG Assistant Case Study or view our AI & Automation practice offerings.
Written By
Nikhil Anand
Principal Software Engineer & Technical Architect. Building high-performance web applications, offline-first desktop software, and enterprise data architectures for ambitious companies across India and globally.
Need an architecture like this for your product?
I design and build resilient, high-speed software systems tailored to your business operations. Schedule an architecture review or discuss project specifications directly.