Baue den Index als SQLite mit FTS5 und sqlite-vec, dazu die Suche

Entscheidung Andreas 2026-09-04 (spine/core#8): kein Datenbankdienst. Der
Index ist eine abgeleitete Sicht auf die Markdown-Ablage in einer Datei im
Volume, jederzeit neu baubar. image/km/index.py zerlegt die Dateien in
Chunks (Chunking aus info/ai/scripts/index_docs.py übernommen), hält
Volltext in FTS5 und Vektoren aus bge-m3 in sqlite-vec, inkrementell über
SHA-256 je Datei. image/km/search.py sucht Volltext führend, Vektor
ergänzend, mit Filtern vor dem Ranking (0 Treffer statt gesperrt).

Ingest: Kommentare bei der Erstbefüllung repo-weit statt je Vorgang —
die erste Fassung hing bei live/live nach 371 Vorgängen; jetzt 2175 in zwei
Minuten. Gemessen: Volltext über 61.000 Chunks in 3 s, Vektoren 0,31 s je
Chunk auf der CPU-Box. Kein Bytecode mehr im Repo.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>

image/README.md

@@ -3,9 +3,10 @@
 Hier entsteht `git.42i.org/spine/km`: die API, die Ablage, der MCP-Endpunkt und
 die prüfende Schreibfunktion.
 
-**Erster Baustein vorhanden:** `km/ingest_gitea.py` holt Tickets und Kommentare
-aus Gitea in die Ablage (spine/core#13, Doku in
-[../docs/ingest-gitea.md](../docs/ingest-gitea.md)). Der Rest steht in
+**Vorhanden:** `km/ingest_gitea.py` holt Tickets und Kommentare aus Gitea in
+die Ablage ([../docs/ingest-gitea.md](../docs/ingest-gitea.md)); `km/index.py`
+und `km/search.py` bauen den SQLite-Index (FTS5 + sqlite-vec) und suchen
+darüber ([../docs/index-suche.md](../docs/index-suche.md)). Der Rest steht in
 [../docs/konzept.md](../docs/konzept.md).
 Die Reihenfolge, in der es sinnvoll wächst:
 

image/km/index.py

@@ -0,0 +1,324 @@
+#!/usr/bin/env python3
+"""Index ueber die km-Ablage: SQLite mit FTS5 (Volltext) und sqlite-vec (Vektor).
+
+Der Index ist eine abgeleitete Sicht auf die Markdown-Dateien der Ablage und
+darf jederzeit geloescht und neu gebaut werden -- die Wahrheit liegt in den
+Dateien (spine/core#8, Entscheidung 2026-09-04). Eine Datei im Volume, kein
+Dienst.
+
+Inkrementell ueber den SHA-256 je Datei: unveraenderte Dateien werden nicht
+angefasst, geaenderte komplett neu zerlegt und eingebettet, verschwundene
+entfernt. Embeddings kommen aus Ollama (`bge-m3`, 1024 Dimensionen) wie beim
+Wissens-MCP; ohne `--embed` wird nur der Volltext gebaut, was in Sekunden
+geht -- die Vektoren lassen sich spaeter nachziehen (`--embed --embed-only`).
+
+Chunking (Absaetze, max_chars, Ueberlappung) ist aus
+`info/ai/scripts/index_docs.py` uebernommen, nicht neu erfunden.
+"""
+
+from __future__ import annotations
+
+import argparse
+import hashlib
+import os
+import re
+import sqlite3
+import sys
+import time
+from pathlib import Path
+
+import requests
+import sqlite_vec
+
+EMBED_DIM = 1024
+SCHEMA = """
+CREATE TABLE IF NOT EXISTS docs (
+  path TEXT PRIMARY KEY,
+  namespace TEXT NOT NULL,
+  projekt TEXT,
+  classification TEXT NOT NULL,
+  typ TEXT NOT NULL,
+  zustand TEXT,
+  titel TEXT NOT NULL,
+  aktualisiert TEXT,
+  sha256 TEXT NOT NULL,
+  indexiert TEXT NOT NULL,
+  eingebettet INTEGER NOT NULL DEFAULT 0
+);
+CREATE TABLE IF NOT EXISTS chunks (
+  id INTEGER PRIMARY KEY,
+  path TEXT NOT NULL REFERENCES docs(path) ON DELETE CASCADE,
+  idx INTEGER NOT NULL,
+  text TEXT NOT NULL
+);
+CREATE INDEX IF NOT EXISTS chunks_path ON chunks(path);
+CREATE VIRTUAL TABLE IF NOT EXISTS chunks_fts USING fts5(
+  text, titel UNINDEXED, path UNINDEXED,
+  content='chunks', content_rowid='id', tokenize='unicode61 remove_diacritics 0'
+);
+CREATE TRIGGER IF NOT EXISTS chunks_ai AFTER INSERT ON chunks BEGIN
+  INSERT INTO chunks_fts(rowid, text) VALUES (new.id, new.text);
+END;
+CREATE TRIGGER IF NOT EXISTS chunks_ad AFTER DELETE ON chunks BEGIN
+  INSERT INTO chunks_fts(chunks_fts, rowid, text) VALUES ('delete', old.id, old.text);
+END;
+CREATE TABLE IF NOT EXISTS verweise (
+  von TEXT NOT NULL, nach TEXT NOT NULL, PRIMARY KEY (von, nach)
+);
+CREATE INDEX IF NOT EXISTS verweise_nach ON verweise(nach);
+"""
+SCHEMA_VEC = f"CREATE VIRTUAL TABLE IF NOT EXISTS chunks_vec USING vec0(embedding float[{EMBED_DIM}]);"
+
+
+def connect(db: Path) -> sqlite3.Connection:
+    con = sqlite3.connect(db)
+    con.enable_load_extension(True)
+    sqlite_vec.load(con)
+    con.enable_load_extension(False)
+    con.execute("PRAGMA journal_mode=WAL")
+    con.execute("PRAGMA foreign_keys=ON")
+    con.executescript(SCHEMA)
+    con.execute(SCHEMA_VEC)
+    return con
+
+
+# ---------------------------------------------------------------- Dateien
+
+def parse_frontmatter(text: str) -> tuple[dict[str, str], str]:
+    if not text.startswith("---"):
+        return {}, text
+    lines = text.splitlines()
+    meta: dict[str, str] = {}
+    for end in range(1, len(lines)):
+        if lines[end].strip() == "---":
+            for line in lines[1:end]:
+                if ":" not in line or line.startswith((" ", "-", "#")):
+                    continue
+                k, v = line.split(":", 1)
+                meta[k.strip()] = v.strip().strip("\"'")
+            return meta, "\n".join(lines[end + 1:]).lstrip("\n")
+    return {}, text
+
+
+def split_list(value: str) -> list[str]:
+    value = (value or "").strip().strip("[]")
+    return [p.strip().strip("\"'") for p in re.split(r"[,;]", value) if p.strip()]
+
+
+def derive_title(body: str, fallback: str) -> str:
+    m = re.search(r"^#\s+(.+)$", body, re.MULTILINE)
+    return m.group(1).strip() if m else fallback
+
+
+def doc_fields(rel: str, meta: dict[str, str], body: str, default_ns: str) -> dict:
+    parts = rel.split("/")
+    namespace = meta.get("namespace") or (parts[0] if len(parts) > 1 else default_ns)
+    typ = meta.get("typ") or meta.get("type") or ("vorgang" if meta.get("quelle") == "gitea" else "doku")
+    projekt = meta.get("projekt") or meta.get("project")
+    if not projekt and typ == "vorgang" and meta.get("repo"):
+        projekt = meta["repo"].split("/")[0]
+    return {
+        "namespace": namespace,
+        "projekt": projekt or None,
+        "classification": meta.get("classification") or "lan",
+        "typ": typ,
+        "zustand": meta.get("zustand") or meta.get("state"),
+        "titel": meta.get("titel") or meta.get("title") or derive_title(body, rel),
+        "aktualisiert": meta.get("aktualisiert") or meta.get("updated"),
+        "verweise": split_list(meta.get("verweise", "")),
+    }
+
+
+# ---------------------------------------------------------------- Chunking (aus index_docs.py)
+
+def split_into_paragraphs(text: str) -> list[str]:
+    return [p.strip() for p in re.split(r"\n\s*\n", text) if p.strip()]
+
+
+def split_long_paragraphs(paragraphs: list[str], max_chars: int, overlap: int) -> list[str]:
+    out: list[str] = []
+    step = max(1, max_chars - max(0, overlap))
+    for p in paragraphs:
+        if len(p) <= max_chars:
+            out.append(p)
+            continue
+        start = 0
+        while start < len(p):
+            end = min(len(p), start + max_chars)
+            out.append(p[start:end])
+            if end == len(p):
+                break
+            start += step
+    return out
+
+
+def chunk_paragraphs(paragraphs: list[str], max_chars: int, overlap: int) -> list[str]:
+    chunks: list[str] = []
+    cur: list[str] = []
+    cur_len = 0
+    for p in paragraphs:
+        if cur and cur_len + len(p) + 2 > max_chars:
+            chunks.append("\n\n".join(cur))
+            if overlap > 0:
+                keep: list[str] = []
+                keep_len = 0
+                for q in reversed(cur):
+                    keep.insert(0, q)
+                    keep_len += len(q) + 2
+                    if keep_len >= overlap:
+                        break
+                cur, cur_len = keep, sum(len(q) + 2 for q in keep)
+            else:
+                cur, cur_len = [], 0
+        cur.append(p)
+        cur_len += len(p) + 2
+    if cur:
+        chunks.append("\n\n".join(cur))
+    return chunks
+
+
+def chunk_body(body: str, max_chars: int, overlap: int) -> list[str]:
+    paras = split_long_paragraphs(split_into_paragraphs(body), max_chars, overlap)
+    return chunk_paragraphs(paras, max_chars, overlap)
+
+
+# ---------------------------------------------------------------- Embeddings
+
+class Embedder:
+    def __init__(self, url: str, model: str) -> None:
+        self.url = url.rstrip("/")
+        self.model = model
+        self.session = requests.Session()
+
+    def embed(self, text: str) -> list[float]:
+        return self.embed_many([text])[0]
+
+    def embed_many(self, texts: list[str]) -> list[list[float]]:
+        """Batch ueber /api/embed. Gemessen am 04.09. gegen bge-m3 (CPU):
+        0,31 s je Text bei 32 im Batch, 1,0 s einzeln -- der Rechner ist
+        gebunden, nicht die Leitung; Batch spart Roundtrips, nicht Rechenzeit."""
+        last: Exception | None = None
+        for attempt in range(3):
+            try:
+                r = self.session.post(f"{self.url}/api/embed",
+                                      json={"model": self.model, "input": [t[:600] for t in texts],
+                                            "keep_alive": "30m"},
+                                      timeout=(10, 600))
+                r.raise_for_status()
+                vs = r.json().get("embeddings") or []
+                if len(vs) != len(texts) or any(len(v) != EMBED_DIM for v in vs):
+                    raise RuntimeError(f"embeddings fehlen oder falsche Dimension ({len(vs)}/{len(texts)})")
+                return vs
+            except Exception as e:  # noqa: BLE001
+                last = e
+                time.sleep(2 ** attempt)
+        raise RuntimeError(f"embedding fehlgeschlagen: {last}")
+
+
+# ---------------------------------------------------------------- Lauf
+
+def sha256(data: bytes) -> str:
+    return hashlib.sha256(data).hexdigest()
+
+
+def index_file(con: sqlite3.Connection, ablage: Path, f: Path, default_ns: str,
+               max_chars: int, overlap: int) -> bool:
+    rel = f.relative_to(ablage).as_posix()
+    raw = f.read_bytes()
+    digest = sha256(raw)
+    row = con.execute("SELECT sha256 FROM docs WHERE path=?", (rel,)).fetchone()
+    if row and row[0] == digest:
+        return False
+    meta, body = parse_frontmatter(raw.decode("utf-8", errors="replace"))
+    fields = doc_fields(rel, meta, body, default_ns)
+    now = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())
+    with con:
+        old_ids = [r[0] for r in con.execute("SELECT id FROM chunks WHERE path=?", (rel,))]
+        if old_ids:
+            con.executemany("DELETE FROM chunks_vec WHERE rowid=?", [(i,) for i in old_ids])
+            con.execute("DELETE FROM chunks WHERE path=?", (rel,))
+        con.execute("DELETE FROM verweise WHERE von=?", (rel,))
+        con.execute(
+            "INSERT OR REPLACE INTO docs(path,namespace,projekt,classification,typ,zustand,titel,aktualisiert,sha256,indexiert,eingebettet)"
+            " VALUES (?,?,?,?,?,?,?,?,?,?,0)",
+            (rel, fields["namespace"], fields["projekt"], fields["classification"], fields["typ"],
+             fields["zustand"], fields["titel"], fields["aktualisiert"], digest, now))
+        # Titel als eigener erster Chunk, damit er im Volltext gefunden wird
+        chunks = [fields["titel"]] + chunk_body(body, max_chars, overlap)
+        con.executemany("INSERT INTO chunks(path,idx,text) VALUES (?,?,?)",
+                        [(rel, i, c) for i, c in enumerate(chunks)])
+        con.executemany("INSERT OR IGNORE INTO verweise(von,nach) VALUES (?,?)",
+                        [(rel, v) for v in fields["verweise"]])
+    return True
+
+
+def remove_missing(con: sqlite3.Connection, ablage: Path) -> int:
+    gone = [r[0] for r in con.execute("SELECT path FROM docs") if not (ablage / r[0]).exists()]
+    with con:
+        for rel in gone:
+            ids = [r[0] for r in con.execute("SELECT id FROM chunks WHERE path=?", (rel,))]
+            con.executemany("DELETE FROM chunks_vec WHERE rowid=?", [(i,) for i in ids])
+            con.execute("DELETE FROM docs WHERE path=?", (rel,))
+    return len(gone)
+
+
+def embed_pending(con: sqlite3.Connection, emb: Embedder, limit: int | None) -> int:
+    paths = [r[0] for r in con.execute("SELECT path FROM docs WHERE eingebettet=0 ORDER BY aktualisiert DESC")]
+    if limit:
+        paths = paths[:limit]
+    done = 0
+    t0 = time.time()
+    for n, rel in enumerate(paths, 1):
+        rows = con.execute("SELECT id, text FROM chunks WHERE path=? ORDER BY idx", (rel,)).fetchall()
+        vecs = []
+        for i in range(0, len(rows), 32):
+            batch = rows[i:i + 32]
+            vecs += [(cid, sqlite_vec.serialize_float32(v))
+                     for (cid, _), v in zip(batch, emb.embed_many([t for _, t in batch]))]
+        with con:
+            con.executemany("DELETE FROM chunks_vec WHERE rowid=?", [(cid,) for cid, _ in vecs])
+            con.executemany("INSERT INTO chunks_vec(rowid, embedding) VALUES (?,?)", vecs)
+            con.execute("UPDATE docs SET eingebettet=1 WHERE path=?", (rel,))
+        done += len(vecs)
+        if n % 25 == 0:
+            print(f"  eingebettet: {n}/{len(paths)} dateien, {done} chunks, {time.time() - t0:.0f}s",
+                  file=sys.stderr, flush=True)
+    return done
+
+
+def main() -> None:
+    ap = argparse.ArgumentParser(description=__doc__.split("\n")[0])
+    ap.add_argument("--ablage", required=True, type=Path)
+    ap.add_argument("--db", type=Path, help="Default: <ablage>/../km-index.sqlite")
+    ap.add_argument("--namespace", default=os.environ.get("KM_NAMESPACE", "42i"),
+                    help="Namespace fuer Dateien ohne Angabe im Frontmatter")
+    ap.add_argument("--max-chars", type=int, default=600)
+    ap.add_argument("--overlap-chars", type=int, default=60)
+    ap.add_argument("--embed", action="store_true", help="Vektoren fuer neue/geaenderte Dateien erzeugen")
+    ap.add_argument("--embed-only", action="store_true", help="nur ausstehende Vektoren nachziehen")
+    ap.add_argument("--embed-limit", type=int, help="hoechstens so viele Dateien einbetten")
+    ap.add_argument("--ollama-url", default=os.environ.get("KM_OLLAMA_URL", "http://10.18.5.30:11434"))
+    ap.add_argument("--embed-model", default=os.environ.get("KM_EMBED_MODEL", "bge-m3"))
+    a = ap.parse_args()
+
+    db = a.db or (a.ablage.parent / "km-index.sqlite")
+    con = connect(db)
+    t0 = time.time()
+    if not a.embed_only:
+        files = sorted(p for p in a.ablage.rglob("*.md") if not any(s.startswith(".") for s in p.parts))
+        changed = sum(index_file(con, a.ablage, f, a.namespace, a.max_chars, a.overlap_chars) for f in files)
+        gone = remove_missing(con, a.ablage)
+        n_chunks = con.execute("SELECT count(*) FROM chunks").fetchone()[0]
+        print(f"volltext: {len(files)} dateien, {changed} neu/geaendert, {gone} entfernt, "
+              f"{n_chunks} chunks, {time.time() - t0:.1f}s", file=sys.stderr)
+    if a.embed or a.embed_only:
+        t1 = time.time()
+        n = embed_pending(con, Embedder(a.ollama_url, a.embed_model), a.embed_limit)
+        pending = con.execute("SELECT count(*) FROM docs WHERE eingebettet=0").fetchone()[0]
+        print(f"vektoren: {n} chunks eingebettet, {pending} dateien ausstehend, {time.time() - t1:.0f}s",
+              file=sys.stderr)
+    con.close()
+
+
+if __name__ == "__main__":
+    main()

