Build a Production RAG Pipeline in 30 Minutes with LangChain 0.5

# Verify installations
python --version  # 3.11+ required
pip --version

# Create isolated environment
python -m venv rag-env
source rag-env/bin/activate  # Windows: rag-env\Scripts\activate

# Install core packages
pip install langchain==0.5.1 \
            langchain-anthropic==0.5.0 \
            langchain-chroma==0.2.0 \
            anthropic==0.43.0 \
            --break-system-packages

mkdir rag-pipeline && cd rag-pipeline
touch main.py config.py documents.py

# Create sample documents
mkdir data
echo "Claude is an AI assistant created by Anthropic. It uses Constitutional AI for safety." > data/doc1.txt
echo "RAG combines retrieval with generation. It fetches relevant documents before answering." > data/doc2.txt

echo "ANTHROPIC_API_KEY=your_key_here" > .env

from dataclasses import dataclass
from pathlib import Path

@dataclass
class RAGConfig:
    # Model settings
    model_name: str = "claude-sonnet-4-5-20250929"
    max_tokens: int = 1024
    temperature: float = 0.0  # Deterministic for factual answers
    
    # Retrieval settings
    chunk_size: int = 1000  # Characters per chunk
    chunk_overlap: int = 200  # Overlap prevents context loss
    top_k: int = 3  # Number of chunks to retrieve
    
    # Paths
    data_dir: Path = Path("data")
    vector_db_dir: Path = Path("chroma_db")
    
    def __post_init__(self):
        self.data_dir.mkdir(exist_ok=True)
        self.vector_db_dir.mkdir(exist_ok=True)

config = RAGConfig()

from langchain_community.document_loaders import DirectoryLoader, TextLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
from config import config

def load_and_chunk_documents():
    """Load documents and split into semantic chunks."""
    
    # Load all .txt files
    loader = DirectoryLoader(
        config.data_dir,
        glob="**/*.txt",
        loader_cls=TextLoader,
        show_progress=True
    )
    documents = loader.load()
    
    # Split using recursive strategy
    # This tries to split on paragraphs, then sentences, then words
    text_splitter = RecursiveCharacterTextSplitter(
        chunk_size=config.chunk_size,
        chunk_overlap=config.chunk_overlap,
        separators=["\n\n", "\n", ".", " ", ""],  # Priority order
        length_function=len,
    )
    
    chunks = text_splitter.split_documents(documents)
    
    print(f"Loaded {len(documents)} documents")
    print(f"Split into {len(chunks)} chunks")
    
    return chunks

if __name__ == "__main__":
    # Test the loader
    chunks = load_and_chunk_documents()
    print(f"\nSample chunk:\n{chunks[0].page_content[:200]}...")

python documents.py

import os
from langchain_anthropic import ChatAnthropic
from langchain_chroma import Chroma
from langchain.embeddings import HuggingFaceEmbeddings
from langchain.chains import RetrievalQA
from langchain.prompts import PromptTemplate
from documents import load_and_chunk_documents
from config import config

# Load API key
from dotenv import load_dotenv
load_dotenv()

def create_vector_store():
    """Initialize vector database with embeddings."""
    
    # Load documents
    chunks = load_and_chunk_documents()
    
    # Use local embeddings (no API calls needed)
    # This model is optimized for semantic search
    embeddings = HuggingFaceEmbeddings(
        model_name="sentence-transformers/all-MiniLM-L6-v2",
        model_kwargs={'device': 'cpu'},
        encode_kwargs={'normalize_embeddings': True}
    )
    
    # Create persistent vector store
    vector_store = Chroma.from_documents(
        documents=chunks,
        embedding=embeddings,
        persist_directory=str(config.vector_db_dir),
        collection_name="rag_docs"
    )
    
    print(f"Vector store created with {vector_store._collection.count()} chunks")
    return vector_store

def build_rag_chain(vector_store):
    """Construct the RAG pipeline with Claude."""
    
    # Initialize Claude
    llm = ChatAnthropic(
        model=config.model_name,
        max_tokens=config.max_tokens,
        temperature=config.temperature,
        anthropic_api_key=os.getenv("ANTHROPIC_API_KEY")
    )
    
    # Custom prompt for RAG
    prompt_template = """You are a helpful assistant answering questions based on provided context.

Context from documentation:
{context}

Question: {question}

Instructions:
- Answer using ONLY information from the context
- If the context doesn't contain the answer, say "I don't have information about that in the provided documents"
- Be concise and specific
- Cite relevant parts of the context when useful

Answer:"""

    PROMPT = PromptTemplate(
        template=prompt_template,
        input_variables=["context", "question"]
    )
    
    # Build retrieval chain
    qa_chain = RetrievalQA.from_chain_type(
        llm=llm,
        chain_type="stuff",  # Puts all retrieved docs in one prompt
        retriever=vector_store.as_retriever(
            search_kwargs={"k": config.top_k}
        ),
        chain_type_kwargs={"prompt": PROMPT},
        return_source_documents=True  # For debugging
    )
    
    return qa_chain

