RAG Pipeline Automation:
Building AI That Knows Your Business

Documents (PDFs, Word, web pages, Notion)
    ↓ Load & extract text
    ↓ Split into chunks (500-800 tokens, 100-token overlap)
    ↓ Embed each chunk (text-embedding-3-small)
    ↓ Store in vector database (FAISS / Pinecone / Supabase)

User question
    ↓ Embed question with same model
    ↓ Similarity search in vector database
    ↓ Retrieve top-3 to 5 most relevant chunks
    ↓ Prompt: [System] + [Retrieved chunks] + [Question]
    ↓ LLM generates grounded response

pip install openai faiss-cpu numpy

import openai, faiss, numpy as np, json
client = openai.OpenAI()

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

def chunk(text, size=600, overlap=100):
    words = text.split()
    return [" ".join(words[i:i+size]) for i in range(0, len(words), size-overlap) if words[i:i+size]]

class RAGSystem:
    def __init__(self):
        self.chunks = []
        self.index = None

    def build_index(self, docs):
        all_chunks = []
        for doc in docs:
            for c in chunk(doc["content"]):
                all_chunks.append({"text": c, "title": doc.get("title","")})
        vecs = np.array([embed(c["text"]) for c in all_chunks], dtype=np.float32)
        faiss.normalize_L2(vecs)
        self.index = faiss.IndexFlatIP(vecs.shape[1])
        self.index.add(vecs)
        self.chunks = all_chunks
        print(f"Indexed {len(all_chunks)} chunks")

    def search(self, query, k=3):
        q = np.array([embed(query)], dtype=np.float32)
        faiss.normalize_L2(q)
        _, idxs = self.index.search(q, k)
        return [self.chunks[i] for i in idxs[0] if i != -1]

    def answer(self, question):
        retrieved = self.search(question)
        context = "

".join(f"[{c['title']}]
{c['text']}" for c in retrieved)
        response = client.chat.completions.create(
            model="gpt-4o",
            messages=[
                {"role": "system", "content": (
                    "Answer questions ONLY using the context provided below. "
                    "If the answer is not in the context, state clearly that "
                    "you do not have that information. Do not use outside knowledge.

"
                    f"KNOWLEDGE BASE:
{context}"
                )},
                {"role": "user", "content": question}
            ],
            temperature=0.1
        )
        return response.choices[0].message.content

# Example usage
rag = RAGSystem()
rag.build_index([
    {"title": "Refund Policy",
     "content": "Full refunds available within 30 days of purchase. After 30 days, we offer store credit only. Contact billing@company.com with your order number. Processing takes 5-7 business days."},
    {"title": "Pro Plan Features",
     "content": "Pro plan includes unlimited workflows, priority support, and API access. Starter plan is limited to 10 active workflows and email support only."},
])
print(rag.answer("Can I get a refund after 30 days?"))

pip install pinecone-client
from pinecone import Pinecone, ServerlessSpec

pc = Pinecone(api_key="your-key")
pc.create_index("kb", dimension=1536, metric="cosine",
                spec=ServerlessSpec(cloud="aws", region="us-east-1"))
index = pc.Index("kb")

# Upsert
index.upsert([{"id": "chunk_001", "values": vector,
               "metadata": {"text": text, "source": "policy"}}])

# Query
results = index.query(vector=query_vec, top_k=3, include_metadata=True)
chunks = [r.metadata["text"] for r in results.matches]