User Query → Embed → Vector Search → Retrieve Context → LLM Generate → Answer 

# embedder.py import os import requests import numpy as np from typing import List class Embedder: def __init__(self, provider: str = "openrouter"): self.provider = provider self.configs = { "openrouter": { "url": "https://openrouter.ai/api/v1/embeddings", "key": os.getenv("OPENROUTER_API_KEY"), "model": "text-embedding-3-small" # 1536 dims  }, "local": { "url": "http://localhost:11434/api/embeddings", "model": "nomic-embed-text" # 768 dims via Ollama  } } def embed(self, texts: List[str]) -> np.ndarray: config = self.configs[self.provider] if self.provider == "openrouter": headers = { "Authorization": f"Bearer {config['key']}", "Content-Type": "application/json" } response = requests.post( config["url"], headers=headers, json={"model": config["model"], "input": texts}, timeout=30 ) response.raise_for_status() data = response.json()["data"] embeddings = [d["embedding"] for d in sorted(data, key=lambda x: x["index"])] elif self.provider == "local": embeddings = [] for text in texts: r = requests.post(config["url"], json={"model": config["model"], "prompt": text}) r.raise_for_status() embeddings.append(r.json()["embedding"]) return np.array(embeddings, dtype=np.float32) def embed_one(self, text: str) -> np.ndarray: return self.embed([text])[0] 

# vector_store.py import numpy as np from abc import ABC, abstractmethod from typing import List, Dict, Any, Optional class VectorStore(ABC): @abstractmethod def add(self, vectors: np.ndarray, metadatas: List[Dict], ids: List[str]): ... @abstractmethod def search(self, query: np.ndarray, top_k: int = 5, filter: Optional[Dict] = None) -> List[Dict]: ... @abstractmethod def delete(self, ids: List[str]): ... # Qdrant Implementation class QdrantStore(VectorStore): def __init__(self, url: str, api_key: str, collection: str = "rag_docs"): from qdrant_client import QdrantClient from qdrant_client.models import Distance, VectorParams self.client = QdrantClient(url=url, api_key=api_key) self.collection = collection # Create collection if not exists  collections = self.client.get_collections().collections if collection not in [c.name for c in collections]: self.client.create_collection( collection_name=collection, vectors_config=VectorParams(size=1536, distance=Distance.COSINE) ) def add(self, vectors: np.ndarray, metadatas: List[Dict], ids: List[str]): from qdrant_client.models import PointStruct points = [ PointStruct(id=id_, vector=vec.tolist(), payload=meta) for id_, vec, meta in zip(ids, vectors, metadatas) ] self.client.upsert(collection_name=self.collection, points=points) def search(self, query: np.ndarray, top_k: int = 5, filter: Optional[Dict] = None) -> List[Dict]: from qdrant_client.models import Filter, FieldCondition, MatchValue q_filter = None if filter: conditions = [ FieldCondition(key=k, match=MatchValue(value=v)) for k, v in filter.items() ] q_filter = Filter(must=conditions) results = self.client.query_points( collection_name=self.collection, query=query.tolist(), query_filter=q_filter, limit=top_k, with_payload=True ) return [ {"id": str(r.id), "score": r.score, "metadata": r.payload} for r in results.points ] def delete(self, ids: List[str]): self.client.delete(collection_name=self.collection, points_selector=ids) # Simple NumPy Store (for testing, no external DB) class NumpyStore(VectorStore): def __init__(self, dim: int = 1536): self.vectors = np.empty((0, dim), dtype=np.float32) self.metadatas = [] self.ids = [] def add(self, vectors: np.ndarray, metadatas: List[Dict], ids: List[str]): self.vectors = np.vstack([self.vectors, vectors]) self.metadatas.extend(metadatas) self.ids.extend(ids) def search(self, query: np.ndarray, top_k: int = 5, filter: Optional[Dict] = None) -> List[Dict]: # Cosine similarity  norms = np.linalg.norm(self.vectors, axis=1) query_norm = np.linalg.norm(query) if query_norm == 0 or np.any(norms == 0): return [] similarities = (self.vectors @ query) / (norms * query_norm) # Apply filter  if filter: mask = np.ones(len(self.metadatas), dtype=bool) for k, v in filter.items(): mask &= np.array([m.get(k) == v for m in self.metadatas]) similarities[~mask] = -1 top_indices = np.argsort(similarities)[-top_k:][::-1] return [ {"id": self.ids[i], "score": float(similarities[i]), "metadata": self.metadatas[i]} for i in top_indices if similarities[i] > 0 ] def delete(self, ids: List[str]): indices = [self.ids.index(id_) for id_ in ids if id_ in self.ids] self.vectors = np.delete(self.vectors, indices, axis=0) for i in sorted(indices, reverse=True): del self.metadatas[i] del self.ids[i] 

# chunker.py from typing import List import re def chunk_text(text: str, chunk_size: int = 500, overlap: int = 50) -> List[str]: """Split text into overlapping chunks by sentences.""" sentences = re.split(r'(?<=[.!?])\s+', text) chunks = [] current_chunk = [] current_length = 0 for sentence in sentences: sent_len = len(sentence) if current_length + sent_len > chunk_size and current_chunk: chunks.append(" ".join(current_chunk)) # Keep overlap  overlap_text = " ".join(current_chunk)[-overlap:] current_chunk = [overlap_text, sentence] current_length = len(overlap_text) + sent_len else: current_chunk.append(sentence) current_length += sent_len if current_chunk: chunks.append(" ".join(current_chunk)) return chunks def chunk_markdown(text: str, chunk_size: int = 500) -> List[str]: """Chunk preserving markdown structure.""" # Split by headers first  sections = re.split(r'(^#{1,3}\s+.+$)', text, flags=re.MULTILINE) chunks = [] current = "" for section in sections: if not section.strip(): continue if len(current) + len(section) > chunk_size and current: chunks.append(current.strip()) current = section else: current += "\n" + section if current.strip(): chunks.append(current.strip()) return chunks 

# retriever.py import numpy as np from typing import List, Dict, Optional from embedder import Embedder from vector_store import VectorStore class Retriever: def __init__(self, embedder: Embedder, store: VectorStore): self.embedder = embedder self.store = store def add_documents(self, texts: List[str], metadatas: List[Dict], ids: List[str]): """Embed and store documents.""" vectors = self.embedder.embed(texts) self.store.add(vectors, metadatas, ids) def add_file(self, filepath: str, metadata: Dict = None, chunk_size: int = 500): """Read, chunk, embed, and store a file.""" from chunker import chunk_text with open(filepath, 'r') as f: content = f.read() chunks = chunk_text(content, chunk_size=chunk_size) base_id = filepath.replace("/", "_").replace(".", "_") ids = [f"{base_id}_{i}" for i in range(len(chunks))] metadatas = [ {**(metadata or {}), "source": filepath, "chunk": i, "text": chunk} for i, chunk in enumerate(chunks) ] self.add_documents(chunks, metadatas, ids) return len(chunks) def retrieve(self, query: str, top_k: int = 5, filter: Optional[Dict] = None) -> List[Dict]: """Retrieve relevant chunks for a query.""" query_vector = self.embedder.embed_one(query) return self.store.search(query_vector, top_k=top_k, filter=filter) 

# generator.py import os import requests from typing import List, Dict class Generator: def __init__(self, provider: str = "openrouter"): self.provider = provider self.configs = { "openrouter":Source：https://dev.to/ipsphil_digital/building-rag-from-scratch-no-frameworks-needed-3j49
My favorites