image/km/ingest_gitea.py

@@ -80,7 +80,7 @@ class Forge:
 
     def get(self, path: str, **params) -> object:
         for attempt in range(4):
-            r = self.session.get(f"{self.url}/api/v1{path}", params=params, timeout=60)
+            r = self.session.get(f"{self.url}/api/v1{path}", params=params, timeout=(10, 60))
             if r.status_code in (429, 502, 503, 504) and attempt < 3:
                 time.sleep(2 ** attempt)
                 continue
@@ -118,6 +118,22 @@ class Forge:
     def comments(self, repo: str, number: int) -> list[dict]:
         return list(self.paged(f"/repos/{repo}/issues/{number}/comments"))
 
+    def all_comments(self, repo: str) -> dict[int, list[dict]]:
+        """Alle Kommentare eines Repos in einem Rutsch, nach Vorgangsnummer.
+
+        Fuer die Erstbefuellung: ein Aufruf je 50 Kommentare statt einer je
+        Vorgang (live/live: ~300 statt ~2200 Aufrufe). Der Endpunkt liefert
+        aelteste zuerst, also bleibt die Reihenfolge je Vorgang erhalten.
+        """
+        by_issue: dict[int, list[dict]] = {}
+        for c in self.paged(f"/repos/{repo}/issues/comments"):
+            url = c.get("issue_url") or ""
+            m = re.search(r"/(?:issues|pulls)/(\d+)$", url)
+            if not m:
+                continue
+            by_issue.setdefault(int(m.group(1)), []).append(c)
+        return by_issue
+
 
 # ---------------------------------------------------------------- Markdown
 
