LLM RAG Production Pipeline: Building Production-Grade Retrieval-Augmented Generation Systems from Scratch

Summary

RAG (Retrieval-Augmented Generation) is the core architecture for LLM production deployment, solving hallucination, knowledge staleness, and domain gap
5 key stages in production RAG: document parsing → chunking → embedding → retrieval → reranking, each with optimization opportunities
Semantic chunking + sliding window improves retrieval recall by 20-30% over fixed-length chunking
Hybrid retrieval (vector + keyword + knowledge graph) improves recall by 15-25% over pure vector search
Reranking models (BGE-Reranker/Cohere Rerank) boost final answer accuracy from 75% to 90%+
This article provides a complete RAG solution from document processing to production deployment, with Python implementations and performance benchmarks

Why RAG Is the Core Architecture for LLM Production
Document Parsing and Smart Chunking
Embedding Pipeline and Vector Index Construction
Hybrid Retrieval and Reranking
RAG Full-Chain Optimization Strategies
Production Deployment and Observability
Summary and Further Reading

Why RAG Is the Core Architecture for LLM Production

LLMs have three inherent flaws: hallucination (generating non-existent facts), knowledge staleness (inability to access post-cutoff information), and domain gaps (lacking vertical domain expertise). RAG solves these by retrieving relevant context from external knowledge bases before generation.

┌──────────────────────────────────────────────────────────────────┐
│              RAG System Full-Chain Architecture                   │
│                                                                    │
│  ┌──────────┐    ┌──────────┐    ┌──────────┐    ┌──────────┐  │
│  │ 1. Parse  │──→│ 2. Chunk │──→│ 3.Embed  │──→│ 4. Vector │  │
│  │ PDF/DOCX │    │ Semantic │    │ Pipeline │    │ Milvus   │  │
│  │ HTML/MD  │    │ Sliding  │    │ Batch    │    │ HNSW     │  │
│  └──────────┘    └──────────┘    └──────────┘    └──────────┘  │
│                                                       │          │
│  ┌──────────┐    ┌──────────┐    ┌──────────┐         │          │
│  │ 8. Answer │←──│ 7.Prompt │←──│ 6.Rerank │←────────┤          │
│  │ LLM Gen  │    │ Engineer │    │ BGE-Rank │  5.Hybrid│         │
│  │ Citation │    │ Context  │    │ Top-K    │  Vec+BM25│         │
│  └──────────┘    └──────────┘    └──────────┘    └──────────┘  │
└──────────────────────────────────────────────────────────────────┘

RAG vs Pure LLM Key Metrics

Dimension	Pure LLM	RAG-Enhanced LLM
Factual Accuracy	60-70%	90-95%
Hallucination Rate	15-30%	3-5%
Domain Knowledge	General	Customizable
Knowledge Updates	Requires retraining	Incremental KB updates
Explainability	Low	High (citation tracing)
Cost	High (large model inference)	Medium (retrieval + small model)

Document Parsing and Smart Chunking

Chunking Strategy Comparison

Strategy	Recall	Context Completeness	Complexity
Fixed length (512 tokens)	65%	Low (truncates semantics)	Simple
Paragraph chunking	72%	Medium	Simple
Semantic chunking	85%	High	Medium
Semantic + sliding window	92%	High	Medium

Semantic Chunker Implementation

class SemanticChunker:
    def __init__(self, embedding_client, max_chunk_tokens: int = 512, overlap_tokens: int = 64, similarity_threshold: float = 0.75):
        self.embedding_client = embedding_client
        self.max_chunk_tokens = max_chunk_tokens
        self.overlap_tokens = overlap_tokens
        self.similarity_threshold = similarity_threshold

    async def chunk_document(self, doc: ParsedDocument) -> list[Chunk]:
        sentences = self._split_sentences(doc.content)
        embeddings = await self._batch_embed(sentences)
        chunks = []
        current_chunk = [sentences[0]]
        current_tokens = self._count_tokens(sentences[0])

        for i in range(1, len(sentences)):
            similarity = self._cosine_similarity(embeddings[i - 1], embeddings[i])
            if similarity < self.similarity_threshold or current_tokens + self._count_tokens(sentences[i]) > self.max_chunk_tokens:
                chunk_content = ' '.join(current_chunk)
                chunks.append(Chunk(chunk_id=f"{doc.doc_id}_c{len(chunks)}", content=chunk_content, metadata={**doc.metadata, "chunk_index": len(chunks)}, token_count=current_tokens))
                overlap_start = max(0, len(current_chunk) - self._sentences_for_tokens(self.overlap_tokens))
                current_chunk = current_chunk[overlap_start:] + [sentences[i]]
                current_tokens = sum(self._count_tokens(s) for s in current_chunk)
            else:
                current_chunk.append(sentences[i])
                current_tokens += self._count_tokens(sentences[i])

        if current_chunk:
            chunks.append(Chunk(chunk_id=f"{doc.doc_id}_c{len(chunks)}", content=' '.join(current_chunk), metadata={**doc.metadata, "chunk_index": len(chunks)}, token_count=current_tokens))
        return chunks

