shashank ms

Posted on Aug 22

Introduction to RAG Pipelines in LLM: A Beginner's Guide

#learnai #oxlo #ai

We are going to build a minimal RAG pipeline that indexes a raw text knowledge base and answers questions using retrieved context. This is for developers who need to ground LLM outputs in private documents without standing up complex infrastructure.

What you'll need

Python 3.10 or newer, the openai and numpy packages, and an API key from https://portal.oxlo.ai. Install the dependencies with pip.

pip install openai numpy

Step 1: Prepare your source document

I will start with a hardcoded internal wiki so the script is fully self-contained. In production, replace the string with a file read from your docs folder.

from openai import OpenAI
import numpy as np

client = OpenAI(base_url="https://api.oxlo.ai/v1", api_key="YOUR_OXLO_API_KEY")

RAW_DOCS = """
AcmeDB Cluster Setup
To set up an AcmeDB cluster, you need three nodes minimum.
The primary node handles all write operations.
Secondary nodes replicate the write-ahead log asynchronously.
Failover is automatic if the primary loses quorum.

AcmeDB Backup Policy
Full snapshots are taken every Sunday at 02:00 UTC.
Incremental backups run every six hours.
Retention is thirty days for snapshots and seven days for incrementals.
Restores require the snapshot plus every incremental taken after it.

AcmeDB Security
All connections must use TLS 1.3.
Client certificates are optional but recommended for service accounts.
The admin port is bound to localhost only by default.
"""

Step 2: Chunk the text into overlapping windows

Embeddings work best on focused passages, so we split the text into chunks with a small overlap to avoid cutting sentences in half.

def chunk_text(text, chunk_size=200, overlap=50):
    chunks = []
    start = 0
    while start < len(text):
        end = start + chunk_size
        chunk = text[start:end]
        if chunk.strip():
            chunks.append(chunk.strip())
        start = max(start + chunk_size - overlap, start + 1)
    return chunks

chunks = chunk_text(RAW_DOCS)
print(f"Created {len(chunks)} chunks")

Step 3: Embed the chunks with Oxlo.ai

We convert each chunk into a dense vector using Oxlo.ai's BGE-Large embedding endpoint. These vectors capture semantic meaning so we can find relevant text later.

def embed_texts(texts):
    response = client.embeddings.create(
        model="bge-large",
        input=texts
    )
    return [item.embedding for item in response.data]

chunk_embeddings = embed_texts(chunks)
print(f"Embedded {len(chunk_embeddings)} chunks")

Step 4: Build a simple vector store

We keep the vectors and their source chunks in memory. At query time, we compare the question vector against every chunk vector using cosine similarity.

def cosine_similarity(a, b):
    a = np.array(a)
    b = np.array(b)
    return np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b))

# Store as a list of (chunk_text, embedding) tuples
vector_store = list(zip(chunks, chunk_embeddings))

Step 5: Retrieve relevant chunks for a question

When a user asks a question, we embed it with the same model and return the top three most similar chunks.

def retrieve(query, top_k=3):
    query_embedding = embed_texts([query])[0]
    scored = []
    for chunk, emb in vector_store:
        score = cosine_similarity(query_embedding, emb)
        scored.append((score, chunk))
    scored.sort(reverse=True, key=lambda x: x[0])
    return [chunk for _, chunk in scored[:top_k]]

# Quick test
question = "How do I restore a backup?"
retrieved = retrieve(question)
for i, chunk in enumerate(retrieved, 1):
    print(f"\n--- Chunk {i} ---\n{chunk}")

Step 6: Generate an answer with context

We feed the retrieved chunks into a system prompt and ask Llama 3.3 70B on Oxlo.ai to answer strictly from the provided context.

SYSTEM_PROMPT = """You are a helpful support assistant.
Answer the user's question using ONLY the provided context.
If the context does not contain the answer, say 'I don't have that information.'
Do not make up facts."""

def answer_question(question):
    context_chunks = retrieve(question, top_k=3)
    context = "\n\n".join(context_chunks)
    
    user_message = f"Context:\n{context}\n\nQuestion: {question}"
    
    response = client.chat.completions.create(
        model="llama-3.3-70b",
        messages=[
            {"role": "system", "content": SYSTEM_PROMPT},
            {"role": "user", "content": user_message},
        ],
        temperature=0.1,
    )
    return response.choices[0].message.content

Run it

Call the function with a question that is covered by the docs, then one that is not.

if __name__ == "__main__":
    q1 = "What security protocol does AcmeDB require?"
    print(f"Q: {q1}")
    print(f"A: {answer_question(q1)}\n")
    
    q2 = "Does AcmeDB support graph queries?"
    print(f"Q: {q2}")
    print(f"A: {answer_question(q2)}")

Expected output:

Q: What security protocol does AcmeDB require?
A: AcmeDB requires all connections to use TLS 1.3.

Q: Does AcmeDB support graph queries?
A: I don't have that information.

Wrap-up

You now have a working RAG pipeline that runs entirely against Oxlo.ai. Because RAG payloads include retrieved context, they can get long. Oxlo.ai's flat per-request pricing (https://oxlo.ai/pricing) keeps your cost predictable even as you add more chunks, which makes it a solid fit for this pattern.

Two concrete next steps: swap the hardcoded string for a directory loader that reads every .md file with pathlib, and wrap the answer_question function in a FastAPI endpoint so other services can call it.

DEV Community