@@ -230,15 +246,27 @@ def ingest_repo(forge: Forge, repo: str, ablage: Path, state: dict, full: bool)
     target = ablage / forge.namespace / "vorgaenge" / repo
     target.mkdir(parents=True, exist_ok=True)
     seen = written = 0
+    # Erstbefuellung: Kommentare repo-weit in einem Rutsch. Inkrementell sind
+    # es wenige geaenderte Vorgaenge, da ist ein Aufruf je Vorgang billiger.
+    bulk = forge.all_comments(repo) if since is None else None
+    t_rep = time.time()
     for kind in ("issues", "pulls"):
         for issue in forge.issues(repo, kind, since):
             seen += 1
-            comments = forge.comments(repo, issue["number"]) if issue.get("comments") else []
+            n = issue["number"]
+            if not issue.get("comments"):
+                comments = []
+            elif bulk is not None:
+                comments = bulk.get(n, [])
+            else:
+                comments = forge.comments(repo, n)
             text = render(forge, repo, issue, comments)
-            f = target / f"{issue['number']}.md"
+            f = target / f"{n}.md"
             if not f.exists() or f.read_text(encoding="utf-8") != text:
                 f.write_text(text, encoding="utf-8")
                 written += 1
+            if seen % 100 == 0:
+                print(f"  {repo}: {seen} vorgaenge, {time.time() - t_rep:.0f}s", file=sys.stderr, flush=True)
     # Stand erst nach vollstaendigem Durchlauf setzen -- bricht der Lauf ab,
     # holt der naechste dieselben Vorgaenge noch einmal (Wiederholen ist billig,
     # Luecken sind es nicht).

