RAG & Vector Bases Skill

Workflows für Retrieval-Augmented Generation, Embeddings, Vector-DBs, Chunking-Strategien.

Chunking Strategien

Fixed-Size Chunking

python
def fixed_chunks(text: str, chunk_size: int = 512, overlap: int = 64) -> list[str]:
    chunks = []
    start = 0
    while start < len(text):
        end = start + chunk_size
        chunks.append(text[start:end])
        start += chunk_size - overlap
    return chunks

Semantic Chunking (Recursive)

python
from langchain.text_splitter import RecursiveCharacterTextSplitter

splitter = RecursiveCharacterTextSplitter(
    chunk_size=1000,
    chunk_overlap=200,
    separators=["\n## ", "\n### ", "\n\n", "\n", ". ", " ", ""],
)
chunks = splitter.split_text(text)
# Mit Metadaten
docs = splitter.create_documents(texts=[text], metadatas=[{"source": "file.md", "page": 1}])

Embeddings

OpenAI-kompatibel (via OmniRoute)

python
import httpx

resp = httpx.post(
    "http://localhost:20128/v1/embeddings",
    json={
        "model": "oc/deepseek-v4-flash-free",
        "input": ["Text to embed", "Another text"],
    },
)
embeddings = resp.json()["data"]  # [{embedding: [...], index: 0}, ...]

Sentence Transformers (lokal)

python
from sentence_transformers import SentenceTransformer

model = SentenceTransformer("all-MiniLM-L6-v2")  # 384-dim
embeddings = model.encode(["Text 1", "Text 2"], normalize_embeddings=True)

Vector Databases

ChromaDB (einfach, embedded)

python
import chromadb

client = chromadb.Client()
collection = client.create_collection(name="docs")
collection.add(
    ids=["doc1", "doc2"],
    documents=["Content A", "Content B"],
    embeddings=embeddings.tolist(),
    metadatas=[{"source": "a.txt"}, {"source": "b.txt"}],
)
results = collection.query(query_embeddings=[query_emb], n_results=3)

Qdrant (Docker, production)

bash
docker run -d --name qdrant -p 6333:6333 qdrant/qdrant

python
from qdrant_client import QdrantClient

client = QdrantClient("localhost", port=6333)
client.upsert(collection_name="docs", points=[{"id": 1, "vector": [0.1] * 384, "payload": {"text": "Content"}}])
results = client.search(collection_name="docs", query_vector=[0.1] * 384, limit=3)

PGVector (PostgreSQL)

bash
docker run -d --name pgvector -e POSTGRES_PASSWORD=pass -p 5432:5432 pgvector/pgvector:pg16

sql
CREATE EXTENSION vector;
CREATE TABLE docs (id SERIAL PRIMARY KEY, embedding vector(384), text text);
INSERT INTO docs (embedding, text) VALUES ('[0.1,0.2,...]', 'Content');
SELECT * FROM docs ORDER BY embedding <-> '[0.1,0.2,...]' LIMIT 3;

RAG Pipeline

python
def rag_pipeline(query: str, top_k: int = 3):
    # 1. Embed Query
    q_emb = model.encode([query], normalize_embeddings=True)[0]
    # 2. Retrieve
    results = collection.query(query_embeddings=[q_emb.tolist()], n_results=top_k)
    # 3. Augment
    context = "\n\n".join(results["documents"][0])
    # 4. Generate via OmniRoute
    resp = httpx.post(
        "http://localhost:20128/v1/chat/completions",
        json={
            "model": "oc/deepseek-v4-flash-free",
            "messages": [
                {"role": "system", "content": f"Context:\n{context}\n\nAnswer using context only."},
                {"role": "user", "content": query},
            ],
        },
    )
    return resp.json()["choices"][0]["message"]["content"]

Evaluierung

python
# Retrieval Quality
def hit_rate(results, relevant_docs):
    return len(set(results) & set(relevant_docs)) / len(relevant_docs)


def mrr(results, relevant_docs):
    for i, r in enumerate(results):
        if r in relevant_docs:
            return 1 / (i + 1)
    return 0