Build a RAG pipeline in pure Python

def chunk_document(text, doc_id, words_per_chunk=12, overlap=3):
    words = text.split()
    step = words_per_chunk - overlap
    chunks = []
    for start in range(0, len(words), step):
        piece = words[start:start + words_per_chunk]
        if not piece:
            break
        chunks.append({
            "doc_id": doc_id,
            "chunk_id": f"{doc_id}-{len(chunks)}",
            "text": " ".join(piece),
        })
        if start + words_per_chunk >= len(words):
            break
    return chunks

doc = ("MLOps is the practice of deploying and maintaining machine learning models in production. "
       "It covers CI/CD, monitoring, retraining, and governance for model systems.")

for chunk in chunk_document(doc, "mlops-intro"):
    print(chunk["chunk_id"], "|", chunk["text"])

import numpy as np

VOCAB = ["mlops", "llmops", "rag", "python", "kubernetes", "monitoring",
         "retrieval", "model", "production", "agents", "vector", "prompt"]

def embed(text):
    vec = np.zeros(len(VOCAB), dtype=np.float32)
    words = text.lower().replace(",", " ").replace(".", " ").split()
    for word in words:
        if word in VOCAB:
            vec[VOCAB.index(word)] += 1.0
    norm = np.linalg.norm(vec)
    return vec / norm if norm > 0 else vec

corpus = [
    {"id": "c1", "text": "MLOps handles model deployment monitoring and production reliability"},
    {"id": "c2", "text": "RAG uses retrieval to ground a prompt in your own documents"},
    {"id": "c3", "text": "Kubernetes runs containers and scales model serving workloads"},
    {"id": "c4", "text": "LLMOps adds prompt versioning and evaluation for production agents"},
]

matrix = np.vstack([embed(c["text"]) for c in corpus])

def retrieve(query, top_k=2, floor=0.15):
    scores = matrix @ embed(query)
    order = np.argsort(-scores)[:top_k]
    return [
        {**corpus[i], "score": float(scores[i])}
        for i in order
        if scores[i] >= floor
    ]

for query in ["how do I ground a prompt in documents", "scaling model serving"]:
    print("query:", query)
    hits = retrieve(query)
    if not hits:
        print("  no confident match — answer: I do not know")
    for hit in hits:
        print(f"  {hit['id']} score={hit['score']:.3f} :: {hit['text'][:50]}")
    print()

def build_rag_prompt(question, chunks):
    if not chunks:
        return None
    sources = "\n\n".join(
        f"[{i}] (source: {c['id']})\n{c['text']}"
        for i, c in enumerate(chunks, start=1)
    )
    return (
        "You are a precise assistant.\n"
        "Answer using ONLY the numbered sources below.\n"
        "Cite sources inline like [1].\n"
        "If the sources do not contain the answer, reply exactly: I do not know.\n\n"
        f"SOURCES:\n{sources}\n\n"
        f"QUESTION: {question}\n"
        "ANSWER:"
    )

chunks = [
    {"id": "c2", "text": "RAG uses retrieval to ground a prompt in your own documents."},
    {"id": "c4", "text": "LLMOps adds prompt versioning and evaluation."},
]
prompt = build_rag_prompt("What is RAG?", chunks)
print(prompt)
print("\nprompt chars:", len(prompt), "| est. tokens:", len(prompt) // 4)

import numpy as np

corpus = [
    "error code E1042 means the model server ran out of GPU memory",
    "vector databases store embeddings for semantic retrieval",
    "the training job failed because the batch size was too large",
]

def keyword_score(query, text):
    q = set(query.lower().split())
    t = set(text.lower().split())
    return len(q & t) / max(1, len(q))

def vector_score(query, text):
    q = set(query.lower().split())
    t = set(text.lower().split())
    return len(q & t) / max(1, len(q | t))

query = "E1042 GPU memory"
alpha = 0.6

ranked = sorted(
    (
        (alpha * keyword_score(query, doc) + (1 - alpha) * vector_score(query, doc), doc)
        for doc in corpus
    ),
    reverse=True,
)
for score, doc in ranked:
    print(f"{score:.3f}  {doc[:60]}")