Build a RAG Application: Step-by-Step Tutorial

from pypdf import PdfReader
import os

def extract_text_from_pdf(pdf_path):
    """Extract text from PDF file"""
    reader = PdfReader(pdf_path)
    text = ""
    for page in reader.pages:
        text += page.extract_text()
    return text

# Extract from all PDFs
documents = []
for filename in os.listdir("./documents"):
    if filename.endswith(".pdf"):
        text = extract_text_from_pdf(f"./documents/{filename}")
        documents.append({
            "title": filename,
            "content": text
        })

print(f"Extracted {len(documents)} documents")

from langchain.text_splitter import RecursiveCharacterTextSplitter

splitter = RecursiveCharacterTextSplitter(
    chunk_size=1000,      # Characters per chunk
    chunk_overlap=200,    # Overlap for context
    separators=["\n\n", "\n", " ", ""]
)

chunks = []
for doc in documents:
    doc_chunks = splitter.split_text(doc["content"])
    for chunk in doc_chunks:
        chunks.append({
            "text": chunk,
            "source": doc["title"]
        })

print(f"Created {len(chunks)} chunks from {len(documents)} documents")

import openai

openai.api_key = "your-api-key"

def create_embeddings(texts):
    """Convert texts to embeddings using OpenAI"""
    response = openai.Embedding.create(
        input=texts,
        model="text-embedding-3-small"  # Cost-effective
    )

    embeddings = []
    for item in response["data"]:
        embeddings.append(item["embedding"])

    return embeddings

# Create embeddings for all chunks
chunk_texts = [chunk["text"] for chunk in chunks]
embeddings = create_embeddings(chunk_texts)

# Store chunks with embeddings
for chunk, embedding in zip(chunks, embeddings):
    chunk["embedding"] = embedding

pip install pinecone-client

from pinecone import Pinecone

# Initialize
pc = Pinecone(api_key="your-api-key")

# Create index (one-time)
pc.create_index(
    name="my-rag-index",
    dimension=1536,  # OpenAI embedding dimension
    metric="cosine"
)

index = pc.Index("my-rag-index")

# Upsert chunks with embeddings
vectors_to_upsert = []
for i, chunk in enumerate(chunks):
    vectors_to_upsert.append({
        "id": f"chunk-{i}",
        "values": chunk["embedding"],
        "metadata": {
            "text": chunk["text"],
            "source": chunk["source"]
        }
    })

# Upload in batches
batch_size = 100
for i in range(0, len(vectors_to_upsert), batch_size):
    batch = vectors_to_upsert[i:i+batch_size]
    index.upsert(vectors=batch)

print(f"Upserted {len(vectors_to_upsert)} vectors")

def retrieve_relevant_chunks(query, top_k=3):
    """Find chunks most relevant to query"""

    # Create embedding for query
    query_response = openai.Embedding.create(
        input=query,
        model="text-embedding-3-small"
    )
    query_embedding = query_response["data"][0]["embedding"]

    # Search vector DB
    results = index.query(
        vector=query_embedding,
        top_k=top_k,
        include_metadata=True
    )

    # Extract chunks
    retrieved_chunks = []
    for result in results["matches"]:
        retrieved_chunks.append({
            "text": result["metadata"]["text"],
            "source": result["metadata"]["source"],
            "score": result["score"]  # Relevance score 0-1
        })

    return retrieved_chunks

# Test retrieval
query = "What's your refund policy?"
results = retrieve_relevant_chunks(query)

for result in results:
    print(f"Source: {result['source']}")
    print(f"Relevance: {result['score']:.2f}")
    print(f"Text: {result['text'][:200]}...\n")

def build_prompt(query, retrieved_chunks):
    """Build prompt with retrieved context"""

    # System message
    system_message = """You are a helpful assistant. Answer questions based on the provided documents.
If you can't find the answer in the documents, say "I don't have information about that."
Always cite which document you're using."""

    # Context from retrieved chunks
    context = "RELEVANT DOCUMENTS:\n"
    for i, chunk in enumerate(retrieved_chunks):
        context += f"\n[{i+1}] From {chunk['source']}:\n{chunk['text']}\n"

    # Full prompt
    prompt = f"""{system_message}

{context}

QUESTION: {query}

ANSWER:"""

    return prompt, system_message

