RAG Systems

The Core Problem RAG Solves LLMs have a knowledge cutoff and can't know about your private data. RAG gives them a "reference library" to look up at query time.

User Question
  ↓
1. EMBED: Convert question to vector
  ↓
2. RETRIEVE: Find top-K similar chunks from vector store
  ↓
3. AUGMENT: Build prompt: "Context: [chunks]\nQuestion: [query]"
  ↓
4. GENERATE: LLM answers using provided context
  ↓
Response (grounded in your documents)

Full Pipeline Code ```python from openai import OpenAI from pinecone import Pinecone

client = OpenAI() pc = Pinecone(api_key="...") index = pc.Index("my-docs")

def embed(text: str) -> list[float]: return client.embeddings.create( model="text-embedding-3-small", input=text ).data[0].embedding

def rag_query(question: str) -> str: # 1. Embed question q_vector = embed(question)

# 2. Retrieve top-5 relevant chunks results = index.query(vector=q_vector, top_k=5, include_metadata=True) context = "\n\n".join(m["metadata"]["text"] for m in results.matches)

# 3. Generate with context response = client.chat.completions.create( model="gpt-4o", messages=[ {"role": "system", "content": "Answer using ONLY the context provided. Say 'I don't know' if not in context."}, {"role": "user", "content": f"Context:\n{context}\n\nQuestion: {question}"} ] ) return response.choices[0].message.content ```

Chunking Strategies ```python from langchain.text_splitter import RecursiveCharacterTextSplitter

# Strategy 1: Fixed size with overlap splitter = RecursiveCharacterTextSplitter( chunk_size=1000, # chars per chunk chunk_overlap=200, # overlap prevents losing context at boundaries separators=["\n\n", "\n", ". ", " "] # try paragraph first, then sentence )

# Strategy 2: Semantic chunking (group by meaning) from langchain_experimental.text_splitter import SemanticChunker from langchain_openai import OpenAIEmbeddings

semantic_splitter = SemanticChunker( OpenAIEmbeddings(), breakpoint_threshold_type="percentile" ) ```

Chunking Best Practices | Content Type | Strategy | Chunk Size | |---|---|---| | Legal documents | Section headers | Variable | | Code files | Function/class level | Per function | | Academic papers | Abstract + sections | 500-1500 chars | | FAQ pages | Q&A pairs | One Q&A per chunk | | General prose | Recursive character | 500-1000 chars |

Metadata Enrichment ``python def process_document(doc_path: str) -> list[dict]: chunks = splitter.split_text(read_file(doc_path)) return [ { "text": chunk, "metadata": { "source": doc_path, "chunk_id": i, "total_chunks": len(chunks), "created_at": datetime.now().isoformat(), "doc_type": detect_doc_type(doc_path) } } for i, chunk in enumerate(chunks) ] ``

Hybrid Search (BM25 + Vector) ```python from langchain_community.retrievers import BM25Retriever from langchain.retrievers import EnsembleRetriever

# BM25: keyword matching (good for exact terms, product names, codes) bm25 = BM25Retriever.from_documents(documents) bm25.k = 5

# Vector: semantic similarity (good for paraphrases, concepts) from langchain_community.vectorstores import Chroma vectorstore = Chroma.from_documents(documents, embedding_fn) vector_retriever = vectorstore.as_retriever(search_kwargs={"k": 5})

# Ensemble: combine both (Reciprocal Rank Fusion) ensemble = EnsembleRetriever( retrievers=[bm25, vector_retriever], weights=[0.4, 0.6] # weight toward semantic ) ```

Reranking use a cross-encoder to rerank initial results ```python from sentence_transformers import CrossEncoder

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

def rerank(query: str, candidates: list[str], top_k: int = 3) -> list[str]: pairs = [(query, doc) for doc in candidates] scores = reranker.predict(pairs) ranked = sorted(zip(scores, candidates), reverse=True) return [doc for _, doc in ranked[:top_k]] ```

Query Expansion / HyDE (Hypothetical Document Embeddings) ```python def hyde_retrieve(question: str) -> list[str]: # Generate hypothetical answer → embed that → find similar real docs hypo = client.chat.completions.create( model="gpt-4o-mini", messages=[{"role": "user", "content": f"Write a short answer to: {question}"}] ).choices[0].message.content

return vector_search(hypo) # retrieve docs similar to hypothetical answer ```

RAGAS Evaluation Framework ```python from ragas import evaluate from ragas.metrics import ( faithfulness, # Does answer match context? (hallucination score) answer_relevancy, # Is answer relevant to question? context_precision, # Are retrieved docs useful? context_recall, # Are all relevant docs retrieved? )

# Build evaluation dataset test_cases = [ { "question": "What is our refund policy?", "ground_truth": "Refunds within 30 days...", "contexts": retrieved_docs, "answer": generated_answer } ]

results = evaluate(dataset=test_cases, metrics=[faithfulness, answer_relevancy]) print(results) # {'faithfulness': 0.87, 'answer_relevancy': 0.91} ```

Production Patterns ```python # Citation tracking: link answer back to source documents def answer_with_citations(question: str) -> dict: chunks = retrieve(question) answer = generate(question, chunks)

return { "answer": answer, "sources": [ {"title": c.metadata["source"], "excerpt": c.text[:200]} for c in chunks ] }

# Streaming responses stream = client.chat.completions.create(stream=True, ...) for chunk in stream: if chunk.choices[0].delta.content: yield chunk.choices[0].delta.content ```

Failure Modes & Fixes | Problem | Symptom | Fix | |---|---|---| | Hallucination | Answer not in context | Add "answer ONLY from context" instruction | | Poor retrieval | Wrong documents retrieved | Improve chunking, add metadata filters | | Context overflow | Answer cuts off | Reduce retrieved chunks, summarize | | Stale data | Outdated answers | Incremental indexing pipeline |