Quickstart
On this page
RAGSystem does not ship a vector database, LLM, or chunker — it orchestrates whatever objects you give it, provided they satisfy the small interfaces documented on the RAGSystem page. This quickstart uses minimal in-memory stand-ins so the example is fully runnable without any external service.
1. Implement the four required components
import re
from dataclasses import dataclass
# ── A trivial keyword-overlap "vector" database ──────────────────────────────
class ToyVectorDB:
def __init__(self):
self._chunks = []
def add(self, chunks):
self._chunks.extend(chunks)
def search(self, query, top_k=5, score_threshold=0.0):
terms = set(re.findall(r"\w+", query.lower()))
scored = []
for c in self._chunks:
words = set(re.findall(r"\w+", c.text.lower()))
score = len(terms & words) / max(len(terms), 1)
if score >= score_threshold:
scored.append((c, score))
scored.sort(key=lambda x: x[1], reverse=True)
return scored[:top_k]
def remove_by_doc_id(self, doc_id):
before = len(self._chunks)
self._chunks = [c for c in self._chunks if c.doc_id != doc_id]
return before - len(self._chunks)
@property
def size(self):
return len(self._chunks)
# ── A minimal chunk container ────────────────────────────────────────────────
@dataclass
class Chunk:
text: str
doc_id: str
# ── A fixed-size chunker ─────────────────────────────────────────────────────
class SimpleChunker:
def chunk_text(self, text, doc_id, size=200):
return [Chunk(text=text[i:i + size], doc_id=doc_id)
for i in range(0, len(text), size)] or [Chunk(text=text, doc_id=doc_id)]
# ── A context builder ────────────────────────────────────────────────────────
class SimpleContextManager:
def build(self, query, chunks):
return "\n---\n".join(c.text for c, _score in chunks)
# ── A stub LLM (replace with a real client in production) ───────────────────
class EchoLLM:
def generate(self, prompt, **kwargs):
return "This is a stub answer generated from the retrieved context."
2. Build the RAG system
from zerda.rag.core import RAGSystem, RAGConfig
config = RAGConfig(
chunk_size=200,
top_k=3,
enable_reranking=False, # heuristic/LLM/hybrid — off by default here
)
rag = RAGSystem(
vector_db=ToyVectorDB(),
llm=EchoLLM(),
chunker=SimpleChunker(),
context_manager=SimpleContextManager(),
config=config,
)
3. Add documents
rag.add_texts({
"policy_1": "Employees are entitled to 21 days of annual leave per year.",
"policy_2": "Remote work requests must be approved by a direct manager.",
})
add_texts (and the lower-level add_documents) return {doc_id: num_chunks}.
4. Ask a question
answer = rag.ask("How many vacation days do employees get?")
print(answer)
ask() is a friendly alias for generate(). Both retrieve relevant chunks, build a prompt via PromptRouter (language + question-type aware), and call your LLM.
5. Inspect what happened
print(rag.get_stats())
# {'total_queries': 1, 'successful_queries': 1, 'failed_queries': 0,
# 'total_documents': 2, 'total_chunks': 2, 'vector_db_size': 2, ...}
for chunk, score in rag.retrieve("vacation days"):
print(f"{score:.2f} {chunk.text}")
6. Clean up
rag.cleanup() # calls .cleanup() on any component that defines it
# or, as a context manager:
with RAGSystem(vector_db=ToyVectorDB(), llm=EchoLLM(),
chunker=SimpleChunker(), context_manager=SimpleContextManager()) as rag2:
rag2.add_text("Some text.")
Where to go next
- Architecture — how ingestion, retrieval, and generation fit together.
- Configuration — every
RAGConfigfield explained. RAGSystemAPI reference — the full method-by-method reference, including the async API and persistence.- Cookbook: Async Usage — using
aask-style workflows andastream.