Stufe 1: komplexe Fragen via Query-Planer, Per-Query-Slots, Scope-Filter (D12, M6)
- Query-Planer (agent/query_planner.py): Heuristik-Gate (Jahr, Vergleichs-/ Aggregationsmarker, Laenge) entscheidet ueber einen kleinen LLM-Call, der komplexe Fragen in 1-3 Sub-Queries zerlegt (JSON, temp 0; Fehler -> Original als Einzel-Query). Je Sub-Query optional stand_year und scope. - Multi-Query-Retrieval: je Sub-Query BM25+Dense mit RRF-Summe; Per-Query-Slots (2 je Sub-Query) sichern jeden Frageaspekt im Kontext (sonst dominieren Eintraege, die in mehreren Sub-Queries mittelgut matchen — q-113-Befund). Scope-Filter: 'gesetz' (nur Lexis/WIKU/RIS) und 'kv' (nur Branchen-KV) mit Fallback auf unscoped bei leerem Ergebnis. Temporal-Intent: stand_year + temporal_boost (default 0, FTS-jahr-Tag-Signal reichte). - Grounding unveraendert: eine Retrieved-Menge (Union), eine Antwort, Post-Validierung ueber die Union, Verweigerungspflicht unveraendert. - Goldset 42 -> 46: q-110 (Temporal 2023; 2024er-Lohnordnung existiert im Korpus nicht - Mantelvertrag ohne Lohntabelle, korrekt verweigert), q-111 (2025), q-112 (Abfertigung-Vergleich), q-113 (Gesetz+KV). - Eval (46 Fragen): Zitier-Praezision 97,8 %, Verweigerung 97,8 % (Gate >94,3 % erfuellt), erwartete Quelle 90,2 %, Latenz mean 33,5 s. - Tests 50 -> 57. Reports lokal: data/eval-qwen38-stage1*.json. API-first festgehalten (D13-Vorbereitung): Odoo bleibt duenner Client; Lohndaten-Zugriff in M4 erfordert Privacy-Neubewertung.
This commit is contained in:
+159
-18
@@ -13,6 +13,7 @@ import numpy as np
|
||||
from .config import Config
|
||||
from .normalize import fts_query
|
||||
from .ollama_client import OllamaClient
|
||||
from .query_planner import SubQuery
|
||||
|
||||
|
||||
@dataclass
|
||||
@@ -71,6 +72,7 @@ class Retriever:
|
||||
self._owns_client = client is None
|
||||
self._mat: np.ndarray | None = None
|
||||
self._mat_chunk_ids: list[int] | None = None
|
||||
self._scope_cache: dict[str, set[int]] = {}
|
||||
row = self._con.execute("SELECT MIN(stand), MAX(stand) FROM chunks").fetchone()
|
||||
self._stand_min = int((row[0] or "2026-01").replace("-", ""))
|
||||
self._stand_max = int((row[1] or "2026-01").replace("-", ""))
|
||||
@@ -104,19 +106,38 @@ class Retriever:
|
||||
|
||||
# -- Einzelverfahren ----------------------------------------------------
|
||||
|
||||
def _bm25(self, question: str, limit: int) -> dict[int, float]:
|
||||
def _scope_ids(self, scope: str | None) -> set[int] | None:
|
||||
"""Chunk-ID-Menge je Sub-Query-Scope (Cache je Instanz): 'gesetz' =
|
||||
alles außer Branchen-KV, 'kv' = nur Branchen-KV."""
|
||||
if scope is None:
|
||||
return None
|
||||
if scope not in self._scope_cache:
|
||||
if scope == "kv":
|
||||
sql = "SELECT chunk_id FROM chunks WHERE entry_id LIKE 'kv-%'"
|
||||
else: # gesetz
|
||||
sql = "SELECT chunk_id FROM chunks WHERE entry_id NOT LIKE 'kv-%'"
|
||||
self._scope_cache[scope] = {
|
||||
row[0] for row in self._con.execute(sql).fetchall()
|
||||
}
|
||||
return self._scope_cache[scope]
|
||||
|
||||
def _bm25(self, question: str, limit: int, scope: str | None = None) -> dict[int, float]:
|
||||
q = fts_query(question)
|
||||
if not q:
|
||||
return {}
|
||||
fetch = limit * 4 if scope else limit
|
||||
rows = self._con.execute(
|
||||
"SELECT rowid, bm25(chunks_fts) AS rank FROM chunks_fts "
|
||||
"WHERE chunks_fts MATCH ? ORDER BY rank LIMIT ?",
|
||||
(q, limit),
|
||||
(q, fetch),
|
||||
).fetchall()
|
||||
scope_ids = self._scope_ids(scope)
|
||||
if scope_ids is not None:
|
||||
rows = [r for r in rows if r["rowid"] in scope_ids][:limit]
|
||||
# bm25(): kleinere Werte = besser -> negieren für "größer = besser"
|
||||
return {r["rowid"]: -float(r["rank"]) for r in rows}
|
||||
|
||||
def _dense(self, question: str, limit: int) -> dict[int, float]:
|
||||
def _dense(self, question: str, limit: int, scope: str | None = None) -> dict[int, float]:
|
||||
if self.cfg.embed_off:
|
||||
return {}
|
||||
self._ensure_matrix()
|
||||
@@ -139,8 +160,17 @@ class Retriever:
|
||||
if qn == 0:
|
||||
return {}
|
||||
sims = self._mat_norm @ (qvec / qn)
|
||||
order = np.argsort(-sims)[:limit]
|
||||
return {self._mat_chunk_ids[i]: float(sims[i]) for i in order}
|
||||
order = np.argsort(-sims)
|
||||
scope_ids = self._scope_ids(scope)
|
||||
out: dict[int, float] = {}
|
||||
for i in order:
|
||||
if len(out) >= limit:
|
||||
break
|
||||
cid = self._mat_chunk_ids[i]
|
||||
if scope_ids is not None and cid not in scope_ids:
|
||||
continue
|
||||
out[cid] = float(sims[i])
|
||||
return out
|
||||
|
||||
def _ensure_matrix(self) -> None:
|
||||
if self._mat is not None:
|
||||
@@ -240,46 +270,102 @@ class Retriever:
|
||||
|
||||
# -- öffentliche Suche --------------------------------------------------
|
||||
|
||||
def search(self, question: str, n_entries: int | None = None) -> list[ChunkResult]:
|
||||
"""Liefert die Top-Kontextblöcke (Hauptretrieval + cross_ref-Erweiterung)."""
|
||||
n = n_entries or self.cfg.context_blocks
|
||||
pool = self.cfg.candidate_pool
|
||||
bm = self._bm25(question, pool)
|
||||
def _search_one(
|
||||
self, sq: SubQuery, pool: int
|
||||
) -> tuple[dict[int, float], dict[int, float], dict[int, float]]:
|
||||
"""BM25+Dense für EINE Sub-Query inkl. Scope-Filter (mit Fallback auf
|
||||
unscoped, wenn die gefilterte Suche leer bleibt) und Query-RRF."""
|
||||
try:
|
||||
dn = self._dense(question, pool)
|
||||
bm = self._bm25(sq.text, pool, scope=sq.scope)
|
||||
except Exception:
|
||||
bm = {}
|
||||
try:
|
||||
dn = self._dense(sq.text, pool, scope=sq.scope)
|
||||
except Exception:
|
||||
dn = {}
|
||||
fused: dict[int, float] = {}
|
||||
if sq.scope and not bm and not dn:
|
||||
try:
|
||||
bm = self._bm25(sq.text, pool)
|
||||
dn = self._dense(sq.text, pool)
|
||||
except Exception:
|
||||
pass
|
||||
fused_q: dict[int, float] = {}
|
||||
for ranking, weight in ((bm, 1.0), (dn, self.cfg.dense_weight)):
|
||||
ordered = sorted(ranking.items(), key=lambda kv: -kv[1])
|
||||
for rank, (cid, _) in enumerate(ordered):
|
||||
fused[cid] = fused.get(cid, 0.0) + weight / (self.cfg.rrf_k + rank)
|
||||
fused_q[cid] = fused_q.get(cid, 0.0) + weight / (
|
||||
self.cfg.rrf_k + rank
|
||||
)
|
||||
return bm, dn, fused_q
|
||||
|
||||
def _fuse_queries(
|
||||
self, sub_queries: list, pool: int
|
||||
) -> tuple[dict[int, float], set[int], set[int]]:
|
||||
"""RRF-Fusion über BM25+Dense je Sub-Query (Beiträge summieren);
|
||||
liefert (fused, bm25-Chunk-IDs, dense-Chunk-IDs) für die Quelle-
|
||||
Markierung. Bei einer einzelnen Sub-Query identisch zum bisherigen
|
||||
Verhalten."""
|
||||
fused: dict[int, float] = {}
|
||||
bm_all: set[int] = set()
|
||||
dn_all: set[int] = set()
|
||||
for sq in sub_queries:
|
||||
bm, dn, fused_q = self._search_one(sq, pool)
|
||||
bm_all.update(bm)
|
||||
dn_all.update(dn)
|
||||
for cid, score in fused_q.items():
|
||||
fused[cid] = fused.get(cid, 0.0) + score
|
||||
return fused, bm_all, dn_all
|
||||
|
||||
def _context_from_fused(
|
||||
self,
|
||||
fused: dict[int, float],
|
||||
n: int,
|
||||
bm_all: set[int],
|
||||
dn_all: set[int],
|
||||
stand_years: set[str] | None = None,
|
||||
reserved: list[str] | None = None,
|
||||
) -> list[ChunkResult]:
|
||||
"""Fusion -> Score (+Recency +Temporal-Boost) -> Entry-Dedup ->
|
||||
Vertreter-Chunk -> cross_ref-Erweiterung."""
|
||||
if not fused:
|
||||
return []
|
||||
|
||||
rows = self._fetch_chunks(list(fused))
|
||||
results: list[ChunkResult] = []
|
||||
for cid, score in fused.items():
|
||||
if cid not in rows:
|
||||
continue
|
||||
source = "fused" if (cid in bm and cid in dn) else (
|
||||
"bm25" if cid in bm else "dense"
|
||||
source = (
|
||||
"fused" if (cid in bm_all and cid in dn_all)
|
||||
else ("bm25" if cid in bm_all else "dense")
|
||||
)
|
||||
score += self.cfg.recency_boost * self._stand_factor(rows[cid]["stand"])
|
||||
if stand_years and rows[cid]["entry_id"].startswith("kv-"):
|
||||
if rows[cid]["stand"][:4] in stand_years:
|
||||
score += self.cfg.temporal_boost
|
||||
results.append(self._row_to_result(rows[cid], score, source))
|
||||
results.sort(key=lambda r: -r.score)
|
||||
|
||||
# Bester Chunk je Eintrag -> Kontext (Entry-Level-Dedup).
|
||||
# Der Vertreter-Chunk ist die beste Inhaltssektion des Eintrags,
|
||||
# nicht die Rangfolge-Beste (vgl. _section_priority).
|
||||
# Per-Query-Slots (Multi-Query): reservierte Einträge zuerst —
|
||||
# jeder Sub-Query-Aspekt hält garantiert seine Top-Quelle(n).
|
||||
main: list[ChunkResult] = []
|
||||
seen: set[str] = set()
|
||||
by_entry: dict[str, list[ChunkResult]] = {}
|
||||
for r in results:
|
||||
by_entry.setdefault(r.entry_id, []).append(r)
|
||||
seen: set[str] = set()
|
||||
reserved = reserved or []
|
||||
for entry_id in reserved:
|
||||
if len(main) >= n or entry_id not in by_entry:
|
||||
continue
|
||||
seen.add(entry_id)
|
||||
main.append(self._representative_chunk(entry_id, by_entry[entry_id]))
|
||||
for entry_id, chunks in by_entry.items():
|
||||
if len(main) >= n:
|
||||
break
|
||||
if entry_id in seen:
|
||||
continue
|
||||
seen.add(entry_id)
|
||||
main.append(self._representative_chunk(entry_id, chunks))
|
||||
|
||||
@@ -297,4 +383,59 @@ class Retriever:
|
||||
extra.append(er)
|
||||
seen.add(ref)
|
||||
budget -= 1
|
||||
return main + extra
|
||||
return main + extra
|
||||
|
||||
def search(self, question: str, n_entries: int | None = None) -> list[ChunkResult]:
|
||||
"""Liefert die Top-Kontextblöcke (Hauptretrieval + cross_ref-Erweiterung)."""
|
||||
n = n_entries or self.cfg.context_blocks
|
||||
fused, bm_all, dn_all = self._fuse_queries(
|
||||
[SubQuery(text=question)], self.cfg.candidate_pool
|
||||
)
|
||||
return self._context_from_fused(fused, n, bm_all, dn_all)
|
||||
|
||||
def search_multi(
|
||||
self, sub_queries: list, n_entries: int | None = None
|
||||
) -> list[ChunkResult]:
|
||||
"""Multi-Query-Retrieval (Stufe 1): je Sub-Query BM25+Dense, RRF-
|
||||
Beiträge summieren. Per-Query-Slots: jede Sub-Query sichert ihre
|
||||
Top-Einträge im Kontext (Multi-Hop: jeder Frageaspekt kommt mit
|
||||
seiner besten Quelle hinein — sonst dominieren Einträge, die in
|
||||
mehreren Sub-Queries mittelgut matchen). Temporal-Intent: kv-
|
||||
Einträge im gefragten Geltungsjahr erhalten temporal_boost."""
|
||||
n = n_entries or self.cfg.context_blocks
|
||||
if len(sub_queries) == 1:
|
||||
return self.search(sub_queries[0].text, n_entries=n)
|
||||
pool = self.cfg.candidate_pool
|
||||
fused_total: dict[int, float] = {}
|
||||
bm_all: set[int] = set()
|
||||
dn_all: set[int] = set()
|
||||
per_query_entries: list[list[str]] = []
|
||||
for sq in sub_queries:
|
||||
bm, dn, fused_q = self._search_one(sq, pool)
|
||||
bm_all.update(bm)
|
||||
dn_all.update(dn)
|
||||
for cid, score in fused_q.items():
|
||||
fused_total[cid] = fused_total.get(cid, 0.0) + score
|
||||
rows_q = self._fetch_chunks(list(fused_q))
|
||||
entry_best: dict[str, float] = {}
|
||||
for cid, score in fused_q.items():
|
||||
row = rows_q.get(cid)
|
||||
if row is not None:
|
||||
eid = row["entry_id"]
|
||||
entry_best[eid] = max(entry_best.get(eid, 0.0), score)
|
||||
per_query_entries.append(
|
||||
[e for e, _ in sorted(entry_best.items(), key=lambda kv: -kv[1])]
|
||||
)
|
||||
reserved: list[str] = []
|
||||
for entries in per_query_entries:
|
||||
taken = 0
|
||||
for eid in entries:
|
||||
if taken >= self.cfg.per_query_slots:
|
||||
break
|
||||
if eid not in reserved:
|
||||
reserved.append(eid)
|
||||
taken += 1
|
||||
years = {sq.stand_year for sq in sub_queries if sq.stand_year}
|
||||
return self._context_from_fused(
|
||||
fused_total, n, bm_all, dn_all, stand_years=years, reserved=reserved
|
||||
)
|
||||
Reference in New Issue
Block a user