# Example
query = "What's your refund policy?"
chunks = retrieve_relevant_chunks(query)
prompt, system_msg = build_prompt(query, chunks)

print("Generated Prompt:")
print(prompt)

import openai

def answer_question(query):
    """Answer question using RAG"""

    # Step 1: Retrieve
    chunks = retrieve_relevant_chunks(query, top_k=3)

    # Step 2: Build prompt
    prompt, system_msg = build_prompt(query, chunks)

    # Step 3: Generate
    response = openai.ChatCompletion.create(
        model="gpt-3.5-turbo",
        messages=[
            {
                "role": "system",
                "content": system_msg
            },
            {
                "role": "user",
                "content": prompt
            }
        ],
        temperature=0.5,  # Lower = more focused on documents
        max_tokens=500
    )

    answer = response["choices"][0]["message"]["content"]

    return {
        "question": query,
        "answer": answer,
        "sources": [chunk["source"] for chunk in chunks]
    }

# Test
result = answer_question("What's your refund policy?")
print(f"Q: {result['question']}")
print(f"A: {result['answer']}")
print(f"Sources: {', '.join(result['sources'])}")

from transformers import pipeline

# Load locally
qa_pipeline = pipeline(
    "text2text-generation",
    model="google/flan-t5-large"
)

def answer_question_local(query):
    """Answer using local model"""
    chunks = retrieve_relevant_chunks(query)
    prompt, _ = build_prompt(query, chunks)

    result = qa_pipeline(prompt, max_length=500)
    return result[0]["generated_text"]

from flask import Flask, request, jsonify

app = Flask(__name__)

@app.route("/ask", methods=["POST"])
def ask():
    """API endpoint to ask questions"""
    data = request.json
    query = data.get("query")

    if not query:
        return jsonify({"error": "No query provided"}), 400

    result = answer_question(query)

    return jsonify(result)

@app.route("/health", methods=["GET"])
def health():
    return jsonify({"status": "ok"})

if __name__ == "__main__":
    app.run(debug=True, port=5000)

<!DOCTYPE html>
<html>
<head>
    <title>RAG Assistant</title>
    <style>
        body { font-family: Arial; max-width: 800px; margin: 50px auto; }
        input { width: 100%; padding: 10px; font-size: 16px; }
        button { padding: 10px 20px; font-size: 16px; cursor: pointer; }
        .response { margin-top: 20px; padding: 15px; background: #f0f0f0; }
        .sources { margin-top: 10px; color: #666; font-size: 14px; }
    </style>
</head>
<body>
    <h1>Ask Me Anything</h1>
    <input type="text" id="query" placeholder="Ask your question...">
    <button onclick="ask()">Ask</button>
    <div id="result"></div>

    <script>
        function ask() {
            const query = document.getElementById("query").value;

            fetch("/ask", {
                method: "POST",
                headers: {"Content-Type": "application/json"},
                body: JSON.stringify({query})
            })
            .then(r => r.json())
            .then(data => {
                document.getElementById("result").innerHTML = `
                    <div class="response">
                        <h3>Answer:</h3>
                        <p>${data.answer}</p>
                        <div class="sources">
                            Sources: ${data.sources.join(", ")}
                        </div>
                    </div>
                `;
            });
        }
    </script>
</body>
</html>

test_questions = [
    ("What's the refund policy?", "refund_policy.md"),
    ("How do I reset my password?", "help_articles.md"),
    ("What's your pricing?", "pricing_page.pdf")
]

results = []
for question, expected_source in test_questions:
    result = answer_question(question)

    # Check if correct source used
    correct = expected_source in result["sources"]

    results.append({
        "question": question,
        "correct_source": correct,
        "answer": result["answer"]
    })

# Summary
correct = sum(1 for r in results if r["correct_source"])
accuracy = correct / len(results) * 100
print(f"Source accuracy: {accuracy:.0f}%")

from functools import lru_cache

@lru_cache(maxsize=1000)
def answer_question_cached(query):
    return answer_question(query)

from flask_limiter import Limiter

limiter = Limiter(app)

@app.route("/ask", methods=["POST"])
@limiter.limit("100 per hour")
def ask():
    # ...