Embedding Pipeline and Vector Index Construction

Batch Embedding Pipeline

class EmbeddingPipeline:
    def __init__(self, model: str = "BAAI/bge-large-zh-v1.5", batch_size: int = 64):
        self.client = AsyncOpenAI(base_url="http://localhost:8000/v1")
        self.model = model
        self.batch_size = batch_size

    async def embed_chunks(self, chunks: list[Chunk]) -> list[dict]:
        all_embeddings = []
        for i in range(0, len(chunks), self.batch_size):
            batch = chunks[i:i + self.batch_size]
            texts = [f"search query: {c.content}" for c in batch]
            response = await self.client.embeddings.create(model=self.model, input=texts)
            for j, item in enumerate(response.data):
                all_embeddings.append({"chunk_id": batch[j].chunk_id, "content": batch[j].content, "metadata": batch[j].metadata, "embedding": item.embedding, "token_count": batch[j].token_count})
        return all_embeddings

Vector Index Construction

class RAGVectorStore:
    def __init__(self, milvus_uri: str = "http://localhost:19530", collection_name: str = "rag_chunks"):
        self.client = MilvusClient(uri=milvus_uri)
        self.collection_name = collection_name
        self._ensure_collection()

    def _ensure_collection(self):
        if self.client.has_collection(self.collection_name): return
        schema = self.client.create_schema(auto_id=True, enable_dynamic_field=True)
        schema.add_field(field_name="id", datatype=DataType.INT64, is_primary=True)
        schema.add_field(field_name="chunk_id", datatype=DataType.VARCHAR, max_length=256)
        schema.add_field(field_name="content", datatype=DataType.VARCHAR, max_length=65535)
        schema.add_field(field_name="embedding", datatype=DataType.FLOAT_VECTOR, dim=1024)
        schema.add_field(field_name="source", datatype=DataType.VARCHAR, max_length=512)
        index_params = self.client.prepare_index_params()
        index_params.add_index(field_name="embedding", index_type="HNSW", metric_type="COSINE", params={"M": 16, "efConstruction": 200})
        self.client.create_collection(collection_name=self.collection_name, schema=schema, index_params=index_params)

    def search(self, query_embedding: list[float], top_k: int = 10, ef: int = 100) -> list[dict]:
        results = self.client.search(self.collection_name, data=[query_embedding], limit=top_k, output_fields=["chunk_id", "content", "source"], search_params={"metric_type": "COSINE", "params": {"ef": ef}})
        return [{"chunk_id": r["entity"]["chunk_id"], "content": r["entity"]["content"], "source": r["entity"]["source"], "score": r["distance"]} for r in results[0]]

Hybrid Retrieval and Reranking

Hybrid Retrieval: Vector + BM25

class HybridRetriever:
    def __init__(self, vector_store: RAGVectorStore, bm25: BM25Retriever, vector_weight: float = 0.7, bm25_weight: float = 0.3):
        self.vector_store = vector_store
        self.bm25 = bm25
        self.vector_weight = vector_weight
        self.bm25_weight = bm25_weight

    async def search(self, query: str, query_embedding: list[float], top_k: int = 10) -> list[dict]:
        vector_results = self.vector_store.search(query_embedding, top_k=top_k * 2)
        bm25_results = self.bm25.search(query, top_k=top_k * 2)
        merged: dict[str, dict] = {}
        for r in vector_results: merged[r["chunk_id"]] = {**r, "vector_score": r["score"], "bm25_score": 0.0}
        for r in bm25_results:
            if r["chunk_id"] in merged: merged[r["chunk_id"]]["bm25_score"] = r["score"]
            else: merged[r["chunk_id"]] = {**r, "vector_score": 0.0, "bm25_score": r["score"]}
        max_vector = max((m["vector_score"] for m in merged.values()), default=1.0) or 1.0
        max_bm25 = max((m["bm25_score"] for m in merged.values()), default=1.0) or 1.0
        for m in merged.values(): m["combined_score"] = self.vector_weight * m["vector_score"] / max_vector + self.bm25_weight * m["bm25_score"] / max_bm25
        return sorted(merged.values(), key=lambda x: x["combined_score"], reverse=True)[:top_k]

