Building RAG Systems

RAG (Retrieval-Augmented Generation) is the pattern that makes AI actually useful for your data. It's how you build a chatbot that knows your docs.

---

Why RAG Works

LLMs are trained on public data. They don't know about:

• Your company's documentation
• Your product's features
• Your internal policies
• Anything after their training cutoff

---

Step 1: Document Chunking

You can't embed an entire document. You need to split it into chunks.

Why Chunk?

• Embedding limits: Most models max out at 8K tokens
• Precision: Smaller chunks = more precise retrieval
• Context window: LLMs have limited prompt space

Chunking Strategies

Simple Chunker Implementation

def chunk_text(text: str, chunk_size: int = 500, overlap: int = 50) -> list[str]:
    """Split text into overlapping chunks."""
    words = text.split()
    chunks = []
    start = 0

    while start < len(words):
        end = start + chunk_size
        chunk = ' '.join(words[start:end])
        chunks.append(chunk)
        start = end - overlap  # Overlap for context continuity

    return chunks

# Example
document = "Your very long document text here..."
chunks = chunk_text(document, chunk_size=300, overlap=30)
print(f"Created {len(chunks)} chunks")

---

Step 2: The RAG Pipeline

from openai import OpenAI
import psycopg2

client = OpenAI()
conn = psycopg2.connect("postgresql://localhost/mydb")

class RAGPipeline:
    def __init__(self):
        self._ensure_table()

    def _ensure_table(self):
        with conn.cursor() as cur:
            cur.execute("""
                CREATE TABLE IF NOT EXISTS chunks (
                    id SERIAL PRIMARY KEY,
                    content TEXT,
                    source TEXT,
                    embedding vector(1536)
                )
            """)
        conn.commit()

    def ingest_document(self, content: str, source: str):
        """Chunk and store a document."""
        chunks = self._chunk_text(content)
        for chunk in chunks:
            embedding = self._get_embedding(chunk)
            with conn.cursor() as cur:
                cur.execute(
                    "INSERT INTO chunks (content, source, embedding) VALUES (%s, %s, %s)",
                    (chunk, source, embedding)
                )
        conn.commit()
        print(f"Ingested {len(chunks)} chunks from {source}")

    def query(self, question: str, top_k: int = 3) -> str:
        """RAG query: retrieve context, then generate answer."""

        # 1. Retrieve relevant chunks
        context_chunks = self._retrieve(question, top_k)

        if not context_chunks:
            return "I don't have information about that."

        # 2. Build prompt with context
        context = "\n\n".join([c['content'] for c in context_chunks])
        sources = list(set([c['source'] for c in context_chunks]))

        # 3. Generate answer
        response = client.chat.completions.create(
            model="gpt-4o-mini",
            messages=[
                {
                    "role": "system",
                    "content": """Answer questions based on the provided context.
                    If the context doesn't contain the answer, say "I don't have information about that."
                    Be concise and cite which document the info came from."""
                },
                {
                    "role": "user",
                    "content": f"""Context:
{context}

Question: {question}"""
                }
            ]
        )

        answer = response.choices[0].message.content
        return f"{answer}\n\nSources: {', '.join(sources)}"

    def _retrieve(self, query: str, top_k: int) -> list[dict]:
        """Find most relevant chunks."""
        query_embedding = self._get_embedding(query)
        with conn.cursor() as cur:
            cur.execute("""
                SELECT content, source, 1 - (embedding <=> %s::vector) AS similarity
                FROM chunks
                WHERE 1 - (embedding <=> %s::vector) > 0.7
                ORDER BY embedding <=> %s::vector
                LIMIT %s
            """, (query_embedding, query_embedding, query_embedding, top_k))
            return [{"content": r[0], "source": r[1], "similarity": r[2]}
                    for r in cur.fetchall()]

    def _chunk_text(self, text: str, size: int = 300, overlap: int = 30) -> list[str]:
        words = text.split()
        chunks = []
        start = 0
        while start < len(words):
            chunk = ' '.join(words[start:start + size])
            chunks.append(chunk)
            start += size - overlap
        return chunks

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

---

Step 3: Using the Pipeline

# Initialize
rag = RAGPipeline()

# Ingest documents
rag.ingest_document("""
Our refund policy: We offer a 14-day money-back guarantee for all annual plans.
Monthly plans can be cancelled anytime but are not refundable.
To request a refund, email support@example.com with your order ID.
""", source="refund-policy.md")

rag.ingest_document("""
Pricing Plans:
- Starter: $9/month or $90/year (save $18)
- Pro: $29/month or $290/year (save $58)
- Enterprise: Custom pricing, contact sales
All plans include 14-day free trial.
""", source="pricing.md")

# Query
answer = rag.query("Can I get my money back if I don't like the product?")
print(answer)

# Output:
# Yes, you can get a refund within 14 days for annual plans.
# Monthly plans are not refundable but can be cancelled anytime.
# Email support@example.com with your order ID to request a refund.
#
# Sources: refund-policy.md

---

RAG Best Practices

---

Common RAG Pitfalls

---

Key Takeaways

Next up: Case Study - building a complete Document Q&A Bot with everything we've learned.