async def query_rag(chain, question: str):
    """Query the RAG system."""
    
    try:
        result = await chain.ainvoke({"query": question})
        
        return {
            "answer": result["result"],
            "sources": [doc.metadata for doc in result["source_documents"]]
        }
    
    except Exception as e:
        return {
            "answer": f"Error: {str(e)}",
            "sources": []
        }

# Example usage
if __name__ == "__main__":
    import asyncio
    
    # Initialize system
    vector_store = create_vector_store()
    chain = build_rag_chain(vector_store)
    
    # Test queries
    async def main():
        questions = [
            "What is Claude?",
            "How does RAG work?",
            "What is the capital of France?"  # Not in docs
        ]
        
        for q in questions:
            print(f"\nQ: {q}")
            result = await query_rag(chain, q)
            print(f"A: {result['answer']}")
            print(f"Sources: {len(result['sources'])} documents")
    
    asyncio.run(main())

# First run will download embedding model (~100MB)
python main.py

# Add caching to avoid re-embedding same queries
from functools import lru_cache

@lru_cache(maxsize=100)
def cached_retrieval(query: str, k: int = 3):
    """Cache retrieval results for identical queries."""
    return vector_store.similarity_search(query, k=k)

# Parallel retrieval and LLM call
import asyncio

async def optimized_query(question: str):
    # Retrieve while Claude warms up
    retrieval_task = asyncio.create_task(
        vector_store.asimilarity_search(question, k=config.top_k)
    )
    
    docs = await retrieval_task
    context = "\n\n".join([d.page_content for d in docs])
    
    # Now call Claude with retrieved context
    response = await llm.ainvoke(f"Context: {context}\n\nQuestion: {question}")
    return response

from anthropic import APIError, RateLimitError
import time

async def resilient_query(chain, question: str, max_retries: int = 3):
    """Query with exponential backoff retry."""
    
    for attempt in range(max_retries):
        try:
            return await query_rag(chain, question)
        
        except RateLimitError:
            wait_time = 2 ** attempt  # Exponential backoff
            print(f"Rate limited. Waiting {wait_time}s...")
            await asyncio.sleep(wait_time)
        
        except APIError as e:
            if attempt == max_retries - 1:
                return {"answer": "Service temporarily unavailable", "sources": []}
            await asyncio.sleep(1)
    
    return {"answer": "Max retries exceeded", "sources": []}

def test_retrieval():
    """Verify chunks are being retrieved correctly."""
    
    test_cases = [
        ("Claude", "Anthropic"),  # Should find Claude doc
        ("retrieval", "RAG"),      # Should find RAG doc
        ("unrelated", None)        # Should still return something
    ]
    
    for query, expected_term in test_cases:
        docs = vector_store.similarity_search(query, k=1)
        content = docs[0].page_content
        
        if expected_term:
            assert expected_term in content, f"Expected '{expected_term}' in results for '{query}'"
            print(f"✓ '{query}' correctly retrieved content about '{expected_term}'")
        else:
            print(f"✓ '{query}' returned fallback results")

test_retrieval()

import anthropic

# Track costs
client = anthropic.Anthropic(api_key=os.getenv("ANTHROPIC_API_KEY"))

def estimate_cost(question: str, context: str):
    """Estimate Claude API cost for a query."""
    
    # Rough estimate: 1 token ≈ 4 characters
    input_tokens = len(question + context) // 4
    output_tokens = config.max_tokens
    
    # Claude Sonnet 4.5 pricing (as of Feb 2026)
    input_cost = input_tokens * 0.000003  # $3 per 1M input tokens
    output_cost = output_tokens * 0.000015  # $15 per 1M output tokens
    
    total = input_cost + output_cost
    print(f"Estimated cost: ${total:.6f} ({input_tokens} in, {output_tokens} out)")
    
    return total

# Try smaller chunks for more precise retrieval
config.chunk_size = 500
config.chunk_overlap = 100

# Or increase top_k to see more candidates
config.top_k = 5

prompt_template = """CRITICAL: You MUST answer using ONLY the context below. Do not use outside knowledge.

Context:
{context}

Question: {question}

If the answer is not in the context, respond with: "This information is not in the provided documents."

Answer based on context:"""

rm -rf chroma_db/
python main.py  # Rebuilds from scratch

pip install pypdf --break-system-packages

from langchain_community.document_loaders import PyPDFLoader

def load_pdfs(pdf_dir: Path):
    pdf_files = list(pdf_dir.glob("*.pdf"))
    all_chunks = []
    
    for pdf_path in pdf_files:
        loader = PyPDFLoader(str(pdf_path))
        pages = loader.load()
        chunks = text_splitter.split_documents(pages)
        all_chunks.extend(chunks)
    
    return all_chunks

async def stream_rag_response(chain, question: str):
    """Stream Claude's response token by token."""
    
    async for chunk in chain.astream({"query": question}):
        if "answer" in chunk:
            print(chunk["answer"], end="", flush=True)

def create_multi_tenant_store(tenant_id: str):
    """Separate vector stores per user/team."""
    
    return Chroma(
        collection_name=f"tenant_{tenant_id}",
        persist_directory=str(config.vector_db_dir / tenant_id),
        embedding_function=embeddings
    )