Scribd
Upload
12 views10 pages

RAG Developer Guide: Ingest to Answer

The document is a developer guide for implementing Retrieval-Augmented Generation (RAG), outlining the process from document ingestion to answer generation. It includes detailed instructions on project structure, environment variables, and code samples for each step, including ingestion, retrieval, reranking, and answer generation. Additionally, it provides a FastAPI endpoint for querying and a production checklist for deployment considerations.

Uploaded by

Muddana Chowdary
Copyright
© All Rights Reserved
We take content rights seriously. If you suspect this is your content, claim it here.
Available Formats
Download as DOCX, PDF, TXT or read online on Scribd
12 views10 pages

RAG Developer Guide: Ingest to Answer

The document is a developer guide for implementing Retrieval-Augmented Generation (RAG), outlining the process from document ingestion to answer generation. It includes detailed instructions on project structure, environment variables, and code samples for each step, including ingestion, retrieval, reranking, and answer generation. Additionally, it provides a FastAPI endpoint for querying and a production checklist for deployment considerations.

Uploaded by

Muddana Chowdary
Copyright
© All Rights Reserved
We take content rights seriously. If you suspect this is your content, claim it here.
Available Formats
Download as DOCX, PDF, TXT or read online on Scribd

Retrieval-Augmented Generation (RAG)

Developer Guide
Basic → Production → Advanced (with real code samples)

0. Quick Start
 Ingest documents: load → clean → chunk → embed → upsert.
 Query: embed question → retrieve (vector + BM25) → rerank → answer with citations.
 Add logs, caching, and evaluation before production rollout.

Project Structure (reference)


rag_app/
app/
[Link]
[Link]
ingest/
[Link]
[Link]
[Link]
[Link]
retrieval/
qdrant_vector.py
[Link]
[Link]
[Link]
generation/
[Link]
[Link]
eval/
[Link]
[Link]
[Link]

Environment Variables
Name Example Notes

OPENAI_API_KEY sk-*** Used for embeddings (and


optionally LLM).

QDRANT_URL [Link] Qdrant endpoint.

QDRANT_API_KEY Optional if Qdrant is


secured.
QDRANT_COLLECTION docs Collection name for your
chunks.

EMBED_MODEL text-embedding-3-large Embedding model name.

LLM_MODEL gpt-4.1-mini LLM model for answering.

1. Ingestion (Load → Chunk → Embed → Upsert)


Goal: convert documents into searchable chunks with vectors and metadata.

1.1 Load & Clean (PDF/text)


# app/ingest/[Link]
from __future__ import annotations
from dataclasses import dataclass
from pathlib import Path
from typing import List, Dict, Any

# Use a robust PDF loader in your stack (LangChain, pypdf, pdfplumber,


etc.)
from pypdf import PdfReader

@dataclass
class DocChunkSource:
doc_id: str
source_path: str

def load_pdf_text(path: str) -> List[Dict[str, Any]]:


"""Return per-page text with metadata."""
p = Path(path)
reader = PdfReader(p)
pages = []
for i, page in enumerate([Link]):
txt = page.extract_text() or ""
txt = " ".join([Link]()) # simple whitespace cleanup
[Link]({
"text": txt,
"metadata": {"doc_id": [Link], "source": str(p), "page": i +
1},
})
return pages

1.2 Chunking (token-aware)


# app/ingest/[Link]
from __future__ import annotations
from typing import List, Dict, Any
def simple_chunk(text: str, chunk_size: int = 1200, overlap: int = 200)
-> List[str]:
"""Character-based chunking (ok for MVP). Replace with token
chunking in production."""
if not text:
return []
chunks = []
start = 0
n = len(text)
while start < n:
end = min(start + chunk_size, n)
[Link](text[start:end])
start = max(end - overlap, end)
return chunks

def pages_to_chunks(pages: List[Dict[str, Any]]) -> List[Dict[str,


Any]]:
out = []
for pg in pages:
for idx, ch in enumerate(simple_chunk(pg["text"])):
[Link]({
"text": ch,
"metadata": {**pg["metadata"], "chunk": idx},
})
return out

1.3 Embeddings
# app/ingest/[Link]
from __future__ import annotations
from typing import List
import os
from openai import OpenAI

client = OpenAI(api_key=[Link]("OPENAI_API_KEY"))
EMBED_MODEL = [Link]("EMBED_MODEL", "text-embedding-3-large")

def embed_texts(texts: List[str]) -> List[List[float]]:


resp = [Link](model=EMBED_MODEL, input=texts)
# Order matches input
return [[Link] for d in [Link]]

1.4 Upsert to Qdrant


# app/ingest/[Link]
from __future__ import annotations
from typing import List, Dict, Any
import os
from qdrant_client import QdrantClient
from qdrant_client.[Link] import VectorParams, Distance,
PointStruct

