Retrieval-Augmented Generation (RAG) is the single most requested integration pattern we see from developers: let an LLM answer questions using your documents instead of whatever it memorized in training. In this tutorial you'll build a complete, working RAG chatbot in Python — document ingestion, vector search, and a streaming chat API — in about 100 lines of code, using any chat model available on Qubax.
By the end you'll have a script you can point at a folder of Markdown files and query in the terminal.
What You'll Build
- Ingestion — chunk your documents into overlapping pieces
- Embeddings — convert chunks into vectors and store them locally
- Retrieval — find the most relevant chunks for a user question
- Generation — send the question + retrieved context to a chat model, streaming the answer
Prerequisites
- Python 3.10+
- A Qubax API key (grab one at qubax.ai/docs)
- An embedding model and a chat model enabled on your account
pip install requests numpyWe'll use plain requests so you can port the code to any language or SDK easily.
Step 1: Configuration
import os, json, requests, numpy as np
BASE_URL = "https://api.qubax.ai/v1" # OpenAI-compatible endpoint
API_KEY = os.environ["QUBAX_API_KEY"]
CHAT_MODEL = "gpt-5.6-luna" # cheap, fast chat model
EMBED_MODEL = "text-embedding-3-small" # or any embedding model on your plan
def api(path, payload, stream=False):
return requests.post(
f"{BASE_URL}/{path}",
headers={"Authorization": f"Bearer {API_KEY}"},
json=payload,
stream=stream,
)Because Qubax exposes an OpenAI-compatible API, any code written for OpenAI works here with a base-URL change.
Step 2: Chunk the Documents
The #1 mistake in RAG systems is bad chunking. Too large and the context drowns the answer; too small and the chunks lack context. A reliable default: split on paragraph boundaries, target ~800 characters, with 100 characters of overlap.
def chunk_text(text: str, size: int = 800, overlap: int = 100) -> list[str]:
paragraphs = [p.strip() for p in text.split("\n\n") if p.strip()]
chunks, buf = [], ""
for para in paragraphs:
if len(buf) + len(para) < size:
buf += para + "\n\n"
else:
if buf:
chunks.append(buf.strip())
# start next chunk with the tail of the previous one (overlap)
buf = buf[-overlap:] + para + "\n\n"
if buf.strip():
chunks.append(buf.strip())
return chunksLoad every .md or .txt file in a folder:
from pathlib import Path
def load_chunks(folder: str) -> list[str]:
chunks = []
for path in Path(folder).rglob("*.md"):
text = path.read_text(encoding="utf-8")
for c in chunk_text(text):
chunks.append(f"[source: {path.name}]\n{c}")
return chunksPrepending the filename to each chunk is a small trick that measurably improves citation quality — the model can name its source.
Step 3: Embed and Store
For a local knowledge base of up to a few thousand chunks, a NumPy array beats a vector database on simplicity:
def embed(texts: list[str]) -> np.ndarray:
r = api("embeddings", {"model": EMBED_MODEL, "input": texts}).json()
vecs = [d["embedding"] for d in r["data"]]
arr = np.array(vecs, dtype=np.float32)
arr /= np.linalg.norm(arr, axis=1, keepdims=True) # normalize for cosine sim
return arr
chunks = load_chunks("./knowledge")
vectors = embed(chunks)
np.save("vectors.npy", vectors)
with open("chunks.json", "w") as f:
json.dump(chunks, f)Run this once (and re-run whenever documents change). At Flash-class embedding prices this costs pennies even for a large docs folder.
Step 4: Retrieve
At query time, embed the question and take the top-k chunks by cosine similarity:
def retrieve(question: str, k: int = 4) -> list[str]:
qvec = embed([question])[0]
scores = vectors @ qvec # cosine similarity, since both are normalized
top = np.argsort(scores)[-k:][::-1]
return [chunks[i] for i in top]k=4 is a good starting point. If your answers miss details, raise k before you blame the model — retrieval failure is far more common than model failure.
Step 5: Generate — With Streaming
Now assemble the prompt and stream the answer token by token:
SYSTEM = """You are a helpful assistant answering questions about the company's documentation.
Use ONLY the provided context. If the answer is not in the context, say you don't know.
Cite the source file for every claim."""
def ask(question: str):
context = "\n\n---\n\n".join(retrieve(question))
payload = {
"model": CHAT_MODEL,
"stream": True,
"messages": [
{"role": "system", "content": SYSTEM},
{"role": "user", "content": f"Context:\n{context}\n\nQuestion: {question}"},
],
}
with api("chat/completions", payload, stream=True) as r:
for line in r.iter_lines():
if not line or not line.startswith(b"data: "):
continue
data = line[6:]
if data == b"[DONE]":
break
delta = json.loads(data)["choices"][0]["delta"]
if delta.get("content"):
print(delta["content"], end="", flush=True)
if __name__ == "__main__":
while True:
q = input("\nYou: ")
if q in ("exit", "quit"):
break
print("Bot: ", end="")
ask(q)That's the whole system. Run it:
export QUBAX_API_KEY=sk-...
python rag_chatbot.py
You: What is our refund policy for annual plans?Production Checklist
Before this touches real users, address the five things every demo skips:
- Persist vectors properly. NumPy is fine to ~50k chunks; beyond that, use pgvector (Postgres), Qdrant, or Pinecone.
- Add reranking. Retrieve k=20 cheaply, then rerank to k=4 with a cross-encoder or a rerank model. This is the highest-ROI accuracy upgrade in RAG.
- Log retrieval scores. If top similarity scores are low, the knowledge base doesn't contain the answer — say so instead of letting the model hallucinate.
- Cache embeddings. Embed documents once, keyed by content hash. Never re-embed an unchanged document.
- Watch the cost math. Every question pays for k chunks of input tokens plus the answer. With a cheap chat model at ~$0.10/M input this is negligible; with a flagship it's 10–50x more. Route the simple questions to a cheap model and escalate on low-confidence answers.
Common Failure Modes
| Symptom | Likely cause | Fix |
|---|---|---|
| Answer ignores your docs | Retrieval returned wrong chunks | Improve chunking, raise k, add reranking |
| "I don't know" too often | Chunks too small or question too specific | Larger chunks (1200 chars), better query rewriting |
| Slow responses | Non-streaming, or huge context | Stream (as above), trim k |
| Hallucinated citations | No source labels in chunks | Keep the [source: file] prefix |
Next Steps
- Swap in different chat models and A/B the answer quality — qubax.ai/models lets you compare pricing side-by-side, and swapping is a one-line change since everything speaks the OpenAI-compatible protocol.
- Add conversation memory: keep the last N turns in the messages array.
- Add a feedback loop: log thumbs-up/down per answer, and use it to tune k and chunk size.
The full API reference, including streaming parameters and rate limits, is at qubax.ai/docs.
FAQ
How much does a RAG chatbot cost to run?
With a cheap chat model (e.g., GPT-5.6 Luna at ~$0.018/M input on Qubax), a typical query with 4 retrieved chunks (~3k tokens) plus a 500-token answer costs roughly $0.0001 — a tenth of a cent. With a flagship model the same query can cost 10–50x more.
Do I need a vector database?
Not initially. For up to ~50k chunks, normalized NumPy vectors with a dot-product search are fast (milliseconds) and dependency-free. Move to pgvector or Qdrant when your corpus outgrows RAM or you need metadata filtering.
Why is my chatbot hallucinating despite RAG?
Usually the retrieved context doesn't actually contain the answer, and the model improvises. Log your retrieval scores; if they're low, gate the answer ("I couldn't find this in the documentation") instead of generating.