image/km/search.py

@@ -0,0 +1,137 @@
+#!/usr/bin/env python3
+"""Suche ueber den km-Index: Volltext fuehrend, Vektor ergaenzend (spine/core#1).
+
+    search.py "worktree prune"                 # hybrid
+    search.py --mode fts "live/live#2181"      # nur Volltext (exakte Treffer)
+    search.py --mode vec "warum bus ohne jetstream"
+    search.py --ns 42i --typ vorgang --zustand open "i18n"
+
+Filter greifen VOR der Suche als WHERE-Klausel -- wer eine Klassifikation
+nicht sehen darf, bekommt 0 Treffer, nicht "vorhanden, aber gesperrt"
+(spine/core#18). Hybrid = Reciprocal Rank Fusion ueber beide Ranglisten.
+"""
+
+from __future__ import annotations
+
+import argparse
+import json
+import os
+import re
+import sqlite3
+import sys
+from pathlib import Path
+
+import sqlite_vec
+
+from index import Embedder, connect  # noqa: E402  (gleicher Ordner)
+
+
+def fts_query(q: str) -> str:
+    """Nutzertext -> FTS5-Ausdruck: jedes Wort als Praefix, Sonderzeichen in Anfuehrungszeichen."""
+    terms = []
+    for tok in re.findall(r"\S+", q):
+        tok = tok.replace('"', '""')
+        terms.append(f'"{tok}"' if not tok.isalnum() else f'"{tok}"*')
+    return " ".join(terms)
+
+
+def where(a, params: list) -> str:
+    conds = []
+    if a.ns:
+        conds.append("d.namespace = ?"); params.append(a.ns)
+    if a.projekt:
+        conds.append("d.projekt = ?"); params.append(a.projekt)
+    if a.typ:
+        conds.append("d.typ = ?"); params.append(a.typ)
+    if a.zustand:
+        conds.append("d.zustand = ?"); params.append(a.zustand)
+    if a.classification:
+        ph = ",".join("?" * len(a.classification))
+        conds.append(f"d.classification IN ({ph})"); params.extend(a.classification)
+    return (" AND " + " AND ".join(conds)) if conds else ""
+
+
+def search_fts(con: sqlite3.Connection, a, k: int) -> list[tuple[int, float]]:
+    params: list = [fts_query(a.query)]
+    w = where(a, params)
+    params.append(k)
+    rows = con.execute(f"""
+        SELECT c.id, bm25(chunks_fts) AS score
+        FROM chunks_fts JOIN chunks c ON c.id = chunks_fts.rowid JOIN docs d ON d.path = c.path
+        WHERE chunks_fts MATCH ? {w}
+        ORDER BY score LIMIT ?""", params).fetchall()
+    return [(r[0], r[1]) for r in rows]
+
+
+def search_vec(con: sqlite3.Connection, a, k: int, emb: Embedder) -> list[tuple[int, float]]:
+    q = sqlite_vec.serialize_float32(emb.embed(a.query))
+    # vec0 kann nicht joinen -- erst k*4 Nachbarn holen, dann filtern
+    rows = con.execute("SELECT rowid, distance FROM chunks_vec WHERE embedding MATCH ? ORDER BY distance LIMIT ?",
+                       (q, k * 4)).fetchall()
+    if not rows:
+        return []
+    params: list = [r[0] for r in rows]
+    w = where(a, params)
+    ph = ",".join("?" * len(rows))
+    ok = {r[0] for r in con.execute(f"SELECT c.id FROM chunks c JOIN docs d ON d.path=c.path WHERE c.id IN ({ph}) {w}", params)}
+    return [(cid, dist) for cid, dist in rows if cid in ok][:k]
+
+
+def rrf(*lists: list[tuple[int, float]], k: int = 60) -> dict[int, float]:
+    scores: dict[int, float] = {}
+    for lst in lists:
+        for rank, (cid, _) in enumerate(lst, 1):
+            scores[cid] = scores.get(cid, 0.0) + 1.0 / (k + rank)
+    return scores
+
+
+def main() -> None:
+    ap = argparse.ArgumentParser(description=__doc__.split("\n")[0])
+    ap.add_argument("query")
+    ap.add_argument("--db", type=Path, default=Path(os.environ.get("KM_INDEX", "km-index.sqlite")))
+    ap.add_argument("--mode", choices=["hybrid", "fts", "vec"], default="hybrid")
+    ap.add_argument("--limit", type=int, default=8)
+    ap.add_argument("--ns"); ap.add_argument("--projekt"); ap.add_argument("--typ"); ap.add_argument("--zustand")
+    ap.add_argument("--classification", action="append", help="erlaubte Klassifikationen (Sichtkreis), mehrfach")
+    ap.add_argument("--ollama-url", default=os.environ.get("KM_OLLAMA_URL", "http://10.18.5.30:11434"))
+    ap.add_argument("--embed-model", default=os.environ.get("KM_EMBED_MODEL", "bge-m3"))
+    ap.add_argument("--json", action="store_true")
+    a = ap.parse_args()
+
+    con = connect(a.db)
+    k = a.limit * 3
+    fts = search_fts(con, a, k) if a.mode in ("hybrid", "fts") else []
+    vec = search_vec(con, a, k, Embedder(a.ollama_url, a.embed_model)) if a.mode in ("hybrid", "vec") else []
+    if a.mode == "fts":
+        ranked = [(cid, -s) for cid, s in fts]
+    elif a.mode == "vec":
+        ranked = [(cid, -d) for cid, d in vec]
+    else:
+        ranked = sorted(rrf(fts, vec).items(), key=lambda x: -x[1])
+
+    # ein Treffer je Dokument, bester Chunk zaehlt
+    out, seen = [], set()
+    for cid, score in ranked:
+        row = con.execute("SELECT c.path, c.idx, c.text, d.titel, d.typ, d.zustand, d.aktualisiert "
+                          "FROM chunks c JOIN docs d ON d.path=c.path WHERE c.id=?", (cid,)).fetchone()
+        if not row or row[0] in seen:
+            continue
+        seen.add(row[0])
+        snippet = re.sub(r"\s+", " ", row[2])[:240]
+        out.append({"path": row[0], "titel": row[3], "typ": row[4], "zustand": row[5],
+                    "aktualisiert": row[6], "score": round(score, 4), "chunk": row[1], "snippet": snippet})
+        if len(out) >= a.limit:
+            break
+
+    if a.json:
+        print(json.dumps(out, ensure_ascii=False, indent=1))
+    else:
+        for r in out:
+            print(f"{r['score']:>8}  {r['path']}  [{r['typ']}/{r['zustand'] or '-'}]  {r['titel']}\n"
+                  f"          {r['snippet']}")
+        if not out:
+            print("0 Treffer", file=sys.stderr)
+
+
+if __name__ == "__main__":
+    main()

image/requirements.txt

@@ -1 +1,2 @@
 requests>=2.31
+sqlite-vec>=0.1.6