QDRANT_URL = [Link]("QDRANT_URL", "[Link]


QDRANT_API_KEY = [Link]("QDRANT_API_KEY") or None
COLLECTION = [Link]("QDRANT_COLLECTION", "docs")

def get_client() -> QdrantClient:


return QdrantClient(url=QDRANT_URL, api_key=QDRANT_API_KEY)

def ensure_collection(client: QdrantClient, dim: int):


exists = False
try:
client.get_collection(COLLECTION)
exists = True
except Exception:
exists = False
if not exists:
client.create_collection(
collection_name=COLLECTION,
vectors_config=VectorParams(size=dim,
distance=[Link]),
)

def upsert_chunks(client: QdrantClient, vectors: List[List[float]],


chunks: List[Dict[str, Any]]):
points = []
for i, (vec, ch) in enumerate(zip(vectors, chunks)):
payload = {"text": ch["text"], **ch["metadata"]}
[Link](PointStruct(id=i, vector=vec, payload=payload))
[Link](collection_name=COLLECTION, points=points)

1.5 End-to-end Ingest Script


# app/ingest/run_ingest.py
from __future__ import annotations
import sys
from [Link] import load_pdf_text
from [Link] import pages_to_chunks
from [Link] import embed_texts
from [Link] import get_client, ensure_collection,
upsert_chunks

def main(pdf_path: str):


pages = load_pdf_text(pdf_path)
chunks = pages_to_chunks(pages)
texts = [c["text"] for c in chunks]
vectors = embed_texts(texts)

client = get_client()
ensure_collection(client, dim=len(vectors[0]))
upsert_chunks(client, vectors, chunks)
print(f"Upserted {len(chunks)} chunks from {pdf_path}")

if __name__ == "__main__":
main([Link][1])
2. Retrieval (Vector + BM25 Hybrid)
Goal: get the best context for a question, not the most chunks.

2.1 Vector Search (Qdrant)


# app/retrieval/qdrant_vector.py
from __future__ import annotations
from typing import List, Dict, Any
import os
from qdrant_client import QdrantClient

QDRANT_URL = [Link]("QDRANT_URL", "[Link]


QDRANT_API_KEY = [Link]("QDRANT_API_KEY") or None
COLLECTION = [Link]("QDRANT_COLLECTION", "docs")

def search_vector(query_vec: List[float], top_k: int = 20, filters:


Dict[str, Any] | None = None):
client = QdrantClient(url=QDRANT_URL, api_key=QDRANT_API_KEY)

qfilter = None
# Add Qdrant filters here if you want (doc_id, domain, etc.)
hits = [Link](
collection_name=COLLECTION,
query_vector=query_vec,
limit=top_k,
query_filter=qfilter,
with_payload=True,
)
out = []
for h in hits:
payload = [Link] or {}
[Link]({
"score": float([Link]),
"text": [Link]("text", ""),
"meta": {k: v for k, v in [Link]() if k != "text"},
})
return out

2.2 BM25 Search (lexical)


# app/retrieval/[Link]
from __future__ import annotations
from typing import List, Dict, Any
import re
from rank_bm25 import BM25Okapi

def tokenize(s: str) -> List[str]:


return [Link](r"[a-zA-Z0-9]+", (s or "").lower())

class BM25Index:
def __init__(self, docs: List[Dict[str, Any]]):
[Link] = docs
[Link] = [tokenize(d["text"]) for d in docs]
self.bm25 = BM25Okapi([Link])

def search(self, query: str, top_k: int = 20) -> List[Dict[str,


Any]]:
q = tokenize(query)
scores = self.bm25.get_scores(q)
ranked = sorted(range(len(scores)), key=lambda i: scores[i],
reverse=True)[:top_k]
out = []
for i in ranked:
d = [Link][i]
[Link]({"score": float(scores[i]), "text": d["text"],
"meta": [Link]("meta", {})})
return out

2.3 Hybrid Merge (Rank Fusion)


# app/retrieval/[Link]
from __future__ import annotations
from typing import List, Dict, Any, Tuple
import hashlib

def _key(item: Dict[str, Any]) -> str:


# stable key to dedupe similar chunks
txt = ([Link]("text") or "").strip()
meta = [Link]("meta") or {}
raw = f"{[Link]('doc_id','')}|{[Link]('page','')}|{txt[:200]}"
return hashlib.md5([Link]("utf-8")).hexdigest()

def rrf_fusion(vector_hits: List[Dict[str, Any]], bm25_hits:


List[Dict[str, Any]], k: int = 60, top_n: int = 8):
"""Reciprocal Rank Fusion. Strong baseline."""
score_map: Dict[str, float] = {}
item_map: Dict[str, Dict[str, Any]] = {}

for rank, it in enumerate(vector_hits):


key = _key(it)
score_map[key] = score_map.get(key, 0.0) + 1.0 / (k + rank + 1)
item_map[key] = it

for rank, it in enumerate(bm25_hits):


key = _key(it)
score_map[key] = score_map.get(key, 0.0) + 1.0 / (k + rank + 1)
item_map[key] = it

merged = [{**item_map[key], "rrf": score} for key, score in


score_map.items()]
[Link](key=lambda x: x["rrf"], reverse=True)
return merged[:top_n]
3. Reranking (optional but powerful)
Rerank top 30–50 candidates down to top 5–10. This improves grounding.

# app/retrieval/[Link]
from __future__ import annotations
from typing import List, Dict, Any

# Option A: Use a cross-encoder (best quality, requires model hosting)


# from sentence_transformers import CrossEncoder
# reranker = CrossEncoder("cross-encoder/ms-marco-MiniLM-L-6-v2")

def simple_llm_free_rerank(query: str, items: List[Dict[str, Any]],


top_n: int = 6) -> List[Dict[str, Any]]:
"""Fallback: keep hybrid order. Replace with CrossEncoder in
production."""
return items[:top_n]

4. Answer Generation (strict + citations)


# app/generation/[Link]
from __future__ import annotations
from typing import List, Dict, Any

def build_prompt(question: str, ctx: List[Dict[str, Any]]) -> str:


blocks = []
for i, c in enumerate(ctx, start=1):
m = [Link]("meta", {})
cite = f"doc={[Link]('doc_id','')} page={[Link]('page','')}
chunk={[Link]('chunk','')}"
[Link](f"[C{i}] ({cite})\n{[Link]('text','')}")
context = "\n\n".join(blocks)

return f"""You are a helpful assistant.


Answer using ONLY the context below.
If the answer is missing, say you don't know.

QUESTION:
{question}

CONTEXT:
{context}

OUTPUT RULES:
- Give a short answer.
- Add citations like [C1], [C2] for every key claim.
"""

# app/generation/[Link]
from __future__ import annotations
import os
from openai import OpenAI
client = OpenAI(api_key=[Link]("OPENAI_API_KEY"))
LLM_MODEL = [Link]("LLM_MODEL", "gpt-4.1-mini")

def generate_answer(prompt: str) -> str:


resp = [Link](
model=LLM_MODEL,
input=prompt,
temperature=0.2,
)
return resp.output_text

5. FastAPI Endpoint (ready to run)


# app/[Link]
from __future__ import annotations
from fastapi import FastAPI
from pydantic import BaseModel

from [Link] import embed_texts


from [Link].qdrant_vector import search_vector
# BM25 index is usually loaded from disk or built from stored chunks.
from [Link] import rrf_fusion
from [Link] import simple_llm_free_rerank
from [Link] import build_prompt
from [Link] import generate_answer

app = FastAPI(title="RAG API")

class AskReq(BaseModel):
question: str
top_k: int = 20

@[Link]("/ask")
def ask(req: AskReq):
qvec = embed_texts([[Link]])[0]

vector_hits = search_vector(qvec, top_k=req.top_k)

# If you want BM25, feed it a corpus (e.g., from your DB dump)


bm25_hits = [] # plug in your [Link]([Link])

merged = rrf_fusion(vector_hits, bm25_hits, top_n=8)


final_ctx = simple_llm_free_rerank([Link], merged, top_n=6)

prompt = build_prompt([Link], final_ctx)


answer = generate_answer(prompt)

return {
"answer": answer,
"context": [
{"meta": [Link]("meta", {}), "preview": ([Link]("text", "")
[:240] + "...")}
for c in final_ctx
],
}

6. Docker Compose (Qdrant)


# [Link]
services:
qdrant:
image: qdrant/qdrant:latest
ports:
- "6333:6333"
volumes:
- qdrant_data:/qdrant/storage
volumes:
qdrant_data: {}

7. Evaluation (must-have for production)


# app/eval/[Link]
from __future__ import annotations
import json
from typing import List, Dict
from [Link] import embed_texts
from [Link].qdrant_vector import search_vector

# [Link] format:
# {"q":"...", "must_contain":"expected phrase", "doc_id":"optional"}

def run(testset_path: str, k: int = 10):


rows = [[Link](l) for l in open(testset_path, "r",
encoding="utf-8") if [Link]()]
ok = 0
for r in rows:
q = r["q"]
must = [Link]("must_contain", "").lower()
qvec = embed_texts([q])[0]
hits = search_vector(qvec, top_k=k)
joined = "\n".join(h["text"] for h in hits).lower()
hit_ok = (must in joined) if must else True
ok += 1 if hit_ok else 0
print("OK" if hit_ok else "MISS", "-", q)
print(f"Recall proxy@{k}: {ok}/{len(rows)}")
8. Production Checklist
 Add request tracing: query, retrieved chunk IDs, scores, latency, tokens.
 Add caching: embeddings + retrieval results (Redis).
 Add guardrails: max context size, strict grounding, safe fallback.
 Add reranker: cross-encoder or hosted rerank API.
 Build eval set: 50–200 real questions and track metrics weekly.

© 2026 — RC Reference

Common questions

Powered by AI

Building an evaluation set for RAG systems is important as it provides structured data to test the system’s ability to deliver accurate and contextually relevant answers. An evaluation set typically includes real user queries and expected outcomes, which helps in measuring key performance indicators like recall, precision, and the system's grounding ability. In the production environment, this evaluation plays a critical role by offering insights into system performance, guiding improvements, and ensuring the system remains aligned with user needs and expectations .

Integrating caching mechanisms in RAG systems is essential for optimizing performance and efficiency. Caching helps store frequently accessed data like embeddings and retrieval results, reducing the need for repeated computations and thereby speeding up response times. Benefits of caching include decreased latency, reduced computational load, enhanced scalability, and improved user experience by providing quicker answers. Additionally, caching can mitigate load on backend systems, leading to more robust and reliable RAG systems .

In a RAG pipeline, embeddings play a crucial role by transforming text data into high-dimensional vector representations that capture semantic meaning. These vectors are then stored in vector databases like Qdrant, which allows for efficient similarity searches when queries are made. The interaction occurs when a query is also embedded and used to retrieve similar vectors, i.e., document chunks, from Qdrant based on cosine similarity or other distance metrics. This capability enables the system to retrieve semantically relevant information, which can then be used to generate accurate and contextually-grounded answers .

Reciprocal Rank Fusion (RRF) is used in RAG systems to merge results from different retrieval mechanisms, like vector searches and BM25 lexical searches. RRF improves retrieval performance by assigning scores based on the rank of each item from different sources and combining these scores to determine an overall ranking. This method ensures diverse retrieval methods contribute to the final selection, providing a balance between lexical relevance and semantic similarity, which can significantly enhance the likelihood of retrieving the most contextually appropriate chunks .

Chunking contributes to the effectiveness of RAG systems by dividing large text bodies into smaller, manageable pieces, making data more accessible and improving retrieval accuracy. Considerations for implementing a chunking strategy include determining the optimal chunk size, balancing granularity and coherence, and ensuring overlap between chunks to maintain context. Proper chunking ensures that relevant information can be retrieved without losing context, which is crucial for contextually accurate responses in question-answering systems .

Reranking can significantly improve the quality of answers in a RAG system by refining the list of potential answer candidates down to the most relevant ones. Methods for reranking include using cross-encoders, which provide high-quality results by considering the interaction between query and context closely. Another method is using simpler models that preserve initial ranking order while applying minor adjustments. These methods help ensure that the retrieved contexts align better with the query's purpose, improving the final answer’s accuracy and relevance .

The main steps involved in the ingestion process for RAG applications are load, clean, chunk, embed, and upsert. Loading and cleaning involve extracting text data from documents and formatting it to ensure consistency and accuracy. Chunking breaks down documents into manageable pieces, making it easier to analyze and handle. Embedding converts text into vector representations, enabling effective retrieval mechanisms. Finally, upserting inserts these vector representations into a database like Qdrant for efficient searching and retrieval . Each step is critical as it transforms raw data into a usable format for a query-answering system, ensuring the system can efficiently find relevant documents for any given query .

Environment variables enhance the development and deployment of RAG applications by providing a flexible and secure way to manage configuration settings like API keys, URLs, and collection names. They allow developers to adapt the application for different environments (e.g., development, testing, production) without altering the codebase. By keeping sensitive information and configuration details external to the code, environment variables improve the security and maintainability of the application, making deployments and updates smoother .

Using both vector and BM25 search methods in a hybrid RAG system is significant because it combines the strengths of semantic retrieval and lexical retrieval. Vector searches, which involve embeddings, excel at capturing nuanced, semantic relationships between queries and document chunks. BM25 focuses on term frequency and inverse document frequency, offering precision based on exact language match. Together, they complement each other by broadening the scope of retrieval to include both term-based relevance and context-based relevance, improving overall search results accuracy and completeness .

To ensure strict grounding in RAG systems, strategies such as using stringent citation rules during answer generation can be employed, where answers must be directly linked to retrieved context. Safety measures may include implementing guardrails to bound context size and employing methods like fallback responses when grounding is insufficient. Additionally, regular evaluations with a curated set of real questions and monitoring metrics can help identify and rectify potential grounding issues, ultimately ensuring that the answers are both accurate and reliable .

You might also like