Blog
Hybride Suche lokal bauen mit SQLite FTS5, EmbeddingGemma 2 und Python
- Embeddings
- Semantische Suche
- Lokale KI
- RAG
Diese Anleitung baut hybride Suche in Python auf deinen eigenen Dateien. SQLite FTS5 übernimmt die Keywords, Vektoren von EmbeddingGemma 2 übernehmen die Bedeutung, gewichtete Reciprocal Rank Fusion mischt beide Listen, ein Reranker mit Gate liest knappe Fälle neu, und MRR plus Hit@k sagen dir, ob etwas davon geholfen hat. Sie folgt Finer, meiner lokalen Suche (Quellcode unter Quellen verlinkt), eingedampft auf wenige hundert Zeilen.
Die Begründung wiederhole ich hier nicht. Warum jede Entscheidung so gefallen ist, mit Messwerten, steht in EmbeddingGemma 2 lokale Suche auf dem Mac. Die Snippets ohne Modell laufen gegen eine SQLite Datenbank im Speicher, mit installiertem NumPy, und die Zahlen unten stammen aus zwei davon. Blöcke mit dem Vermerk „An deine Laufzeit anpassen“ laden ein Modell herunter, dieser Teil liegt bei dir.
Kernpunkte
- Mische Ranking Listen mit gewichteter Reciprocal Rank Fusion, k = 60. Sie braucht keine Score Normierung, weil nur Ränge zählen, darum lassen sich BM25 Werte und Kosinus Werte direkt kombinieren.
- Nutze die Task Präfixe von EmbeddingGemma 2 exakt: „task: search result | query: “ für Anfragen und „title: {title} | text: {content}“ für Chunks. Rechne in bfloat16 oder float32, nie in float16.
- Ein 1-Bit Index mit Shortlist in 16-facher Größe von k plus exaktem Rescoring sind wenige Zeilen NumPy. Miss die Top-10 Überlappung mit einem exakten Scan auf deinen eigenen Vektoren, bevor du ihm vertraust.
- Lass einen Reranker nur laufen, wenn die zwei besten Scores nah beieinander liegen, und mische per Rang, nicht per Wahrscheinlichkeit.
- Schreib etwa 30 Fragen mit erwarteten Dateien, bewerte mit MRR und Hit@k, und nimm eine Tuning Änderung nur an, wenn sie eine Leitplanke passiert.
Was du baust
Neun Schritte, in der Reihenfolge von Finer. Jedes Snippet erweitert dieselbe Datei, hybrid.py, und nutzt die Namen und Imports der Schritte davor. Du brauchst Python 3.10 oder neuer, NumPy und ein SQLite mit FTS5. Das Embedding Modell und der Reranker laden von Hugging Face, und die offiziellen Embedding Gewichte sind etwa 1,5 GB groß.
Bewusst draußen: OCR, Audio und Video, die Englisch zu Deutsch Brücke, Favoriten und Query Parsing. Dazu steht alles im Grundlagen Beitrag.
- Text extrahieren und in Chunks teilen.
- Die Chunks in eine FTS5 Tabelle legen und mit BM25 ranken.
- Die Chunks mit EmbeddingGemma 2 und seinen Task Präfixen einbetten.
- Die Vektoren in SQLite speichern.
- Einen 1-Bit Index bauen und seine Shortlist exakt nachbewerten.
- Die Listen mit gewichteter Reciprocal Rank Fusion mischen.
- Nur knappe Fälle neu ranken.
- Mit einer kleinen Gold Menge, MRR und Hit@k evaluieren.
- Optional: Gewichte hinter einer Leitplanke tunen.
SQLite FTS5 einrichten und „no such module: fts5“ beheben
FTS5 steckt in den meisten SQLite Builds. Wenn CREATE VIRTUAL TABLE mit no such module: fts5 scheitert, hängt dein Python an einem SQLite ohne FTS5, dann nimm einen Python Build, dessen SQLite FTS5 enthält. Das Schema in Schritt 2 nutzt contentless_delete, das braucht SQLite 3.43. Auf einem älteren Build löschst du die Zeile mit content='' und contentless_delete=1 aus dem Schema, und die Tabelle behält ihre eigene Kopie des Texts. Prüfe deinen Build:
import sqlite3
print("SQLite", sqlite3.sqlite_version) # contentless_delete braucht 3.43 oder neuer
con = sqlite3.connect(":memory:")
con.execute("CREATE VIRTUAL TABLE t USING fts5(x)") # wirft 'no such module: fts5' wenn es fehlt
print("FTS5 is available")Dateien extrahieren und in Chunks teilen
Die Extraktion hängt von deinen Dateien ab. Reiner Text und Markdown brauchen nichts. Für PDFs, Word Dateien und E-Mails nimm einen Extraktor, dem du vertraust; der Grundlagen Beitrag nennt einige Optionen für Linux und Windows, als Vorschläge gemeint; Finer nutzt keinen davon. Die Einheit, in der du suchst, ist der Chunk.
Finer teilt nach Zeichen: etwa 1.100 pro Chunk, Texte bis 1.600 Zeichen bleiben am Stück, 150 Zeichen Überlappung, und Struktur zuerst wo es sie gibt (Überschriften, Seiten, Folien, Tabellenblätter). Zeichen zählen braucht keinen Tokenizer. Das hier ist eine vereinfachte Version.
import re
def chunk_text(text: str, target: int = 1100, whole: int = 1600, overlap: int = 150) -> list[str]:
"""Packt Sätze in Chunks von etwa target Zeichen, mit kurzer Überlappung."""
text = text.strip()
if len(text) <= whole: # kurze Texte bleiben am Stück
return [text] if text else []
pieces = []
for block in re.split(r"\n\s*\n", text): # erst Leerzeilen, dann Sätze
for sent in re.findall(r"[^.!?:;\n]+[.!?:;\n]*", block):
sent = sent.strip()
while len(sent) > target: # sehr lange Sätze hart schneiden
pieces.append(sent[:target])
sent = sent[target - overlap:]
if sent:
pieces.append(sent)
chunks, cur = [], ""
for p in pieces:
if cur and len(cur) + len(p) + 1 > target:
chunks.append(cur)
cur = cur[-overlap:] + " " + p # ein Schwänzchen in den nächsten Chunk tragen
else:
cur = f"{cur} {p}".strip()
if cur:
chunks.append(cur)
return chunksFTS5 Index anlegen und mit BM25 ranken
Eine FTS5 Tabelle mit drei Spalten: Name, Überschrift und Text. Die Namensspalte hält die Pfadwörter, damit eine Datei über ihren Namen gefunden wird. Die Zeilen sind contentless: der Text liegt in einer normalen Items Tabelle, FTS5 hält nur den Index. Der unicode61 Tokenizer mit remove_diacritics 2 lässt „Müller“ auch „Muller“ finden, und prefix='3' legt einen Präfix Index an.
Die Funktion bm25() liefert kleinere Zahlen für bessere Treffer, weil FTS5 den Score mit minus eins multipliziert, darum sortiert die Abfrage aufsteigend. Ihre Argumente sind ein Gewicht pro Spalte: ein Treffer im Namen zählt viermal so viel wie im Fließtext, einer in der Überschrift doppelt. Die FTS5 Dokumentation setzt k1 auf 1,2 und b auf 0,75. Der Abfrage Bauer verodert die Wörter, wirft ein paar Stoppwörter raus und macht aus Wörtern ab vier Buchstaben Präfix Terme. Die Stoppwort Menge im Snippet ist ein Platzhalter; nimm echte Listen für deine Sprachen.
import re
import sqlite3
SCHEMA = """
CREATE TABLE IF NOT EXISTS files (id INTEGER PRIMARY KEY, path TEXT UNIQUE);
CREATE TABLE IF NOT EXISTS items (
id INTEGER PRIMARY KEY, file_id INTEGER, heading TEXT, text TEXT);
CREATE VIRTUAL TABLE IF NOT EXISTS fts USING fts5(
name, heading, text,
content='', contentless_delete=1, -- braucht SQLite 3.43 oder neuer
tokenize='unicode61 remove_diacritics 2', prefix='3');
"""
STOP = {"the", "and", "for", "with", "der", "die", "das", "und", "mit", "von", "ein"}
def open_db(path: str = ":memory:") -> sqlite3.Connection:
con = sqlite3.connect(path)
con.executescript(SCHEMA)
return con
def path_words(path: str) -> str:
"""'2024/Mietvertrag_final.pdf' -> '2024 Mietvertrag final pdf'"""
words = re.sub(r"[_\-./]+", " ", path)
return re.sub(r"(?<=[a-z])(?=[A-Z])", " ", words).strip()
def add_file(con, path: str, chunks: list[tuple[str, str]]) -> list[int]:
"""chunks = [(heading, text), ...]. Gibt die neuen Item Ids zurück."""
file_id = con.execute("INSERT INTO files(path) VALUES (?)", (path,)).lastrowid
ids = []
for heading, text in chunks:
item_id = con.execute(
"INSERT INTO items(file_id, heading, text) VALUES (?,?,?)",
(file_id, heading, text)).lastrowid
con.execute("INSERT INTO fts(rowid, name, heading, text) VALUES (?,?,?,?)",
(item_id, path_words(path), heading, text))
ids.append(item_id)
return ids
def fts_query(q: str) -> str:
"""ODER der Abfrage Wörter; Wörter ab 4 Buchstaben matchen auch als Präfix."""
terms = [t for t in re.findall(r"\w+", q.lower()) if t not in STOP]
terms = list(dict.fromkeys(terms))
return " OR ".join(f'"{t}"*' if len(t) >= 4 and not t.isdigit() else f'"{t}"'
for t in terms)
def keyword_search(con, q: str, limit: int = 60) -> list[int]:
match = fts_query(q)
if not match:
return []
rows = con.execute( # bm25 ist negativ: kleiner ist besser
"SELECT rowid FROM fts WHERE fts MATCH ? ORDER BY bm25(fts, 4.0, 2.0, 1.0) LIMIT ?",
(match, limit))
return [r[0] for r in rows]Embeddings mit EmbeddingGemma 2 und den richtigen Task Präfixen
EmbeddingGemma 2 erwartet kurze Präfixe, und die Modellkarte sagt, dass Text Eingaben sie tragen sollen. Anfragen bekommen „task: search result | query: “ vorangestellt. Dokumente bekommen „title: {title} | text: {content}“, mit „title: none“ ohne Titel. Bilder, Video und Audio bekommen kein Präfix. Ohne Präfix läuft es auch, kann aber Qualität kosten. In sentence-transformers legt prompt_name="SearchQuery" das Anfrage Präfix an. Der Document Prompt setzt „title: none“, darum müssen betitelte Dokumente von Hand formatiert werden, wie im nächsten Block.
Zwei Regeln von der Karte dazu. Rechne in bfloat16 oder float32, nie in float16: die Aktivierungen sprengen den float16 Bereich, und du bekommst NaN oder still schlechtere Vektoren. Und halte Vektoren auf Einheitslänge, weil der Suchcode unten davon ausgeht.
Das Laden hängt von deiner Plattform ab. Der erste Block nutzt sentence-transformers und lädt nur den Text Encoder über config_kwargs, die Einstellung, die die Modellkarte für reines Text Laden nennt. Der zweite ist die Apple Silicon Route, übernommen aus der mlx-community Modellkarte. Beide liefern dieselbe Funktion embed(texts), und beide erwarten das Präfix schon auf dem Text. Bette Chunks in längensortierten Batches ein, damit das Padding klein bleibt. Finer nutzt Batches von 32, oder 16 wenn der längste Prompt im Batch 600 Zeichen erreicht.
# An deine Laufzeit anpassen. pip install -U sentence-transformers transformers
import numpy as np
import torch
from sentence_transformers import SentenceTransformer
# bfloat16 wo die Hardware es kann, sonst float32. Nie float16.
dtype = torch.bfloat16 if torch.cuda.is_available() and torch.cuda.is_bf16_supported() else torch.float32
model = SentenceTransformer(
"google/embeddinggemma-2",
model_kwargs={"torch_dtype": dtype},
config_kwargs={"vision_config": None, "audio_config": None}, # nur Text
)
def embed(texts: list[str]) -> np.ndarray:
"""Texte tragen ihr Präfix schon: QUERY + Frage, oder doc_prompt(title, chunk)."""
vecs = model.encode(texts, batch_size=16, normalize_embeddings=True)
return np.asarray(vecs, dtype=np.float32) # (n, 768), Einheitslänge# An deine Laufzeit anpassen. Apple Silicon, aus der mlx-community Modellkarte:
# pip install "git+https://github.com/Blaizzy/mlx-vlm.git@3d87e88402f307efbf68e568971aa887ee7d9ed0"
# pip install -U "mlx>=0.32.3" "transformers>=5.18.0"
import mlx.core as mx
import numpy as np
from mlx_vlm.embedding_loader import load_embedding_model
from mlx_vlm.utils import get_model_path
from transformers import AutoTokenizer
path = get_model_path("mlx-community/embeddinggemma-2-8bit")
model = load_embedding_model(path)
tokenizer = AutoTokenizer.from_pretrained(path)
def embed(texts: list[str]) -> np.ndarray:
enc = tokenizer(texts, padding=True, truncation=True, max_length=2048, return_tensors="np")
out = model(**{k: mx.array(v) for k, v in enc.items()}).text_embeds
return np.array(out.astype(mx.float32)) # (n, 768), EinheitslängeVektoren in SQLite speichern, mit oder ohne sqlite-vec
SQLite hat keinen eigenen Vektortyp. Der einfachste Speicher ist ein BLOB pro Chunk. Finer hält zwei: den exakten Vektor als float16, 1.536 Bytes, und eine Kopie, die nur das Vorzeichen jeder Dimension behält, gepackt in 96 Bytes. float16 Speicherung ist in Ordnung, obwohl das Modell nicht in float16 rechnen darf. Der Cast passiert nach der Inferenz, auf Vektoren in Einheitslänge.
Eine Cache Tabelle mit einem Hash des Prompts als Schlüssel sorgt dafür, dass gleiche Prompts, etwa Kopien einer Datei, ihren Vektor wiederverwenden statt das Modell zu rufen. Finers README sagt, das drücke das erneute Einbetten eines unveränderten Ordners von 320 Sekunden auf 6, und kein Skript im Repo reproduziert das.
import hashlib
import numpy as np
QUERY = "task: search result | query: "
def doc_prompt(title: str, chunk: str) -> str:
title = title.replace("|", " ")[:200] or "none"
return f"title: {title} | text: {chunk}"
VECTOR_SCHEMA = """
CREATE TABLE IF NOT EXISTS vectors (item_id INTEGER PRIMARY KEY, v BLOB); -- float16, 1536 Bytes
CREATE TABLE IF NOT EXISTS vbits (item_id INTEGER PRIMARY KEY, b BLOB); -- Vorzeichen Bits, 96 Bytes
CREATE TABLE IF NOT EXISTS vcache (key BLOB PRIMARY KEY, v BLOB); -- Prompt Hash -> Vektor
"""
def embed_cached(con, prompts: list[str], embed) -> np.ndarray:
"""Bettet nur nie gesehene Prompts ein. Gleiche Prompts (Kopien einer Datei) kosten nichts."""
keys = [hashlib.blake2b(p.encode(), digest_size=16).digest() for p in prompts]
got = {}
for k in set(keys):
row = con.execute("SELECT v FROM vcache WHERE key=?", (k,)).fetchone()
if row:
got[k] = np.frombuffer(row[0], np.float16).astype(np.float32)
todo = {k: p for k, p in zip(keys, prompts) if k not in got}
if todo:
vecs = embed(list(todo.values()))
for k, v in zip(todo, vecs):
v16 = (v / np.linalg.norm(v)).astype(np.float16)
con.execute("INSERT OR REPLACE INTO vcache VALUES (?,?)", (k, v16.tobytes()))
got[k] = v16.astype(np.float32)
return np.stack([got[k] for k in keys])
def store_vector(con, item_id: int, v: np.ndarray) -> None:
v = v / np.linalg.norm(v)
con.execute("INSERT OR REPLACE INTO vectors VALUES (?,?)", (item_id, v.astype(np.float16).tobytes()))
con.execute("INSERT OR REPLACE INTO vbits VALUES (?,?)", (item_id, np.packbits(v > 0).tobytes()))1-Bit Index plus exaktes Rescoring
Jeden Float Vektor zu scannen kostet mit jedem Chunk mehr. Die Abkürzung: Vorzeichen Bits per Hamming Distanz vergleichen (XOR, dann Einsen zählen), eine Shortlist von k mal 16 Chunks nehmen und nur die exakt per Kosinus nachbewerten. Das exakte Rescoring stellt die Reihenfolge wieder her; die 16-fache Shortlist sorgt dafür, dass die echten Nachbarn drin sind.
Finers README behauptet 99 Prozent gleiche Top-10 gegen exakte Suche bei etwa 10-facher Geschwindigkeit. Kein Skript im Repo erzeugt diese Zahl, und das README nennt keinen Korpus. Darum liegt unten overlap_at bei, das misst, welchen Anteil der exakten Top-k die Abkürzung findet, auf deinen Vektoren. Der Check danach baut 30.000 synthetische geclusterte Vektoren und druckt die Überlappung bei 1-facher, 4-facher und 16-facher Überabtastung: 0,4, 0,81 und 1,0. Auf 30.000 rein zufälligen Vektoren, in denen es keine Struktur zu finden gibt, liefert 16-fach 0,42. Der Check kann eine Minute oder so dauern, und die Werte wandern mit dem Seed. Synthetische Daten sagen nichts über EmbeddingGemma 2, also lass overlap_at auf deinen eigenen Vektoren laufen.
import numpy as np
POPCOUNT = np.array([bin(i).count("1") for i in range(256)], dtype=np.uint8)
def load_bits(con):
"""Im Speicher halten und neu laden wenn sich die Tabelle ändert."""
rows = con.execute("SELECT item_id, b FROM vbits ORDER BY item_id").fetchall()
ids = np.array([r[0] for r in rows], dtype=np.int64)
bits = np.frombuffer(b"".join(r[1] for r in rows), np.uint8).reshape(len(rows), -1)
return ids, bits
def rescore(con, ids, qv: np.ndarray, k: int):
"""Exakter Kosinus auf den float16 Vektoren. ids=None scannt jeden Vektor."""
if ids is None:
rows = con.execute("SELECT item_id, v FROM vectors").fetchall()
else:
marks = ",".join("?" * len(ids))
rows = con.execute(f"SELECT item_id, v FROM vectors WHERE item_id IN ({marks})",
[int(i) for i in ids]).fetchall()
M = np.stack([np.frombuffer(v, np.float16).astype(np.float32) for _, v in rows])
sims = (M @ qv) / np.linalg.norm(M, axis=1)
return [(rows[i][0], float(sims[i])) for i in np.argsort(-sims)[:k]]
def vector_search(con, index, qv: np.ndarray, k: int = 60, oversample: int = 16):
"""1-Bit Hamming Shortlist aus k mal oversample Items, dann exaktes Rescoring."""
ids, bits = index
if len(ids) == 0:
return []
dist = POPCOUNT[bits ^ np.packbits(qv > 0)].sum(axis=1, dtype=np.int32)
n = min(len(ids), k * oversample)
cand = ids[np.argpartition(dist, n - 1)[:n]] if n < len(ids) else ids
return rescore(con, cand, qv, k)
def overlap_at(con, index, queries, k: int = 10, oversample: int = 16) -> float:
"""Anteil der exakten Top-k, die der 1-Bit Pfad auch liefert. Auf deinen eigenen Vektoren laufen lassen."""
hits = 0
for qv in queries:
exact = {i for i, _ in rescore(con, None, qv, k)}
fast = {i for i, _ in vector_search(con, index, qv, k, oversample)}
hits += len(exact & fast)
return hits / (k * len(queries))# Schritt 5, Check. Braucht VECTOR_SCHEMA und store_vector (Schritt 4), load_bits und overlap_at (Schritt 5).
import sqlite3
import numpy as np
def unit(a: np.ndarray) -> np.ndarray:
return a / np.linalg.norm(a, axis=1, keepdims=True)
def overlap_demo(X: np.ndarray, Q: np.ndarray, oversamples=(1, 4, 16)) -> dict:
con = sqlite3.connect(":memory:")
con.executescript(VECTOR_SCHEMA)
for i, v in enumerate(X, start=1):
store_vector(con, i, v)
index = load_bits(con)
return {o: round(overlap_at(con, index, list(Q), 10, o), 2) for o in oversamples}
rng = np.random.default_rng(7)
n, clusters, dims = 30_000, 300, 768
centers = rng.standard_normal((clusters, dims)).astype(np.float32)
X = unit(centers[rng.integers(0, clusters, n)] + 0.9 * rng.standard_normal((n, dims)).astype(np.float32))
Q = unit(X[rng.integers(0, n, 60)] + 1.5 * rng.standard_normal((60, dims)).astype(np.float32) / np.sqrt(dims))
print("clustered:", overlap_demo(X, Q))
R = unit(rng.standard_normal((n, dims)).astype(np.float32))
QR = unit(R[rng.integers(0, n, 40)] + 0.7 * rng.standard_normal((40, dims)).astype(np.float32) / np.sqrt(dims))
print("random: ", overlap_demo(R, QR, (1, 16)))
# Ausgabe, mit Seed 7 und NumPy 2.5.3 (die Werte wandern mit dem Seed):
# clustered: {1: 0.4, 4: 0.81, 16: 1.0}
# random: {1: 0.17, 16: 0.42}Die Shortlist an sqlite-vec abgeben
Die NumPy Version hält alle Bits im Speicher: 96 Bytes pro Chunk, etwa 29 MB für 300.000 Chunks. Wenn du lieber SQL magst, legt sqlite-vec vec0 Tabellen mit einer bit[768] Spalte an (seine Anleitung zur binären Quantisierung zeigt eine) plus eine Nächste Nachbarn Abfrage. Es braucht pip install sqlite-vec und ein Python, das Extensions laden kann. Vergleiche seine Shortlist mit dem NumPy Pfad auf deinen eigenen Vektoren, bevor du dich darauf verlässt.
# Optional: die 1-Bit Shortlist an sqlite-vec abgeben. pip install sqlite-vec
# Braucht ein Python, dessen sqlite3 Modul Extensions laden kann.
import sqlite_vec
def use_sqlite_vec(con) -> None:
con.enable_load_extension(True)
sqlite_vec.load(con)
con.enable_load_extension(False)
con.execute("CREATE VIRTUAL TABLE IF NOT EXISTS vbin USING vec0(embedding bit[768])")
def store_bits(con, item_id: int, v: np.ndarray) -> None:
con.execute("INSERT INTO vbin(rowid, embedding) VALUES (?, vec_bit(?))",
(item_id, np.packbits(v > 0).tobytes()))
def vector_search_vec(con, qv: np.ndarray, k: int = 60, oversample: int = 16):
cand = [r[0] for r in con.execute(
"SELECT rowid FROM vbin WHERE embedding MATCH vec_bit(?) AND k = ?",
(np.packbits(qv > 0).tobytes(), k * oversample))]
return rescore(con, cand, qv, k) if cand else []Listen mit gewichteter Reciprocal Rank Fusion in Python mischen
Keyword Scores und Kosinus Werte teilen keine Skala, also mische Ränge. Reciprocal Rank Fusion, aus Cormack, Clarke und Büttcher (SIGIR 2009), gibt einem Dokument 1 / (k plus Rang) aus jeder Liste, die es enthält, und addiert die Beiträge, mit k = 60. Das Paper nennt 60 fast optimal und die Wahl unkritisch. Die gewichtete Version multipliziert den Beitrag jeder Liste mit einem Gewicht.
Starte mit gleichen Gewichten. Finers getunte Gewichte passen auf eine kleine Menge, darum sind sie keine Defaults zum Kopieren. Eine Handrechnung: ein Item auf Rang 1 in einer Liste mit Gewicht 1,0 und Rang 2 in einer Liste mit Gewicht 0,5 bekommt 1/61 plus 0,5/62 = 0,02446. Der Grundlagen Beitrag hat Finers eigene Zahlen für Keywords allein, Bedeutung allein und hybrid auf 28 Fragen, alle in-sample.
Chunks rollen auf Dateien hoch als bester plus 0,3 mal zweiter plus 0,1 mal dritter, damit eine Datei mit mehreren guten Chunks einen Schub über eine Datei mit einem glücklichen Chunk bekommt. Die Suchfunktion bindet die Teile dann zusammen. Um nur Keywords mit hybrid zu vergleichen, übergib Gewichte mit einer Null für die Liste, die du abschalten willst.
def rrf(ranked: dict[str, list[int]], weights: dict[str, float], k: int = 60) -> dict[int, float]:
"""Gewichtete Reciprocal Rank Fusion. ranked mappt einen Listennamen auf Item Ids, beste zuerst."""
score: dict[int, float] = {}
for name, ids in ranked.items():
w = weights.get(name, 1.0)
for rank, item in enumerate(ids, start=1):
score[item] = score.get(item, 0.0) + w / (k + rank)
return score
WEIGHTS = {"keyword": 1.0, "meaning": 1.0} # erst gleich, später auf eigenen Fragen tunen
def by_file(scores: dict[int, float], item_file: dict[int, int]):
"""Rollt Chunk Scores auf Dateien hoch: bester plus 0,3 mal zweiter plus 0,1 mal dritter."""
per_file: dict[int, list[tuple[float, int]]] = {}
for item, s in scores.items():
per_file.setdefault(item_file[item], []).append((s, item))
out = []
for file_id, hits in per_file.items():
hits.sort(reverse=True)
total = sum(w * s for w, (s, _) in zip((1.0, 0.3, 0.1), hits))
out.append((file_id, total, hits[0][1])) # Datei, Score, bester Chunk
return sorted(out, key=lambda t: -t[1])def search(con, index, query: str, embed, limit: int = 10, weights=WEIGHTS) -> list[dict]:
qv = embed([QUERY + query])[0]
qv = qv / np.linalg.norm(qv)
ranked = {
"keyword": keyword_search(con, query),
"meaning": [i for i, _ in vector_search(con, index, qv)],
}
scores = rrf(ranked, weights)
if not scores:
return []
marks = ",".join("?" * len(scores))
item_file = dict(con.execute(f"SELECT id, file_id FROM items WHERE id IN ({marks})", list(scores)))
paths = dict(con.execute("SELECT id, path FROM files"))
results = []
for file_id, score, best in by_file(scores, item_file)[:limit]:
text = con.execute("SELECT text FROM items WHERE id=?", (best,)).fetchone()[0]
results.append({"path": paths[file_id], "score": score, "text": text})
return resultsNur knappe Fälle neu ranken
Ein Reranker liest die Anfrage und ein Dokument zusammen. Das ist meist genauer als zwei Vektoren zu vergleichen, und langsamer. Qwen3-Reranker-0.6B antwortet ja oder nein und legt die Ja und Nein Logits als Score offen. Ihn bei jeder Anfrage laufen zu lassen verbrennt Zeit bei Anfragen, die nie zweifelhaft waren, darum lässt Finer ihn nur laufen, wenn der Zweite mindestens 0,8 des Ersten erreicht.
Mische per Rang, nicht per Reranker Wahrscheinlichkeit. Finers erster Versuch mischte die rohe Wahrscheinlichkeit in den Score und landete bei 0,88 MRR gegen 0,90 ohne Reranker auf seinen 28 Fragen, weil die Wahrscheinlichkeiten sättigen (ein Kommentar im Code nennt den Grund); die Zahlen stehen im Grundlagen Beitrag. Die aktuelle Regel gibt einem Dokument auf Retrieval Rang i und Reranker Rang j den Score 0,75 / (10 plus i) plus 1 / (10 plus j), nur über die Top 10, jede auf etwa 128 Tokens geschnitten.
Der Modell Block übernimmt das sentence-transformers CrossEncoder Beispiel der Karte, mit eigener Anweisung über Prompts. Scores sind rohe Logit Differenzen, und nur ihre Reihenfolge zählt. Cache Scores per Anfrage und Dokumenttext, damit eine wiederholte Anfrage oder eine Seitenänderung nicht neu scored. Finer hält 2.048 Einträge im Speicher.
def gated_rerank(query, results, score_fn, top=10, gate=0.8, k=10, w_retrieval=0.75):
"""Rankt nur knappe Fälle neu. score_fn(query, texts) liefert eine Zahl pro Text."""
n = min(top, len(results))
if n < 2 or results[1]["score"] < gate * results[0]["score"]:
return results # Retrieval ist eindeutig: Modell überspringen
rr = score_fn(query, [f'{r["path"]}\n{r["text"]}' for r in results[:n]])
rr_rank = {i: r for r, i in enumerate(sorted(range(n), key=lambda i: -rr[i]))}
fused = {i: w_retrieval / (k + i) + 1.0 / (k + rr_rank[i]) for i in range(n)}
order = sorted(range(n), key=lambda i: -fused[i])
return [results[i] for i in order] + results[n:]# An deine Laufzeit anpassen. pip install -U sentence-transformers
from sentence_transformers import CrossEncoder
reranker = CrossEncoder(
"Qwen/Qwen3-Reranker-0.6B",
prompts={"files": "Given a search query over a person's documents, "
"judge whether the document is what they are looking for"},
default_prompt_name="files",
)
def score_fn(query: str, texts: list[str]) -> list[float]:
# Rohe Ja/Nein Logit Differenzen. Nur ihre Reihenfolge zählt.
# Schneiden per Zeichen steht grob für Schneiden bei etwa 128 Tokens.
return reranker.predict([(query, t[:600]) for t in texts]).tolist()Mit MRR und Hit@k auf eigenen Fragen evaluieren
Bevor du irgendwas tunst, schreib etwa 30 echte Fragen aus deinen eigenen Dateien, jede mit einem oder mehreren Pfadfragmenten, die als richtige Antwort zählen. Nutze Fragen, die du gestellt hast, nicht solche, deren Antwort du schon kennst.
Hit@k ist der Anteil der Fragen, deren richtige Datei in den Top k steht. MRR, der mittlere reziproke Rang, mittelt 1 / Rang des ersten richtigen Treffers und zählt 0 ohne Treffer. Ränge 1, 2 und 4 plus ein Fehlschuss geben (1 plus 0,5 plus 0,25 plus 0) / 4 = 0,4375, und report([1, 2, 4, None]) liefert genau diesen Wert als mrr.
Bepunkte Keywords allein, Bedeutung allein und hybrid auf derselben Menge. Bei 30 Fragen ist eine Frage 3,3 Hit@k Punkte, also sind kleine Unterschiede Rauschen. Wenn du auf den Fragen tunst, die du berichtest, ist das Ergebnis in-sample, wie Finers eigene Tabelle. Halte einige Fragen zurück.
GOLD = [ # eigene Fragen; "expect" hält Pfadfragmente, die als richtige Antwort zählen
{"q": "tomato soup recipe", "expect": ["recipes"]},
{"q": "flight booking Lisbon", "expect": ["lisbon", "trip"]},
]
def first_hit(results: list[dict], expect: list[str]):
for rank, r in enumerate(results, start=1):
if any(frag.lower() in r["path"].lower() for frag in expect):
return rank
return None # gar nicht in den Ergebnissen
def evaluate(search_fn, gold, depth: int = 10) -> list:
"""Ein Rang (oder None) pro Frage. search_fn(question) zu Ergebnissen, beste zuerst."""
return [first_hit(search_fn(g["q"])[:depth], g["expect"]) for g in gold]
def report(ranks: list) -> dict:
n = len(ranks)
def hit(k):
return sum(1 for r in ranks if r is not None and r <= k) / n
return {"hit@1": hit(1), "hit@3": hit(3), "hit@10": hit(10),
"mrr": sum(1 / r for r in ranks if r) / n}Wenige Gewichte hinter einer Leitplanke tunen
Tunen auf der eigenen Gold Menge lädt zum Overfitting ein. Finer ändert eine Einstellung nur, wenn der gewichtete MRR um mindestens 0,01 steigt, keine Gold Menge mehr als 0,01 MRR verliert und keine Menge den Rang-1 Treffer irgendeiner Frage verliert. Jeder Lauf landet als eine JSON Zeile im Log, und das Log macht Rollback erst möglich. Das ist eine vereinfachte Version dieser Regel.
run_eval(weights) liefert die Ränge pro Gold Menge, etwa {"hand": ranks}. Eine Favoriten Menge oder generierte Known-Item Fragen hängst du genauso an; Finer wertet Known Items mit halbem Gewicht. Die Suche ist ein kleines Grid, ein Gewicht nach dem anderen, und behält die beste angenommene Änderung. Eine Änderung, die durchkommt, ist Evidenz auf deiner Gold Menge, nicht auf deinen künftigen Anfragen.
import json
import time
def accept(before: dict, after: dict, weights=None, min_gain=0.01, max_loss=0.01):
"""before/after: Gold Mengen Name zu Rängen aus evaluate(). Liefert (ok, Grund)."""
weights = weights or {name: 1.0 for name in before}
gain = sum(weights[n] * (report(after[n])["mrr"] - report(before[n])["mrr"])
for n in before) / sum(weights.values())
for n in before:
loss = report(before[n])["mrr"] - report(after[n])["mrr"]
if loss > max_loss:
return False, f"{n} lost {loss:.3f} MRR"
if any(b == 1 and a != 1 for b, a in zip(before[n], after[n])):
return False, f"{n} lost a rank-1 hit"
return gain >= min_gain, f"weighted MRR {gain:+.3f}"
def tune(run_eval, current: dict, step=0.2, log="quality.jsonl") -> dict:
"""run_eval(weights) zu {Mengenname: Ränge}. Ein Gewicht nach dem anderen probieren, beste angenommene Änderung behalten."""
before, best, best_gain, best_why = run_eval(current), None, 0.0, ""
for name in current:
for delta in (-step, step):
cand = {**current, name: round(max(0.05, current[name] + delta), 2)}
after = run_eval(cand)
ok, why = accept(before, after)
gain = sum(report(after[n])["mrr"] - report(before[n])["mrr"] for n in before) / len(before)
if ok and gain > best_gain:
best, best_gain, best_why = cand, gain, why
entry = {"when": time.strftime("%Y-%m-%d %H:%M:%S"), "previous": current,
"action": "applied" if best else "kept", "candidate": best, "reason": best_why or "no change passed"}
with open(log, "a") as f:
f.write(json.dumps(entry) + "\n")
return best or current
def rollback(log="quality.jsonl"):
"""Gewichte von vor der letzten angewendeten Änderung, oder None."""
entries = [json.loads(line) for line in open(log)]
applied = [e for e in entries if e["action"] == "applied"]
return applied[-1]["previous"] if applied else NoneDeutscher Text in FTS5: Umlaute, ß und Komposita
Das Snippet unten lässt das Schema aus Schritt 2 (unicode61 mit remove_diacritics 2) gegen vier kleine deutsche Dokumente laufen und druckt, was jede Anfrage findet. Auf SQLite 3.53.4 ist die Ausgabe der Kommentarblock an seinem Ende. „Müller“ und „Muller“ finden beide Schreibweisen des Namens. „Mueller“ findet nichts, weil es keine Transliteration von ue zu ü gibt. FTS5 faltet „ß“ nicht, darum findet „Straße“ nur das Dokument mit „Straße“ und „Strasse“ nur das andere.
Finer hat hier einen Bug: sein Abfrage Code faltet ß vor der Abfrage zu ss, aber der Index behält ß, darum verhält sich eine Anfrage nach „Straße“ mit diesem Code wie die „Strasse“ Zeile oben und verpasst Dokumente mit ß. Die Funktion fts_query in Schritt 2 faltet nichts, weil FTS5 Abfrage Terme mit dem Tokenizer der Tabelle selbst tokenisiert. Wenn du willst, dass ss und ß matchen, lege beide Schreibweisen selbst in die Abfrage.
Die eingebauten Tokenizer von FTS5 stemmen kein Deutsch und teilen keine Komposita. In der Ausgabe oben erreicht der Präfix Term „miet“* zwar „Mietvertrag“, aber nicht „Untermietvertrag“, und „Vertrag“ findet keins von beiden. Komposita, die auf dein Wort enden, brauchen einen anderen Weg.
# Schritt 2, Check. Deutsche Schreibweisen gegen das Schema aus Schritt 2. Braucht open_db, add_file und keyword_search.
con = open_db()
add_file(con, "notes/a.txt", [("", "Herr Müller wohnt in der Straße")])
add_file(con, "notes/b.txt", [("", "Herr Muller wohnt in der Strasse")])
add_file(con, "notes/c.txt", [("", "Der Mietvertrag läuft")])
add_file(con, "notes/d.txt", [("", "Der Untermietvertrag läuft")])
def paths_for(query: str) -> list[str]:
ids = keyword_search(con, query)
if not ids:
return []
marks = ",".join("?" * len(ids))
rows = con.execute(
f"SELECT DISTINCT f.path FROM items i JOIN files f ON f.id = i.file_id "
f"WHERE i.id IN ({marks}) ORDER BY f.path", ids)
return [r[0] for r in rows]
for q in ["Müller", "Muller", "Mueller", "Straße", "Strasse", "miet", "Vertrag"]:
print(f"{q:8} -> {paths_for(q)}")
# Ausgabe, auf SQLite 3.53.4:
# Müller -> ['notes/a.txt', 'notes/b.txt']
# Muller -> ['notes/a.txt', 'notes/b.txt']
# Mueller -> []
# Straße -> ['notes/a.txt']
# Strasse -> ['notes/b.txt']
# miet -> ['notes/c.txt']
# vertrag -> []Häufige Fragen
Was ist Reciprocal Rank Fusion (RRF)?
Eine Art, mehrere Ranking Listen zu mischen. Jede Liste gibt einem Dokument 1 / (k plus Rang), mit k = 60 im Original Paper, und die Beiträge summieren sich. Sie ignoriert rohe Scores, darum lassen sich Listen auf verschiedenen Skalen, etwa BM25 und Kosinus Ähnlichkeit, ohne Normierung kombinieren.
Gibt es ein Beispiel für hybride Suche?
Ein Chunk auf Platz 1 der Keyword Liste und Platz 3 der Vektor Liste bekommt bei gleichen Gewichten 1/61 plus 1/63 = 0,0323. Ein Chunk auf Platz 1 nur einer Liste bekommt 1/61 = 0,0164. Ein Chunk, den beide Signale finden, schlägt also einen, den nur eins findet, und genau darum geht es bei der Fusion.
Was ist MRR in RAG, und wie wird er berechnet?
Der mittlere reziproke Rang bewertet einen Retriever auf einer Fragenmenge. Pro Frage nimm 1 / den Rang des ersten richtigen Treffers, oder 0 ohne Treffer, dann mittle. Ränge 1, 2 und ein Fehlschuss geben (1 plus 0,5 plus 0) / 3 = 0,5. Er belohnt, das richtige Dokument nach oben zu stellen; lies ihn neben Hit@k.
Wie baue ich ein lokales RAG?
Diese Anleitung ist die Retrieval Hälfte. Für die Antwort Hälfte nimm die beste Passage aus jedem Top Treffer, nummeriere sie, bitte ein Modell, nur aus ihnen zu antworten und mit [n] zu zitieren, und streiche Zitate ins Leere. Finers optionales Ask kann genau das. Mit Ask an nutzt es per Default ein Cloud Modell, also ist es nur lokal, wenn du auch das Antwort Modell lokal hältst.
Kann SQLite Vektorsuche?
Nicht nativ. Du kannst Vektoren als BLOBs speichern und mit NumPy scannen, wie oben, oder die Erweiterung sqlite-vec laden, die eine vec0 Tabelle und eine Nächste Nachbarn Abfrage mitbringt. Finer nutzt ihre bit[768] Spalte. FTS5 dagegen steckt in den meisten SQLite Builds.
Wie implementiere und evaluiere ich einen Reranker in RAG?
Bewerte die Top 10 mit einem Cross-Encoder, mische seine Reihenfolge per Rang mit der Retrieval Reihenfolge, und lass ihn nur laufen, wenn die zwei besten Scores nah beieinander liegen. Zum Evaluieren lass deine Gold Menge einmal mit und einmal ohne Reranker laufen, vergleiche MRR und Hit@1, und schau, welche Fragen gewandert sind. Auf Finers 28 Fragen hat er zwei Hit@1 Fragen bewegt, auf einer Menge, auf der die Einstellungen getunt waren.
Warum meldet SQLite no such module: fts5?
Dein Python SQLite wurde ohne FTS5 gebaut. Lass den Check in Schritt 0 laufen, dann wechsle zu einem Python Build, dessen SQLite FTS5 enthält.
Quellen
- SQLite FTS5 Dokumentation sqlite.org
- EmbeddingGemma 2 Modellkarte (Hugging Face) huggingface.co
- mlx-community 8-Bit Variante von EmbeddingGemma 2 huggingface.co
- Qwen3-Reranker-0.6B Modellkarte huggingface.co
- Cormack, Clarke, Büttcher: Reciprocal Rank Fusion schlägt Condorcet und einzelne Rank Verfahren (SIGIR 2009) research.google
- sqlite-vec: Vektorsuche als SQLite Erweiterung github.com
- sqlite-vec Anleitung zur binären Quantisierung alexgarcia.xyz
- Finer Quellcode github.com
Sprich mich an
Baust du ein KI-Produkt und brauchst einen Senior für die technische Seite? Buche einen gratis 30-Minuten-Call.