Reranking

class Reranker:
    def __init__(self, llm_client, model: str = "BAAI/bge-reranker-v2-m3"):
        self.llm = llm_client
        self.model = model

    async def rerank(self, query: str, candidates: list[dict], top_k: int = 5) -> list[dict]:
        pairs = [[query, c["content"][:512]] for c in candidates]
        scores = await self._compute_scores(pairs)
        for i, candidate in enumerate(candidates): candidate["rerank_score"] = scores[i]
        candidates.sort(key=lambda x: x["rerank_score"], reverse=True)
        return candidates[:top_k]

RAG Full-Chain Optimization Strategies

RAG Full-Chain Performance Benchmarks

Stage	Latency(P50)	Latency(P99)	Notes
Document parsing (PDF 10 pages)	500ms	1.5s	PyMuPDF
Semantic chunking (10 pages)	2s	5s	Includes embedding calls
Batch embedding (64 chunks)	800ms	2s	BGE-large
Vector search (top-10)	5ms	15ms	Milvus HNSW
BM25 search (top-10)	2ms	5ms	In-memory index
Hybrid fusion	1ms	3ms	Score normalization
Reranking (top-5)	200ms	500ms	BGE-Reranker
LLM generation (7B)	1.5s	3s	Qwen2.5-7B
End-to-end RAG	3s	6s	Full chain

Retrieval Recall Comparison

Retrieval Method	Top-5 Recall	Top-10 Recall	Top-20 Recall
Pure vector search	72%	82%	88%
Pure BM25	65%	75%	80%
Hybrid retrieval	82%	90%	95%
Hybrid + reranking	88%	94%	97%

Production Deployment and Observability

RAG Service K8s Deployment

apiVersion: apps/v1
kind: Deployment
metadata:
  name: rag-service
  namespace: ai-rag
spec:
  replicas: 2
  selector:
    matchLabels:
      app: rag-service
  template:
    metadata:
      labels:
        app: rag-service
    spec:
      containers:
        - name: rag-api
          image: myregistry/rag-service:v1.0
          ports:
            - containerPort: 8000
          resources:
            requests:
              cpu: "2"
              memory: 4Gi
            limits:
              cpu: "4"
              memory: 8Gi
          env:
            - name: MILVUS_URI
              value: "http://milvus:19530"
            - name: LLM_API_BASE
              value: "http://vllm-qwen2-72b:8000/v1"

Summary and Further Reading

RAG is the core architecture for LLM production deployment. Each of the 5 key stages (parsing → chunking → embedding → retrieval → reranking) has optimization opportunities. Semantic chunking + sliding window improves recall 20-30% over fixed chunking, hybrid retrieval improves recall 15-25% over pure vector search, and reranking boosts accuracy from 75% to 90%+.

Key Development Takeaways:

Document parsing: PyMuPDF for PDF, section-based for Markdown, BeautifulSoup for HTML denoising
Smart chunking: Semantic + sliding window, max_chunk_tokens=512, overlap_tokens=64
Embedding: BGE-large-zh-v1.5, batch size 64, query prefix "search query:"
Hybrid retrieval: Vector weight 0.7 + BM25 weight 0.3, score normalization + weighted fusion
Reranking: BGE-Reranker-v2-m3, select top-5 from top-20 candidates

Related Reading:

Rust Vector Database Internals and Performance Optimization — Vector index optimization for RAG retrieval backends
K8s 1.30+ LLM Inference Autoscaling — K8s orchestration for RAG inference services
AI Agent Multi-Agent Orchestration — RAG integration in agent systems

Authoritative References: