Fused retrieval from directive and løn

README.md

@@ -12,4 +12,87 @@ license: apache-2.0
 short_description: Multilingual AI assistant to guarantee fair pay
 ---
 
-Check out the configuration reference at https://huggingface.co/docs/hub/spaces-config-reference
+# Pay Equity for EU
+
+Multilingual Gradio Space for Danish users: **expected pay** (IDA lønstatistik)
+and **pay-transparency rights** (EU Directive 2023/970), grounded with citations.
+
+## Quick start
+
+```bash
+uv sync
+uv run python main.py
+```
+
+Requires a built vector index at `data/processed/index/` (see below). Backend
+details: [`src/backend/README.md`](src/backend/README.md).
+
+## Build the vector index
+
+Embed the corpus once (directive articles + IDA salary rows), then commit the
+output:
+
+```bash
+uv run python scripts/build_index.py   # GPU / ZeroGPU Space admin button
+```
+
+Produces `data/processed/index/index.faiss` + `chunks.jsonl`.
+
+### Refresh metadata only (no re-embed)
+
+After changing `Chunk.metadata` mapping in `src/backend/indexing/ingest.py`:
+
+```bash
+uv run python scripts/refresh_metadata.py
+```
+
+Rewrites `chunks.jsonl` only; vectors stay unchanged. Re-embed if chunk ids or
+order changed.
+
+## Run the RAG benchmark
+
+Offline evaluation of retrieval (and lightweight generation checks) against
+15 ground-truth questions in `scripts/eval/benchmark_questions.json`.
+
+**Prerequisites:** committed `chunks.jsonl` plus a local `index.faiss`. If
+`index.faiss` is missing or a Git LFS pointer, rebuild it from the committed
+chunks (CPU-only, ~few minutes):
+
+```bash
+uv run python scripts/eval/build_local_index.py
+```
+
+**Run the benchmark:**
+
+```bash
+# Full run: retrieval metrics + generation fact/citation checks (default k=5)
+uv run python scripts/eval/run_benchmark.py
+
+# Retrieval only (faster)
+uv run python scripts/eval/run_benchmark.py --no-generation
+
+# Custom top-k
+uv run python scripts/eval/run_benchmark.py --k 10
+```
+
+**Output:** human-readable summary on stdout + JSON report at
+`scripts/eval/results/benchmark_YYYYMMDD_HHMMSS.json`.
+
+**Metrics:** Recall@k, MRR, NDCG@k (retrieval); fact coverage and citation
+presence (generation). See [`scripts/eval/BENCHMARK_FINDINGS.md`](scripts/eval/BENCHMARK_FINDINGS.md)
+for the latest scored run and failure analysis.
+
+> **Note:** The benchmark script uses a single global FAISS top-k search, not
+> the production per-source + RRF fusion in `rag.py`. Use it to track embedding
+> quality; for end-to-end production behaviour, test via the Gradio UI or
+> `/chat` API.
+
+## Data
+
+| Path | Contents |
+|---|---|
+| `data/processed/directive/` | Parsed EU Directive 2023/970 |
+| `data/processed/lonstatistik/` | IDA + Djøf salary statistics ([README](data/processed/lonstatistik/README.md)) |
+| `data/processed/index/` | FAISS index + aligned `chunks.jsonl` |
+
+Raw PDFs live under `data/raw/` (gitignored).

scripts/refresh_metadata.py

@@ -0,0 +1,75 @@
+"""Regenerate chunks.jsonl metadata without re-embedding (CPU-only, offline).
+
+The dual-source retrieval work enriches each ``Chunk`` with structured
+``metadata`` (IDA: sector/category/experience/measures; directive: article
+title). That metadata is **not embedded**, so the vectors and chunk order are
+unchanged — meaning the committed ``index.faiss`` stays byte-identical and we
+only need to rewrite ``chunks.jsonl``.
+
+This script loads the existing index + old chunks, rebuilds the corpus with the
+new metadata, asserts the rebuilt chunks line up 1:1 with the live vectors
+(same ids, same order, same count), then re-persists. No GPU, no embedding —
+runs in well under a second.
+
+Usage::
+
+    uv run python scripts/refresh_metadata.py
+"""
+
+from __future__ import annotations
+
+import sys
+from pathlib import Path
+
+# Allow running as a plain script (`python scripts/refresh_metadata.py`).
+sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
+
+from src.backend.indexing.ingest import (  # noqa: E402
+    build_corpus,
+    load_directive_chunks,
+    load_ida_chunks,
+)
+from src.backend.indexing.store import DEFAULT_INDEX_DIR, VectorStore  # noqa: E402
+
+
+def main() -> None:
+    store = VectorStore()
+    if not store.load():
+        raise SystemExit(
+            f"No index at {DEFAULT_INDEX_DIR}. Build it first (scripts/build_index.py "
+            "or scripts/build_index_modal.py) before refreshing metadata."
+        )
+
+    old_ids = [c.id for c in store._chunks]
+    ntotal = store._index.ntotal
+
+    n_dir = len(load_directive_chunks())
+    n_ida = len(load_ida_chunks())
+    new_chunks = build_corpus()
+    new_ids = [c.id for c in new_chunks]
+
+    # Alignment guard: the rebuilt chunks must line up exactly with the stored
+    # vectors, else chunks.jsonl would point at the wrong embeddings.
+    if len(new_chunks) != ntotal:
+        raise SystemExit(
+            f"Count mismatch: rebuilt {len(new_chunks)} chunks but index has "
+            f"{ntotal} vectors. Re-embed (build_index) instead of refreshing."
+        )
+    if new_ids != old_ids:
+        raise SystemExit(
+            "Order/id mismatch between rebuilt chunks and the existing index. "
+            "The corpus changed shape — re-embed (build_index) instead."
+        )
+
+    store._chunks = new_chunks
+    store.persist()  # rewrites chunks.jsonl (+ identical index.faiss)
+
+    n_meta = sum(1 for c in new_chunks if c.metadata)
+    print(
+        f"refreshed: directive={n_dir} ida={n_ida} total={len(new_chunks)} "
+        f"ntotal={ntotal} with_metadata={n_meta} → {DEFAULT_INDEX_DIR}"
+    )
+
+
+if __name__ == "__main__":
+    main()

src/backend/README.md

@@ -14,33 +14,38 @@ From the repo root:
 uv run python main.py
 ```
 
-`main.py` calls `build_server().launch()`. The server prints a local URL
-(default `http://127.0.0.1:7860`). Open it for the chat UI; the API lives under
-the same origin.
+`main.py` builds `demo = build_blocks()` and calls `demo.launch(...)`. The
+server prints a local URL (default `http://127.0.0.1:7860`). Open it for the
+chat UI.
 
 Before first run, build the vector index (see **RAG pipeline** below).
 
 ## RAG pipeline
 
 ```
-query ─► embed (Nemotron) ─► FAISS top-k ─► grounded prompt ─► Tiny Aya ─► cited answer
+query ─► embed (Nemotron) ─► per-source top-k ─► RRF fuse ─► grounded prompt ─► Tiny Aya ─► cited answer
 ```
 
 - **Ingestion** (`indexing/ingest.py`): the Directive `document.json` is grouped
-  into one `Chunk` per article (`Art. N` citations); IDA rows
-  (`union == "IDA"`) become chunks from their ready-made `rag_text` (cited by
-  source page). Djøf is excluded for now.
+  into one `Chunk` per article (`Art. N` citations + `metadata.article_number` /
+  `article_title`); IDA rows (`union == "IDA"`) become chunks from their
+  ready-made `rag_text` (cited by source page) with structured salary
+  `metadata` (sector, category, experience, measures, …). Djøf is excluded for
+  now. Only `Chunk.text` is embedded; `metadata` is persisted in `chunks.jsonl`
+  for downstream use.
 - **Offline index build** (`scripts/build_index.py`): embeds the whole corpus
   on ZeroGPU (the embedding call is wrapped in `@spaces.GPU`) and persists
-  `data/processed/index/{index.faiss, chunks.jsonl}`. Run it once on the ZeroGPU
-  Space (or any GPU machine) and **commit the output** — so the live app only
+  `data/processed/index/{index.faiss, chunks.jsonl}`. Run once on the ZeroGPU
+  Space (or any GPU machine) and **commit the output** — the live app only
   *loads* the index and never re-embeds the corpus at runtime.
+- **Metadata refresh** (`scripts/refresh_metadata.py`): rewrites `chunks.jsonl`
+  from updated ingest logic **without** re-embedding. Safe only when chunk ids and
+  order are unchanged; otherwise re-run `build_index.py`.
 - **Runtime** (`rag.py`): `answer_stream()` is the single grounding path shared
-  by the UI and (via `answer()`) the `/chat` API. Query embedding + retrieval +
-  **streamed** generation run inside one `@spaces.GPU` function
-  (`_retrieve_and_stream`) — a single ZeroGPU allocation per turn. Token
-  streaming keeps the generation experience smooth; the structured `/chat` API
-  consumes the same stream and returns the final response.
+  by the UI and (via `answer()`) the `/chat` API. Retrieval runs **per source**
+  (`directive`, `lonstatistik`) — top-5 each — then **Reciprocal Rank Fusion**
+  into the final context (default 6 chunks). Query embedding + generation run
+  inside one `@spaces.GPU` function (`_retrieve_and_stream`).
 
 Models are pinned by revision in `configs/models.yaml`. The LLM backend is
 selected by `llm.<role>.backend` (`tiny_aya` | `minicpm` | `llama_cpp` |
@@ -87,5 +92,7 @@ indexing/embedder.py Embedder (config-driven, sentence-transformers | transforme
 indexing/store.py    VectorStore (FAISS IndexFlatIP + persist/load/search)
 parsing/             directive.py (→ ingest), lonstatistik.py (placeholder)
 api/endpoints.py     register(app): the @app.api endpoints
-scripts/build_index.py   offline corpus embedding → data/processed/index/
+scripts/build_index.py       offline corpus embedding → data/processed/index/
+scripts/refresh_metadata.py  rewrite chunks.jsonl metadata without re-embedding
+scripts/eval/run_benchmark.py  offline retrieval benchmark (see root README.md)
 ```

src/backend/indexing/ingest.py

@@ -82,6 +82,10 @@ def load_directive_chunks(
                 article=f"Art. {cur_num}",
                 page=cur_page,
                 citation=f"Art. {cur_num}",
+                metadata={
+                    "article_number": cur_num,
+                    "article_title": cur_title,
+                },
             )
         )
         cur_body = []
@@ -134,6 +138,20 @@ def load_ida_chunks(path: str | Path = SALARY_JSONL) -> list[Chunk]:
                 source="lonstatistik",
                 page=page,
                 citation=f"{rec['source_doc']}, p.{page}",
+                metadata={
+                    "sector": rec.get("sector"),
+                    "category": rec.get("table_title"),
+                    "row_label": rec.get("row_label"),
+                    "segment_label": rec.get("segment_label"),
+                    "experience_years_min": rec.get("experience_years_min"),
+                    "experience_years_max": rec.get("experience_years_max"),
+                    "graduation_year_start": rec.get("graduation_year_start"),
+                    "graduation_year_end": rec.get("graduation_year_end"),
+                    "period": rec.get("period"),
+                    "pay_concept": rec.get("pay_concept"),
+                    "currency": rec.get("currency"),
+                    "measure": rec.get("measure"),
+                },
             )
         )
     return chunks

src/backend/indexing/store.py

@@ -13,6 +13,7 @@ On-disk layout (``data/processed/index/``)::
 
 from __future__ import annotations
 
+from collections import defaultdict
 from pathlib import Path
 
 import numpy as np
@@ -33,6 +34,9 @@ class VectorStore:
         self.backend = backend
         self._index = None  # faiss.IndexFlatIP
         self._chunks: list[Chunk] = []
+        # Populated on load() for source-masked retrieval (see search_by_source).
+        self._vectors: np.ndarray | None = None  # (N, D) float32, row-aligned to _chunks
+        self._rows_by_source: dict[str, list[int]] = {}
 
     def add(self, chunks: list[Chunk], vectors) -> int:
         """Add chunks and their (N, D) vectors; returns the number added."""
@@ -76,8 +80,34 @@ class VectorStore:
             for line in chunks_path.read_text(encoding="utf-8").splitlines()
             if line.strip()
         ]
+        # Reconstruct the stored vectors once (flat index → exact) so we can run
+        # cheap per-source masked search in numpy. The corpus is ~2k vectors, so
+        # this is trivial in both time and memory (~16 MB).
+        n = self._index.ntotal
+        self._vectors = self._index.reconstruct_n(0, n) if n else None
+        rows: dict[str, list[int]] = defaultdict(list)
+        for i, c in enumerate(self._chunks):
+            rows[c.source].append(i)
+        self._rows_by_source = dict(rows)
         return True
 
+    def search_by_source(
+        self, query_vector, source: str, k: int = 5
+    ) -> list[tuple[Chunk, float]]:
+        """Return up to ``k`` nearest (chunk, score) pairs from one ``source`` only.
+
+        Searches just that source's rows, so each source contributes its own
+        top-k regardless of the other's score distribution. Scores are cosine
+        (vectors are L2-normalized at build time). Requires ``load()`` first.
+        """
+        rows = self._rows_by_source.get(source, [])
+        if not rows or self._vectors is None:
+            return []
+        q = np.asarray(query_vector, dtype="float32")
+        scores = self._vectors[rows] @ q
+        top = np.argsort(-scores)[: min(k, len(rows))]
+        return [(self._chunks[rows[i]], float(scores[i])) for i in top]
+
     def search(self, query_vector, k: int = 5) -> list[tuple[Chunk, float]]:
         """Return up to ``k`` nearest (chunk, score) pairs, best first."""
         if self._index is None or not self._chunks:

src/backend/rag.py

@@ -23,6 +23,33 @@ from .schemas import ChatMessage, ChatResponse, Chunk
 _store: VectorStore | None = None
 _llm: LLMClient | None = None
 
+# Retrieval is run per knowledge source then fused, so each source contributes
+# regardless of the other's similarity-score scale (directive = long English
+# legal text; IDA = short salary summaries). Sources to retrieve from.
+_SOURCES = ("directive", "lonstatistik")
+K_PER_SOURCE = 5  # top-k pulled from each source before fusion
+K_FINAL = 6       # chunks kept after fusion (passed to the LLM)
+RRF_C = 60        # Reciprocal Rank Fusion constant (standard default)
+
+
+def _rrf_fuse(
+    ranked_lists: list[list[tuple[Chunk, float]]], k_final: int, c: int = RRF_C
+) -> list[Chunk]:
+    """Fuse per-source ranked lists with Reciprocal Rank Fusion.
+
+    Rank-based and score-scale-agnostic: each chunk scores ``sum 1/(c + rank)``
+    across the lists it appears in. Our sources are disjoint, so this cleanly
+    interleaves the lists by rank. Returns the top ``k_final`` chunks.
+    """
+    score: dict[str, float] = {}
+    keep: dict[str, Chunk] = {}
+    for lst in ranked_lists:
+        for rank, (chunk, _s) in enumerate(lst):
+            score[chunk.id] = score.get(chunk.id, 0.0) + 1.0 / (c + rank + 1)
+            keep[chunk.id] = chunk
+    ordered = sorted(keep, key=lambda cid: score[cid], reverse=True)
+    return [keep[cid] for cid in ordered[:k_final]]
+
 
 def _get_store() -> VectorStore:
     """Load the persisted FAISS index once (lazy singleton).
@@ -113,16 +140,18 @@ def build_index() -> str:
 
 @spaces.GPU(duration=180)
 def _retrieve_and_stream(query: str, lang: str, k: int):
-    """Embed query, retrieve, and stream generation — one GPU allocation.
+    """Embed query, retrieve per source + fuse, and stream generation.
 
-    A generator that yields ``(partial_reply, chunks)`` as tokens arrive, so the
-    UI can render incrementally. The first call also lazily loads the models, so
-    a longer GPU duration is allowed. Streaming is the main smoothness lever on
-    ZeroGPU.
+    One GPU allocation. Retrieves the top ``K_PER_SOURCE`` from each knowledge
+    source independently, fuses with Reciprocal Rank Fusion, and keeps the top
+    ``k`` fused chunks — so both the Directive and IDA statistics can surface
+    even though their similarity scores live on different scales. Yields
+    ``(partial_reply, chunks)`` as tokens arrive so the UI renders incrementally.
     """
     qvec = Embedder.get().embed_query(query)
-    hits = _get_store().search(qvec, k=k)
-    chunks = [c for c, _ in hits]
+    store = _get_store()
+    ranked = [store.search_by_source(qvec, s, K_PER_SOURCE) for s in _SOURCES]
+    chunks = _rrf_fuse(ranked, k_final=k)
     prompt = build_grounded_prompt(query, chunks, lang)
     acc = ""
     for piece in _get_llm().stream(prompt):
@@ -130,7 +159,7 @@ def _retrieve_and_stream(query: str, lang: str, k: int):
         yield acc, chunks
 
 
-def answer_stream(messages: list[ChatMessage], lang: str = "en", k: int = 10):
+def answer_stream(messages: list[ChatMessage], lang: str = "en", k: int = K_FINAL):
     """Stream grounded answers as ``ChatResponse`` snapshots (growing reply)."""
     query = next((m.content for m in reversed(messages) if m.role == "user"), "")
     if not query.strip():
@@ -140,7 +169,7 @@ def answer_stream(messages: list[ChatMessage], lang: str = "en", k: int = 10):
         yield ChatResponse(reply=acc, citations=_citations(chunks))
 
 
-def answer(messages: list[ChatMessage], lang: str = "en", k: int = 5) -> ChatResponse:
+def answer(messages: list[ChatMessage], lang: str = "en", k: int = K_FINAL) -> ChatResponse:
     """Non-streaming answer (used by the structured ``/chat`` API)."""
     resp = ChatResponse(reply="", citations=[])
     for resp in answer_stream(messages, lang=lang, k=k):

src/backend/schemas.py

@@ -27,6 +27,11 @@ class Chunk(BaseModel):
     article: str = Field(default="", description="Directive citation label, e.g. 'Art. 7'.")
     page: int | None = Field(default=None, description="Source page (lønstatistik).")
     citation: str = Field(default="", description="Human-readable citation label.")
+    metadata: dict = Field(
+        default_factory=dict,
+        description="Structured, non-embedded source metadata for filtering/display "
+        "(e.g. IDA sector/category/measures, directive article title).",
+    )
 
 
 class ParseResult(BaseModel):

src/frontend/README.md

@@ -8,8 +8,8 @@ selector (en/da/fr) and inline source citations.
 The Scandinavian/sage style is implemented through Gradio's own theming engine
 (`gr.themes.Soft(...).set(...)` + a small CSS block for the Bricolage Grotesque
 / Inter fonts and message-bubble colors) — **not** custom HTML. Theme and CSS
-are passed to `mount_gradio_app(...)` (Gradio 6 moved them off the `Blocks`
-constructor).
+are passed to `demo.launch(...)` in `main.py` (Gradio 6 moved them off the
+`Blocks` constructor).
 
 ## Start it