fegger/pv-agent
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

M1+M2: RAG-Pipeline mit verbindlichem Grounding

Browse Source 
agent/-Paket: Ingest (601 Layer-2-Eintraege -> 3005 Chunks, FTS5-BM25 +
Vektoren-Cache), Hybrid-Retrieval (RRF, Stand-Boost, cross_ref-Erweiterung),
Ollama-Client (embed/chat, think-Flag-Fallback, kurzes Connect-Budget),
Systemprompt mit Zitierpflicht, Post-Validierung (zitierte IDs gemaess
Retrieved-Set, 1x Regenerierung, dann Verweigerung), FastAPI (/ask, /health,
/reindex), CLI, Goldset (31 Fragen, IDs gegen kb.json verifiziert, inkl.
ATZ-Konfliktfall + 4 Verweigerungsfaelle), Eval-Suite, Test-Chat.

41 Offline-Tests gruen. Baseline BM25-only: Hit-Rate 0,871 / Recall@8 0,855 /
MRR 0,476. Hybrid-Messung, Antwortmodus-Eval und Modell-Bake-off (M3) auf
dem Host ausstaendig (Ollama aus der Zed-Sandbox nicht erreichbar).

MEMORY.md und planung.md Umsetzungsstand aktualisiert.
This commit is contained in:
Florian Egger
2026-09-14 16:53:04 +02:00
parent bf8191b013
commit 2cba72aeb0
26 changed files with 2574 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.agents/MEMORY.md
+77
View File
@@ -0,0 +1,77 @@
# Agent-Memory — pv-agent
Rollender Übergabe-Log für agent-Threads. Workflow: `.agents/SKILL.md`
(agent-memory-Skill). Ergänzen, nicht überschreiben.
## Current focus
M1 (Index + Retrieval) und M2 (Ollama-Generierung + Grounding + API) sind
implementiert. Ausstehend: Host-Validierung mit Ollama (Dense-Index,
Antwortmodus-Eval) und Modell-Bake-off (M3). Odoo-Integration (M4) ist
separat zu planen.
## Completed (2026-09-14)
- **Planung** (`planung.md`): Architektur, Grounding-Regeln, Modellfeld
(Bake-off: qwen3.8:27b primär, qwen3:32b, gemma3:27b, mistral-small3.2:24b,
qwen3:14b als Latenz-Untergrenze), Meilensteine M1–M4.
- **Skill** `.agents/skills/pv-rag-agent/SKILL.md`: verbindliche Regeln für
die Implementierung (Grounding, Architektur-Entscheidungen, Gates,
Modellwechsel-Protokoll).
- **Commits**: `25eb285` (Wissensbasis-Import 601 Layer-2-Einträge +
.gitignore), `bf8191b` (Planung + Skills). Noch nicht gepusht — Remote
`http://localhost:3003/fegger/pv-agent.git` ist aus der Zed-Sandbox nicht
erreichbar; User muss vom Host pushen.
- **Implementierung M1+M2** (`agent/`-Paket): kb.py (Parsing + kb.json-Gate),
ingest.py (3005 Chunks aus 601 Einträgen, FTS5 + Vektoren-Cache),
retrieve.py (Hybrid BM25+Dense/RRF, Stand-Boost, cross_ref-Erweiterung),
generate.py (Systemprompt, Post-Validierung, 1× Regenerierung, dann
Verweigerung), ollama_client.py (embed/chat, think-Flag-Fallback),
api.py (/ask /health /reindex), cli.py, eval/ (Goldset 31 Fragen,
evaluate.py), web/index.html, 41 offline Tests (grün).
- **Baseline BM25-only**: Hit-Rate 0,871 · Recall@8 0,855 · MRR 0,476
(31 Fragen). 4 Fehltreffer: Komposita/Stamm-Schwächen (aliquotiert↔
Aliquotierung, Mindestlohngesetz) — Dense-Suche soll diese beheben.
## Open issues / blockers
- **Ollama aus Zed-Sandbox nicht erreichbar** (100.183.83.12:11435 und
localhost:3003 beide geblockt): Dense-Index, Antwortmodus-Eval und
Bake-off müssen auf dem Host laufen. Kommandos: `agent/README.md`
Abschnitt „Deployment auf dem Host".
- **Modelle noch nicht gepullt**: auf dem Host `ollama pull qwen3.8:27b`,
`ollama pull bge-m3` (plus Bake-off-Kandidaten).
- **Reranker** (bge-reranker-v2-m3): API-Unterstützung der installierten
Ollama-Version prüfen — Design funktioniert ohne.
- **qwen3.8-Think-Parameter**: `think: false` wird im Request gesendet
(Auto-Fallback ohne Flag bei 400/404); exaktes Verhalten am Host testen.
- **M4 Odoo**: native LLM-Module des konkreten Odoo-19-Stands verifizieren
(keine API-Annahmen); Option A (dünnes Custom-Modul + Service-API) ist
Default.
## Decisions & conventions
- **D1 (Planung):** Schlanke Eigen-Pipeline statt LangChain/LlamaIndex —
Grounding-Kontrolle schlägt Framework-Komfort bei 601 Dokumenten.
- **D2:** Retrieval-Korpus ist **nur Layer 2**; Layer 1 bleibt aus Prompts
(Lizenz); Antworten zitieren `[kb-id]` + `(Stand YYYY-MM)`.
- **D3:** Umlaut-Folding für FTS (NFKD, ß→ss) — gilt konsistent für Index
und Query; ASCII-Slug-Konvention der Wissensbasis bleibt davon unberührt.
- **D4:** Post-Validierung strikt: zitierte IDs ⊆ Retrieved-Set (Block-Kopf-
IDs); Fließtext-Verweis-IDs sind KEINE Belege (Systemprompt-Regel 2) —
Verstoß → 1× Regenerierung → Verweigerung (UNCERTAIN_MESSAGE).
- **D5:** Vektoren-Tabelle ist Cache (Content-Hash × Modell), Rebuild
löscht sie nicht; `--no-embed` setzt `embed_off` im Config-Copy.
- **D6:** Leeres Retrieval → deterministische Verweigerung ohne LLM-Call.
- **Bake-off-Protokoll** (Skill): Modellwechsel nur über dokumentierten
Goldset-Vergleich; Kriterium: Zitier-Präzision > Verweigerungs-
korrektheit > Latenz.
## Files that matter right now
- `planung.md` — Plan + Entscheidungspunkte (Abschnitt 12) + Stand.
- `.agents/skills/pv-rag-agent/SKILL.md` — verbindliche Regeln.
- `agent/README.md` — Betrieb, Konfiguration, Host-Schritte.
- `agent/eval/goldset.yaml` — Goldset (IDs gegen kb.json verifiziert).
- `agent/generate.py` — Grounding-Kern (Prompt, Post-Validierung).
- `wissensbasis/README.md` — Layer-2-Schema (unverändert gültig).
.gitignore
+1
View File
@@ -11,3 +11,4 @@ data/
__pycache__/ __pycache__/
*.pyc *.pyc
.venv/ .venv/
.pytest_cache/
agent/README.md
+114
View File
@@ -0,0 +1,114 @@
# PV RAG Agent
Lokaler RAG-Agent für österreichische Personalverrechnung: beantwortet
Fragen **ausschließlich** aus der kuratierten Wissensbasis (Layer 2,
`wissensbasis/`, 601 Einträge) — mit ID- und Stand-Beleg, ohne
Trainingswissen, ohne Web-Zugriff. Verbindliche Regeln:
`.agents/skills/pv-rag-agent/SKILL.md`, Plan: `planung.md`.
## Architektur (Kurzfassung)
```
wissensbasis/dokumente/*.md ──ingest──▶ data/index.db
├─ chunks (FTS5, BM25, Umlaut-Folding)
├─ vectors (bge-m3, Content-Hash-Cache)
└─ Metadaten (stand, topic, tags, …)
Frage ──retrieve──▶ Hybrid BM25+Dense (RRF) + cross_ref-Erweiterung
──generate──▶ Ollama (Systemprompt, Zitierpflicht)
──validate──▶ zitierte IDs ⊆ Retrieved-Set? sonst 1× regenerieren, dann verweigern
```
- **Nur Layer 2** als Korpus (kuratiert, lizenzkonform). Layer-1-Volltexte
(`.lexis360/`, `.wiku/`) bleiben außen vor (offener Lizenzpunkt).
- **Kein Ausweg nach außen:** keine Tools, kein Browsing — der einzige
HTTP-Client spricht mit Ollama.
- **Verweigerungspflicht:** leeres/schwaches Retrieval → deterministische
Antwort „Dazu enthält die Wissensbasis keine Aussage." (kein LLM-Call).
## Schnellstart
```bash
pip install -r requirements.txt
# 1) Index bauen (mit Embeddings, wenn Ollama erreichbar)
python -m agent.cli ingest # --no-embed erzwingt BM25-only
# 2) Frage im Terminal
python -m agent.cli ask "Wie hoch ist die AMS-Ersatzquote bei geblockter Altersteilzeit?"
# 3) Goldset-Evaluation (offline: Retrieval-Metriken)
python -m agent.cli eval
# inkl. Antworten + Verweigerungsfälle (benötigt Ollama):
python -m agent.cli eval --answers --json-out data/eval-report.json
# 4) HTTP-API + Test-Chat
python -m agent.cli serve # http://127.0.0.1:8080 (/ask, /health, /reindex)
```
## Konfiguration (Umgebungsvariablen)
| Variable | Default | Bedeutung |
|---|---|---|
| `OLLAMA_URL` | `http://100.183.83.12:11435` | Ollama-Server (Custom-Port!) |
| `PV_ANSWER_MODEL` | `qwen3.8:27b` | Antwortmodell (provisorisch bis Bake-off M3) |
| `PV_EMBED_MODEL` | `bge-m3` | Embedding-Modell |
| `PV_DB_PATH` | `data/index.db` | SQLite-Index |
| `PV_KB_DIR` | `wissensbasis` | Wissensbasis-Verzeichnis |
| `PV_THINK` | `false` | Thinking per Request (qwen3.8: default an) |
| `PV_EMBED_OFF` | `false` | `true` = BM25-only |
| `PV_CONTEXT_BLOCKS` | `8` | Kontextblöcke im Prompt |
| `PV_PORT` | `8080` | API-Port |
## Deployment auf dem Host (Ollama-Maschine)
```bash
# Modelle einmalig pullen
ollama pull qwen3.8:27b
ollama pull bge-m3
curl http://100.183.83.12:11435/api/tags # Erreichbarkeit + Modelle
# Vollständiger Index (BM25 + Dense) — danach eval, Ziel: Recall@8 > 0,9
python -m agent.cli ingest
python -m agent.cli eval
python -m agent.cli eval --answers # Zitier-Präzision, Verweigerungen, Latenz
```
## Baseline (2026-09-14, BM25-only, ohne Dense)
Goldset (31 Fragen, `agent/eval/goldset.yaml`): Hit-Rate 0,871 ·
Recall@8 0,855 · MRR 0,476. Die vier Fehltreffer sind klassische
BM25-Schwächen (Komposita: „aliquotiert"↔„Aliquotierung";
„Mindestlohngesetz") — genau die Fälle, die die Dense-Suche abdecken soll.
Hybrid-Messung auf dem Host aussteht (Ollama aus der Zed-Sandbox nicht
erreichbar).
## Dateien
```
agent/
config.py Env-Konfiguration
kb.py Layer-2-Parsing + kb.json-Gate
normalize.py Umlaut-Folding, FTS-Query-Bau
ingest.py Index-Bau (chunks + FTS5 + Vektoren-Cache)
retrieve.py Hybrid-Retrieval (BM25 + Dense, RRF, cross_refs)
ollama_client.py Ollama-HTTP (embed + chat, think-Fallback)
generate.py Systemprompt, Post-Validierung, Verweigerung
api.py FastAPI (/ask, /health, /reindex)
cli.py ingest | ask | eval | serve
eval/ goldset.yaml + evaluate.py
web/index.html Minimaler Test-Chat
tests/ 41 Tests (offline, Fake-Ollama)
data/ index.db (gitignored)
```
## Tests
```bash
python -m pytest -q # 41 Tests, alle offline
```
## Lizenz-Disziplin
`.lexis360/`, `.wiku/`, `.firecrawl/`, `.ris/` sind lokal und unversioniert
(`.gitignore`). Der Index enthält ausschließlich Layer-2-Kuratierung;
Layer-1-Prompts wären ein Lizenzverstoß und sind im Code nicht vorgesehen.
agent/__init__.py
+8
View File
@@ -0,0 +1,8 @@
"""PV RAG Agent — Wissensbasis-Copilot für österreichische Personalverrechnung.
Schlanke RAG-Pipeline über die kuratierte Wissensbasis (Layer 2):
Ingest -> SQLite-Index (FTS5-BM25 + Dense-Vektoren) -> Hybrid-Retrieval ->
Ollama-Generierung mit verbindlichen Grounding-Regeln.
Verbindliche Regeln: siehe .agents/skills/pv-rag-agent/SKILL.md und planung.md.
"""
agent/api.py
+156
View File
@@ -0,0 +1,156 @@
"""FastAPI-Oberfläche des PV RAG Agent.
Endpunkte:
POST /ask — Frage -> belegte Antwort (oder Verweigerung)
GET /health — Index- und Ollama-Status
POST /reindex — Index-Neuaufbau (nach neuem Wissensbasis-Batch)
"""
from __future__ import annotations
import threading
from contextlib import asynccontextmanager
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, Field
from .config import Config
from .generate import answer_question
from .ingest import build_index
from .ollama_client import OllamaClient
from .retrieve import Retriever
class AskRequest(BaseModel):
question: str = Field(min_length=3, max_length=2000)
top_k: int | None = Field(default=None, ge=1, le=20)
class SourceOut(BaseModel):
id: str
title: str
section: str | None = None
stand: str | None = None
work: str | None = None
class AskResponse(BaseModel):
question: str
answer: str
refused: bool
verified: bool
citations: list[str]
sources: list[SourceOut]
n_context: int
model: str
latency_ms: int
regenerations: int = 0
class AppState:
def __init__(self) -> None:
self.cfg: Config | None = None
self.client: OllamaClient | None = None
self.retriever: Retriever | None = None
self.lock = threading.Lock()
def ensure(self) -> Config:
if self.cfg is None:
self.cfg = Config.from_env()
return self.cfg
def get_client(self) -> OllamaClient:
if self.client is None:
cfg = self.ensure()
self.client = OllamaClient(
cfg.ollama_url,
embed_timeout_s=cfg.embed_timeout_s,
chat_timeout_s=cfg.chat_timeout_s,
)
return self.client
def get_retriever(self) -> Retriever:
if self.retriever is None:
cfg = self.ensure()
self.retriever = Retriever(cfg)
return self.retriever
def reset_retriever(self) -> None:
if self.retriever is not None:
self.retriever.close()
self.retriever = None
@asynccontextmanager
async def lifespan(app: FastAPI):
app.state.rag = AppState()
yield
rag: AppState = app.state.rag
if rag.retriever:
rag.retriever.close()
if rag.client:
rag.client.close()
app = FastAPI(title="PV RAG Agent", version="0.1.0", lifespan=lifespan)
@app.post("/ask", response_model=AskResponse)
def ask(req: AskRequest) -> AskResponse:
rag: AppState = app.state.rag
cfg = rag.ensure()
try:
result = answer_question(
req.question, cfg,
client=rag.get_client(),
retriever=rag.get_retriever(),
top_k=req.top_k,
)
except RuntimeError as e: # Index fehlt
raise HTTPException(status_code=503, detail=str(e)) from e
except Exception as e: # Ollama nicht erreichbar o. Ä.
raise HTTPException(
status_code=503,
detail=f"Antwortgenerierung fehlgeschlagen: {type(e).__name__}: {e}",
) from e
result.pop("draft", None)
return AskResponse(**result)
@app.get("/health")
def health() -> dict:
rag: AppState = app.state.rag
cfg = rag.ensure()
out: dict = {"service": "pv-rag-agent", "config": {
"ollama_url": cfg.ollama_url,
"answer_model": cfg.answer_model,
"embed_model": cfg.embed_model,
}}
try:
retriever = rag.get_retriever()
out["index"] = retriever.stats()
except RuntimeError as e:
out["index"] = {"error": str(e)}
client = rag.get_client()
out["ollama_up"] = client.is_up()
if out["ollama_up"]:
try:
out["ollama_models"] = client.list_models()
except Exception:
out["ollama_models"] = None
return out
@app.post("/reindex")
def reindex() -> dict:
rag: AppState = app.state.rag
cfg = rag.ensure()
with rag.lock:
stats = build_index(cfg, client=rag.get_client())
rag.reset_retriever()
result = stats.as_dict()
result["warning"] = (
"Index ohne Dense-Vektoren aufgebaut (Ollama-Embedding nicht verfügbar) — "
"BM25-only. 'ollama pull " + cfg.embed_model + "' prüfen und erneut reindexen."
if stats.embed_error else None
)
return result
agent/cli.py
+124
View File
@@ -0,0 +1,124 @@
"""CLI des PV RAG Agent.
python -m agent.cli ingest [--no-embed] Index (neu) aufbauen
python -m agent.cli ask "Frage?" [--top-k N] [--json]
python -m agent.cli eval [--answers] [--limit N] [--k 8] [--json-out FILE]
python -m agent.cli serve [--host 0.0.0.0]
"""
from __future__ import annotations
import argparse
import json
import sys
from .config import Config
def _cmd_ingest(args: argparse.Namespace, cfg: Config) -> int:
import dataclasses
from .ingest import build_index
client = None
if args.no_embed:
cfg = dataclasses.replace(cfg, embed_off=True)
elif not cfg.embed_off:
from .ollama_client import OllamaClient
client = OllamaClient(
cfg.ollama_url,
embed_timeout_s=cfg.embed_timeout_s,
chat_timeout_s=cfg.chat_timeout_s,
)
if not client.is_up():
print(
f"[warn] Ollama unter {cfg.ollama_url} nicht erreichbar — "
"Index wird BM25-only aufgebaut.",
file=sys.stderr,
)
try:
stats = build_index(cfg, client=client)
finally:
if client is not None:
client.close()
print(json.dumps(stats.as_dict(), indent=2, ensure_ascii=False))
if stats.embed_error:
print(f"[warn] {stats.embed_error}", file=sys.stderr)
return 0
def _cmd_ask(args: argparse.Namespace, cfg: Config) -> int:
from .generate import answer_question
try:
result = answer_question(args.question, cfg, top_k=args.top_k)
except Exception as e:
print(
f"[Fehler] Antwortgenerierung fehlgeschlagen: "
f"{type(e).__name__}: {e}\n"
f"Ollama erreichbar unter {cfg.ollama_url}? 'curl {cfg.ollama_url}/api/tags'",
file=sys.stderr,
)
return 2
if args.json:
print(json.dumps(result, indent=2, ensure_ascii=False))
return 0
print(result["answer"])
if result["sources"]:
print("\nQuellen:")
for s in result["sources"]:
print(f" - {s['id']} — {s['title']} ({s['stand']})")
status = "VERWEIGERT" if result["refused"] else (
"OK" if result["verified"] else "UNVERIFIZIERT"
)
print(
f"\n[{status} · {result['model']} · {result['latency_ms']} ms · "
f"{result['n_context']} Kontextblöcke]"
)
return 0
def _cmd_eval(args: argparse.Namespace, cfg: Config) -> int:
from .eval.evaluate import run_eval
return run_eval(cfg, args)
def _cmd_serve(args: argparse.Namespace, cfg: Config) -> int:
import uvicorn
uvicorn.run("agent.api:app", host=args.host, port=cfg.port, log_level="info")
return 0
def main(argv: list[str] | None = None) -> int:
cfg = Config.from_env()
parser = argparse.ArgumentParser(prog="agent.cli", description=__doc__)
sub = parser.add_subparsers(dest="cmd", required=True)
p_ingest = sub.add_parser("ingest", help="Index (neu) aufbauen")
p_ingest.add_argument("--no-embed", action="store_true",
help="Keine Embeddings erzeugen (BM25-only)")
p_ask = sub.add_parser("ask", help="Frage stellen")
p_ask.add_argument("question")
p_ask.add_argument("--top-k", type=int, default=None)
p_ask.add_argument("--json", action="store_true")
p_eval = sub.add_parser("eval", help="Goldset-Evaluation")
p_eval.add_argument("--answers", action="store_true",
help="inkl. Antwortgenerierung (benötigt Ollama)")
p_eval.add_argument("--limit", type=int, default=None)
p_eval.add_argument("--k", type=int, default=8, help="K für Recall@k")
p_eval.add_argument("--json-out", default=None)
p_serve = sub.add_parser("serve", help="HTTP-API starten")
p_serve.add_argument("--host", default="127.0.0.1")
args = parser.parse_args(argv)
handlers = {
"ingest": _cmd_ingest,
"ask": _cmd_ask,
"eval": _cmd_eval,
"serve": _cmd_serve,
}
return handlers[args.cmd](args, cfg)
if __name__ == "__main__":
raise SystemExit(main())
agent/config.py
+87
View File
@@ -0,0 +1,87 @@
"""Konfiguration des PV RAG Agent (alle Werte per Umgebungsvariable übersteuerbar)."""
from __future__ import annotations
import os
from dataclasses import dataclass
def _env_str(name: str, default: str) -> str:
v = os.environ.get(name)
return v if v not in (None, "") else default
def _env_int(name: str, default: int) -> int:
try:
return int(os.environ.get(name, default))
except (TypeError, ValueError):
return default
def _env_float(name: str, default: float) -> float:
try:
return float(os.environ.get(name, default))
except (TypeError, ValueError):
return default
def _env_bool(name: str, default: bool) -> bool:
v = os.environ.get(name)
if v is None:
return default
return v.strip().lower() in ("1", "true", "yes", "on")
@dataclass
class Config:
# Pfade (relativ zum Repo-Root, sofern nicht absolut)
kb_dir: str = "wissensbasis"
db_path: str = "data/index.db"
# Ollama (Custom-Port 11435 — nicht "korrigieren", s. Skill)
ollama_url: str = "http://100.183.83.12:11435"
embed_model: str = "bge-m3"
answer_model: str = "qwen3.8:27b" # provisorisch bis Bake-off (M3)
# Generierung
temperature: float = 0.1
num_ctx: int = 16384
num_predict: int = 1024
think: bool = False # Thinking per Request abschalten (Latenz)
chat_timeout_s: float = 300.0
embed_timeout_s: float = 240.0
# Retrieval
embed_off: bool = False # True = BM25-only (ohne Dense-Index/-Suche)
candidate_pool: int = 50 # Kandidaten je Liste vor der Fusion
context_blocks: int = 8 # Kontext-Blöcke im Prompt
cross_ref_expand: int = 3 # Top-Einträge, deren cross_refs ergänzt werden
cross_ref_max_extra: int = 6 # Obergrenze der Ergänzungen
rrf_k: int = 60
recency_boost: float = 0.005 # additiv auf RRF-Score, gewichtet nach Stand
# Service
port: int = 8080
@classmethod
def from_env(cls) -> "Config":
d = cls()
return cls(
kb_dir=_env_str("PV_KB_DIR", d.kb_dir),
db_path=_env_str("PV_DB_PATH", d.db_path),
ollama_url=_env_str("OLLAMA_URL", d.ollama_url),
embed_model=_env_str("PV_EMBED_MODEL", d.embed_model),
answer_model=_env_str("PV_ANSWER_MODEL", d.answer_model),
temperature=_env_float("PV_TEMPERATURE", d.temperature),
num_ctx=_env_int("PV_NUM_CTX", d.num_ctx),
num_predict=_env_int("PV_NUM_PREDICT", d.num_predict),
think=_env_bool("PV_THINK", d.think),
chat_timeout_s=_env_float("PV_CHAT_TIMEOUT_S", d.chat_timeout_s),
embed_timeout_s=_env_float("PV_EMBED_TIMEOUT_S", d.embed_timeout_s),
embed_off=_env_bool("PV_EMBED_OFF", d.embed_off),
candidate_pool=_env_int("PV_CANDIDATE_POOL", d.candidate_pool),
context_blocks=_env_int("PV_CONTEXT_BLOCKS", d.context_blocks),
cross_ref_expand=_env_int("PV_CROSS_REF_EXPAND", d.cross_ref_expand),
cross_ref_max_extra=_env_int("PV_CROSS_REF_MAX_EXTRA", d.cross_ref_max_extra),
recency_boost=_env_float("PV_RECENCY_BOOST", d.recency_boost),
port=_env_int("PV_PORT", d.port),
)
agent/eval/__init__.py
+1
View File
@@ -0,0 +1 @@
"""Eval-Paket: Goldset + Metriken für Retrieval und belegte Antworten."""
agent/eval/evaluate.py
+178
View File
@@ -0,0 +1,178 @@
"""Goldset-Evaluation.
Modus 1 (offline, ohne Ollama): Retrieval-Metriken — Recall@k, Hit-Rate, MRR.
Modus 2 (--answers, benötigt Ollama): Zitier-Präzision (validated),
Verweigerungskorrektheit, erwartete Quelle zitiert, Latenz.
Kriterien laut Skill: Recall@8 > 0,9; Zitier-Präzision 100 %;
Verweigerungen korrekt.
"""
from __future__ import annotations
import json
import statistics
from pathlib import Path
import yaml
from ..config import Config
from ..retrieve import Retriever
GOLDSET_PATH = Path(__file__).parent / "goldset.yaml"
def load_goldset(path: Path = GOLDSET_PATH) -> list[dict]:
data = yaml.safe_load(Path(path).read_text(encoding="utf-8"))
questions = data.get("questions", [])
if not questions:
raise SystemExit(f"goldset leer: {path}")
return questions
def evaluate_retrieval(cfg: Config, questions: list[dict], k: int = 8) -> list[dict]:
retriever = Retriever(cfg)
rows: list[dict] = []
try:
for q in questions:
if q.get("expect_refusal"):
continue # Verweigerungsfälle werden nur im Antwortmodus gemessen
expected = set(q.get("expected_ids", []))
if not expected:
continue
results = retriever.search(q["question"], n_entries=k)
retrieved = [r.entry_id for r in results]
hits = expected & set(retrieved)
rank = next(
(retrieved.index(e) + 1 for e in expected if e in retrieved), None
)
rows.append(
{
"id": q["id"],
"question": q["question"],
"expected": sorted(expected),
"retrieved": retrieved,
"recall": len(hits) / len(expected),
"hit": bool(hits),
"mrr": (1.0 / rank) if rank else 0.0,
}
)
finally:
retriever.close()
return rows
def summarize_retrieval(rows: list[dict]) -> dict:
if not rows:
return {"n": 0}
return {
"n": len(rows),
"hit_rate": round(sum(r["hit"] for r in rows) / len(rows), 4),
"mean_recall_at_k": round(statistics.mean(r["recall"] for r in rows), 4),
"mrr": round(statistics.mean(r["mrr"] for r in rows), 4),
}
def evaluate_answers(cfg: Config, questions: list[dict]) -> list[dict]:
from ..generate import answer_question # Ollama nötig — hier erst importieren
rows: list[dict] = []
for q in questions:
result = answer_question(q["question"], cfg)
expected = set(q.get("expected_ids", []))
should_refuse = bool(q.get("expect_refusal"))
rows.append(
{
"id": q["id"],
"question": q["question"],
"refused": result["refused"],
"should_refuse": should_refuse,
"refusal_correct": result["refused"] == should_refuse,
"verified": result["verified"],
"citations": result["citations"],
"expected_cited": (
any(c in expected for c in result["citations"]) if expected else None
),
"regenerations": result["regenerations"],
"latency_ms": result["latency_ms"],
}
)
return rows
def summarize_answers(rows: list[dict]) -> dict:
if not rows:
return {"n": 0}
lat = [r["latency_ms"] for r in rows]
return {
"n": len(rows),
"citation_precision": round(
sum(r["verified"] for r in rows) / len(rows), 4
),
"refusal_correct_rate": round(
sum(r["refusal_correct"] for r in rows) / len(rows), 4
),
"expected_cited_rate": round(
sum(bool(r["expected_cited"]) for r in rows)
/ max(1, sum(1 for r in rows if r["expected_cited"] is not None)),
4,
),
"latency_ms_mean": round(statistics.mean(lat)),
"latency_ms_p95": round(sorted(lat)[int(len(lat) * 0.95) - 1]),
"regenerations_total": sum(r["regenerations"] for r in rows),
}
def run_eval(cfg: Config, args) -> int:
questions = load_goldset()
if args.limit:
questions = questions[: args.limit]
mode = "retrieval + Antworten" if args.answers else "nur Retrieval (offline)"
print(f"Goldset: {len(questions)} Fragen · Modus: {mode} · k={args.k}")
print(f"Antwortmodell: {cfg.answer_model} · Embedding: {cfg.embed_model}")
print("-" * 78)
report: dict = {"mode": mode, "k": args.k, "model": cfg.answer_model}
rows = evaluate_retrieval(cfg, questions, k=args.k)
for r in rows:
mark = "✓" if r["hit"] else "✗"
print(
f"{mark} {r['id']:8s} recall={r['recall']:.2f} "
f"mrr={r['mrr']:.2f} {r['question'][:56]}"
)
if not r["hit"]:
print(f" erwartet: {', '.join(r['expected'])}")
print(f" erhalten: {', '.join(r['retrieved'][:args.k])}")
summary = summarize_retrieval(rows)
report["retrieval"] = {"summary": summary, "rows": rows}
print("-" * 78)
print(f"Retrieval: n={summary['n']} Hit-Rate={summary['hit_rate']} "
f"Recall@{args.k}={summary['mean_recall_at_k']} MRR={summary['mrr']}")
if args.answers:
arows = evaluate_answers(cfg, questions)
for r in arows:
mark = "✓" if (r["verified"] and r["refusal_correct"]) else "✗"
print(
f"{mark} {r['id']:8s} refused={r['refused']} verified={r['verified']} "
f"zit={','.join(r['citations'][:4])} {r['latency_ms']}ms"
)
asummary = summarize_answers(arows)
report["answers"] = {"summary": asummary, "rows": arows}
print("-" * 78)
print(
f"Antworten: n={asummary['n']} Zitier-Präzision={asummary['citation_precision']} "
f"Verweigerung korrekt={asummary['refusal_correct_rate']} "
f"erwartete Quelle zitiert={asummary['expected_cited_rate']}"
)
print(
f"Latenz: mean={asummary['latency_ms_mean']}ms p95={asummary['latency_ms_p95']}ms "
f"Regenerierungen={asummary['regenerations_total']}"
)
if args.json_out:
Path(args.json_out).parent.mkdir(parents=True, exist_ok=True)
Path(args.json_out).write_text(
json.dumps(report, indent=2, ensure_ascii=False), encoding="utf-8"
)
print(f"Report geschrieben: {args.json_out}")
return 0
agent/eval/goldset.yaml
+116
View File
@@ -0,0 +1,116 @@
# Goldset für die Evaluation des PV RAG Agent.
# expected_ids: verifiziert gegen wissensbasis/kb.json (Titel-Suche).
# expect_refusal: Frage darf/darf nicht aus der Wissensbasis beantwortet
# werden — Antwortmodus (eval --answers) muss verweigern.
# q-002 ist der bekannte Korpuskonflikt (ATZ-Ersatzquote 28,5 vs. 27,5 %).
questions:
- id: q-001
question: "Was ist Altersteilzeit und wie funktioniert der Lohnausgleich?"
expected_ids: [lb-atz-07, lb-atz-04]
- id: q-002
question: "Wie hoch ist die AMS-Ersatzquote bei geblockter Altersteilzeit?"
expected_ids: [lb-atz-07, lb-atz-09, lb-atz-12]
note: "Korpuskonflikt — beide Werte mit ⚠ nennen."
- id: q-003
question: "Für welche Arbeitnehmer ist die Altersteilzeit förderbar?"
expected_ids: [lb-atz-03]
- id: q-004
question: "Wie viele Wochen gesetzlicher Urlaub stehen einem Arbeitnehmer zu?"
expected_ids: [lb-url-05]
- id: q-005
question: "Wie wird das Urlaubsentgelt beim Wechsel zwischen Teilzeit und Vollzeit berechnet?"
expected_ids: [lb-url-13]
- id: q-006
question: "Wie lange dauert die Entgeltfortzahlung im Krankenstand?"
expected_ids: [lb-krs-08]
- id: q-007
question: "Welche Fristen gelten für die Auflösung in der Probezeit?"
expected_ids: [lb-bnd-37]
- id: q-008
question: "Welche Verfügungsmöglichkeiten gibt es bei der Abfertigung neu?"
expected_ids: [lb-end-03]
- id: q-009
question: "Welche Bezugsbestandteile sind beitragsfrei nach § 49 Abs. 3 ASVG?"
expected_ids: [lb-sva-03, lb-sva-04]
- id: q-010
question: "Wie hoch sind die Sozialversicherungs-Beitragssätze für Dienstnehmer?"
expected_ids: [lb-sva-06]
- id: q-011
question: "Was gilt als Nachtschwerarbeit und welche Folgen hat das?"
expected_ids: [lb-nsc-01]
- id: q-012
question: "Wie sind Mitarbeiterrabatte abgabenrechtlich zu behandeln?"
expected_ids: [lb-sac-02]
- id: q-013
question: "Wie wird die Privatnutzung eines Dienstwagens besteuert?"
expected_ids: [lb-sac-03]
- id: q-014
question: "Unter welchen Voraussetzungen gibt es Pendlerförderung?"
expected_ids: [lb-pen-01]
- id: q-015
question: "Wie werden Tagesgelder bei Dienstreisen abgerechnet?"
expected_ids: [lb-rei-09]
- id: q-016
question: "Wie läuft eine GPLB ab?"
expected_ids: [lb-gpl-01]
- id: q-017
question: "Was ist bei der Entsendung von Arbeitnehmern ins Ausland zu beachten?"
expected_ids: [lb-grz-06, lb-grz-07]
- id: q-018
question: "Wie funktioniert die betriebliche Vorsorgekasse mit Beitragszahlung?"
expected_ids: [lb-vor-02, lb-vor-03]
- id: q-019
question: "Welche Pflichten gelten bei der Einstellung von Lehrlingen?"
expected_ids: [lb-leh-03]
- id: q-020
question: "Welche Beschäftigungsverbote gelten für Schwangere?"
expected_ids: [lb-sch-04]
- id: q-021
question: "Wann beginnt die Elternkarenz und wie lange kann sie dauern?"
expected_ids: [lb-kar-01]
- id: q-022
question: "Wie ist eine Überstundenpauschale zu behandeln?"
expected_ids: [lb-ues-02]
- id: q-023
question: "Was muss ein Dienstzeugnis enthalten?"
expected_ids: [lb-bso-04]
- id: q-024
question: "Was ist bei Kurzarbeit arbeitsrechtlich zu beachten?"
expected_ids: [lb-azm-04, lb-azm-05, lb-azm-06]
- id: q-025
question: "Was passiert mit den Ansprüchen der Arbeitnehmer beim Betriebsübergang?"
expected_ids: [lb-ins-01]
- id: q-026
question: "Wie sind Nachzahlungen abzurechnen?"
expected_ids: [lb-naz-01]
- id: q-027
question: "Was ist beim Einlangen einer Lohnpfändung zu beachten?"
expected_ids: [lb-pfa-01, lb-pfa-02]
- id: q-028
question: "Wird die Sonderzahlung bei Eintritt oder Austritt aliquotiert?"
expected_ids: [lb-son-01]
- id: q-029
question: "Welche Neuerungen behandelt WIKU Personal aktuell 2026?"
expected_ids: [wk-akt-01]
- id: q-030
question: "Wer zahlt das Wochengeld und wie wird es berechnet?"
expected_ids: [lb-msf-03]
- id: q-031
question: "Was regelt das Mindestlohngesetz und für wen gilt es?"
expected_ids: [lb-ent-09]
# --- Verweigerungsfälle (Antwortmodus) ---
- id: r-001
question: "Wie hoch ist der aktuelle EUR-USD-Wechselkurs?"
expect_refusal: true
- id: r-002
question: "Wer war Bundeskanzler Österreichs im Jahr 2000?"
expect_refusal: true
- id: r-003
question: "Bis wann muss die Umsatzsteuervoranmeldung abgegeben werden?"
expect_refusal: true
note: "Schwerer Fall: 13 Chunks erwähnen 'Umsatzsteuer' beiläufig — Retrieval nicht leer, aber inhaltlich nicht gedeckt."
- id: r-004
question: "Wie bereite ich einen Pitch für Investoren vor?"
expect_refusal: true
agent/generate.py
+218
View File
@@ -0,0 +1,218 @@
"""Antwort-Generierung mit verbindlichen Grounding-Regeln.
Kern der Pipeline: Systemprompt (nur Kontext, Zitierpflicht, Verweigerung),
kontrollierte cross_ref-Erweiterung und Post-Validierung — jede zitierte ID
muss im Retrieved-Set stehen, sonst eine Regenerierung, dann Verweigerung.
Keine Antwort verlässt die Pipeline mit ungültigen Zitaten.
"""
from __future__ import annotations
import re
import time
from .config import Config
from .normalize import normalize_text
from .ollama_client import OllamaClient
from .retrieve import ChunkResult, Retriever
REFUSAL_MESSAGE = "Dazu enthält die Wissensbasis keine Aussage."
UNCERTAIN_MESSAGE = (
"⚠ Zu dieser Frage kann ich keine verlässlich belegte Antwort "
"aus der Wissensbasis geben."
)
SYSTEM_PROMPT = """Du bist ein präziser Assistent für österreichische Personalverrechnung.
Du beantwortest Fragen AUSSCHLIESSLICH auf Basis der nummerierten Kontextblöcke
aus der internen Wissensbasis.
Verbindliche Regeln:
1. Jede fachliche Aussage muss durch die Kontextblöcke gedeckt sein. Verwende
KEIN Wissen aus deinem Training und ergänze nichts aus eigenem Wissen.
2. Belege jede fachliche Aussage mit der KB-ID in eckigen Klammern, z. B.
[lb-atz-07]. Zitiere NUR die IDs aus den Block-Köpfen („Block N — [id] …“).
IDs, die nur im Fließtext als Verweis genannt werden, sind Querverweise
und KEINE Belege.
3. Gib jeden Wert mit seinem Stand an, z. B. „28,5 % (Stand 2026-01)“.
4. Beantworten die Kontextblöcke die Frage nicht, antworte exakt:
„Dazu enthält die Wissensbasis keine Aussage.“ — und schlage nichts vor.
5. Widersprechen sich Kontextblöcke, nenne beide Werte mit ihren IDs und
kennzeichne den Widerspruch mit ⚠. Löse Widersprüche niemals stillschweigend auf.
6. Nenne Paragraphen und Gesetze nur, wenn ein Kontextblock sie nennt.
7. Antworte auf Deutsch, prägnant (Stichpunkte, wo sinnvoll), und füge am
Ende eine Zeile „Quellen:“ mit den verwendeten IDs (ID — Titel, Stand) an.
Verletze Regel 2 oder Regel 4 niemals — im Zweifel verweigere die Antwort."""
CITE_RE = re.compile(r"\b(?:lb|wk)-[a-z0-9]+-\d+\b")
_THINK_RE = re.compile(r"<think>.*?</think>", re.DOTALL)
def strip_think(text: str) -> str:
"""Entfernt <think>-Blöcke defensiv (falls Thinking nicht abschaltbar war)."""
return _THINK_RE.sub("", text).strip()
def looks_like_refusal(answer: str) -> bool:
folded = normalize_text(answer)
return (
"keine aussage" in folded
or "keine verlasslich belegte" in folded
or "nicht in der wissensbasis" in folded
)
def build_user_content(question: str, results: list[ChunkResult]) -> str:
blocks = []
for i, r in enumerate(results, 1):
header = (
f"Block {i} — [{r.entry_id}] {r.title} · Abschnitt: {r.section} "
f"· Stand: {r.stand} · Werk: {r.work}"
)
blocks.append(f"{header}\n{r.text}")
context = "\n\n---\n\n".join(blocks)
return f"Kontextblöcke aus der Wissensbasis:\n\n{context}\n\nFrage: {question}"
def validate_answer(answer: str, allowed_ids: list[str]) -> list[str]:
"""Regel-2/4-Prüfung: zitierte IDs ⊆ Kontext; keine unbelegte Fachantwort."""
cited = set(CITE_RE.findall(answer))
violations: list[str] = []
unknown = sorted(cited - set(allowed_ids))
if unknown:
violations.append(f"zitierte IDs außerhalb des Kontexts: {', '.join(unknown)}")
if not cited and not looks_like_refusal(answer):
violations.append("keine KB-ID zitiert")
return violations
def _source_rows(results: list[ChunkResult]) -> list[dict]:
rows = []
seen: set[str] = set()
for r in results:
if r.entry_id in seen:
continue
seen.add(r.entry_id)
rows.append(
{
"id": r.entry_id,
"title": r.title,
"section": r.section,
"stand": r.stand,
"work": r.work,
"source": r.source,
}
)
return rows
def answer_question(
question: str,
cfg: Config,
client: OllamaClient | None = None,
retriever: Retriever | None = None,
top_k: int | None = None,
) -> dict:
"""Vollständiger Ask-Zyklus: Retrieval -> Prompt -> LLM -> Post-Validierung."""
t0 = time.perf_counter()
own_retriever = retriever is None
if retriever is None:
retriever = Retriever(cfg)
try:
results = retriever.search(question, n_entries=top_k)
finally:
if own_retriever:
retriever.close()
def finish(answer, refused, verified, citations, regenerations=0, draft=None):
return {
"question": question,
"answer": answer,
"refused": refused,
"verified": verified,
"citations": citations,
"sources": _source_rows(results),
"n_context": len(results),
"model": cfg.answer_model,
"regenerations": regenerations,
"latency_ms": round((time.perf_counter() - t0) * 1000),
"draft": draft,
}
if not results:
# Verweigerungspflicht: leeres Retrieval -> deterministische Antwort
return finish(REFUSAL_MESSAGE, refused=True, verified=True, citations=[])
allowed = [r.entry_id for r in results]
by_id = {r.entry_id: r for r in results}
if client is None:
client = OllamaClient(
cfg.ollama_url,
embed_timeout_s=cfg.embed_timeout_s,
chat_timeout_s=cfg.chat_timeout_s,
)
messages = [
{"role": "system", "content": SYSTEM_PROMPT},
{"role": "user", "content": build_user_content(question, results)},
]
def chat(msgs):
return strip_think(
client.chat(
cfg.answer_model,
msgs,
temperature=cfg.temperature,
num_ctx=cfg.num_ctx,
num_predict=cfg.num_predict,
think=cfg.think,
)
)
final = chat(messages)
violations = validate_answer(final, allowed)
regenerations = 0
if violations:
regenerations = 1
warn = (
"Deine letzte Antwort verstieß gegen die Regeln: "
+ "; ".join(violations)
+ f". Erlaubte KB-IDs sind ausschließlich: {', '.join(sorted(set(allowed)))}. "
"Beantworte die Frage erneut und zitiere nur diese IDs — oder verweigere "
f"mit dem vorgesehenen Satz („{REFUSAL_MESSAGE}“)."
)
retry = chat(
messages
+ [{"role": "assistant", "content": final},
{"role": "user", "content": warn}]
)
retry_violations = validate_answer(retry, allowed)
if not retry_violations:
final = retry
violations = []
else:
return finish(
UNCERTAIN_MESSAGE,
refused=True,
verified=False,
citations=[],
regenerations=regenerations,
draft=retry,
)
citations = sorted(set(CITE_RE.findall(final)))
refused = looks_like_refusal(final)
sources = [
{
"id": cid,
"title": by_id[cid].title,
"section": by_id[cid].section,
"stand": by_id[cid].stand,
"work": by_id[cid].work,
}
for cid in citations
if cid in by_id
]
return finish(
final, refused=refused, verified=not violations,
citations=citations, regenerations=regenerations,
)
agent/ingest.py
+202
View File
@@ -0,0 +1,202 @@
"""Index-Bau: Layer-2-Einträge -> SQLite (FTS5-BM25 + Dense-Vektoren).
Chunking: H2-Sektion je Eintrag (Parent-Child: Retrieval auf Sektion,
Kontext = Sektion + Metadatenkopf). Die Vektoren-Tabelle ist ein Cache
(Content-Hash) und überlebt Rebuilds — ein Reindex bettet nur Neues ein.
"""
from __future__ import annotations
import hashlib
import json
import sqlite3
import time
from dataclasses import asdict, dataclass
from datetime import datetime, timezone
from pathlib import Path
import numpy as np
from .config import Config
from .kb import KBEntry, Section, load_kb
from .normalize import normalize_text
SCHEMA_VERSION = 1
EMBED_BATCH = 32
SCHEMA = """
CREATE TABLE IF NOT EXISTS meta (key TEXT PRIMARY KEY, value TEXT);
CREATE TABLE IF NOT EXISTS chunks (
chunk_id INTEGER PRIMARY KEY,
entry_id TEXT NOT NULL,
section TEXT NOT NULL,
text TEXT NOT NULL,
norm TEXT NOT NULL,
content_hash TEXT NOT NULL,
title TEXT, work TEXT, chapter TEXT, topic TEXT,
stand TEXT, batch INTEGER,
tags TEXT, legal_bases TEXT, cross_refs TEXT,
source_pdf TEXT, source_text TEXT
);
CREATE INDEX IF NOT EXISTS idx_chunks_entry ON chunks(entry_id);
CREATE VIRTUAL TABLE IF NOT EXISTS chunks_fts USING fts5(norm);
CREATE TABLE IF NOT EXISTS vectors (
content_hash TEXT NOT NULL,
model TEXT NOT NULL,
dim INTEGER NOT NULL,
vec BLOB NOT NULL,
PRIMARY KEY (content_hash, model)
);
"""
@dataclass
class IndexStats:
n_entries: int = 0
n_chunks: int = 0
n_embedded: int = 0
embed_error: str | None = None
duration_s: float = 0.0
kb_dir: str = ""
db_path: str = ""
schema_version: int = SCHEMA_VERSION
def as_dict(self) -> dict:
return asdict(self)
def embed_text(title: str, section_title: str, text: str) -> str:
"""Einheitlicher Embedding-Input (Title + Abschnitt + Body)."""
return f"{title}\n{section_title}\n\n{text}"
def content_hash_for(title: str, section_title: str, text: str) -> str:
return hashlib.sha256(
embed_text(title, section_title, text).encode("utf-8")
).hexdigest()
def norm_text_for(entry: KBEntry, section: Section) -> str:
"""FTS-Text: Titel + Abschnitt + Tags + Rechtsgrundlagen + Kapitel + Body."""
parts = [
entry.title,
section.title,
" ".join(entry.tags),
" ".join(entry.legal_bases),
entry.chapter,
section.text,
]
return normalize_text("\n".join(p for p in parts if p))
def _insert_chunk(con: sqlite3.Connection, entry: KBEntry, section: Section) -> int:
h = content_hash_for(entry.title, section.title, section.text)
cur = con.execute(
"""INSERT INTO chunks (
entry_id, section, text, norm, content_hash,
title, work, chapter, topic, stand, batch,
tags, legal_bases, cross_refs, source_pdf, source_text
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""",
(
entry.id, section.title, section.text, norm_text_for(entry, section), h,
entry.title, entry.work, entry.chapter, entry.topic, entry.stand,
entry.batch,
json.dumps(entry.tags, ensure_ascii=False),
json.dumps(entry.legal_bases, ensure_ascii=False),
json.dumps(entry.cross_refs, ensure_ascii=False),
entry.source.get("pdf", ""), entry.source.get("text", ""),
),
)
chunk_id = cur.lastrowid
con.execute(
"INSERT INTO chunks_fts(rowid, norm) VALUES (?, ?)",
(chunk_id, norm_text_for(entry, section)),
)
return chunk_id
def _embed_missing(cfg: Config, con: sqlite3.Connection, client) -> tuple[int, str | None]:
"""Bettet alle Chunk-Hashes ein, die für cfg.embed_model fehlen."""
rows = con.execute(
"SELECT DISTINCT content_hash, title, section, text FROM chunks"
).fetchall()
unique: dict[str, str] = {}
for h, title, section, text in rows:
if h not in unique:
unique[h] = embed_text(title, section, text)
have = {
r[0] for r in con.execute(
"SELECT content_hash FROM vectors WHERE model = ?",
(cfg.embed_model,),
).fetchall()
}
todo = [h for h in unique if h not in have]
if not todo:
return 0, None
n = 0
for i in range(0, len(todo), EMBED_BATCH):
batch = todo[i:i + EMBED_BATCH]
texts = [unique[h] for h in batch]
try:
embs = client.embed(cfg.embed_model, texts)
except Exception as e: # httpx/Ollama-Fehler -> BM25-only weiterlaufen
return n, f"embedding failed at batch {i // EMBED_BATCH + 1}: {e}"
for h, vec in zip(batch, embs):
arr = np.asarray(vec, dtype=np.float32)
con.execute(
"INSERT OR REPLACE INTO vectors(content_hash, model, dim, vec) "
"VALUES (?, ?, ?, ?)",
(h, cfg.embed_model, int(arr.shape[0]), arr.tobytes()),
)
n += 1
return n, None
def build_index(cfg: Config, client=None) -> IndexStats:
"""Vollständiger Rebuild von chunks/FTS; Vektoren-Cache bleibt erhalten."""
t0 = time.perf_counter()
entries = load_kb(cfg.kb_dir, verify_registry=True)
db_path = Path(cfg.db_path)
db_path.parent.mkdir(parents=True, exist_ok=True)
stats = IndexStats(kb_dir=str(cfg.kb_dir), db_path=str(db_path))
con = sqlite3.connect(db_path)
try:
con.executescript(SCHEMA)
con.execute("DELETE FROM chunks")
con.execute("DELETE FROM chunks_fts")
con.execute("DELETE FROM meta")
for entry in entries:
for section in entry.sections:
_insert_chunk(con, entry, section)
con.commit()
stats.n_entries = len(entries)
stats.n_chunks = con.execute("SELECT COUNT(*) FROM chunks").fetchone()[0]
if not cfg.embed_off:
if client is None:
from .ollama_client import OllamaClient
client = OllamaClient(
cfg.ollama_url,
embed_timeout_s=cfg.embed_timeout_s,
chat_timeout_s=cfg.chat_timeout_s,
)
try:
stats.n_embedded, stats.embed_error = _embed_missing(cfg, con, client)
except Exception as e:
stats.embed_error = f"{type(e).__name__}: {e}"
con.executemany(
"INSERT OR REPLACE INTO meta(key, value) VALUES (?, ?)",
[
("schema_version", str(SCHEMA_VERSION)),
("built_at", datetime.now(timezone.utc).isoformat()),
("kb_dir", str(cfg.kb_dir)),
("embed_model", "" if cfg.embed_off else cfg.embed_model),
("n_entries", str(stats.n_entries)),
("n_chunks", str(stats.n_chunks)),
],
)
con.commit()
finally:
con.close()
stats.duration_s = round(time.perf_counter() - t0, 2)
return stats
agent/kb.py
+164
View File
@@ -0,0 +1,164 @@
"""Layer-2-Wissensbasis: Parsing (Frontmatter + H2-Sektionen) und kb.json-Gate.
Die Layer-2-Frontmatter ist die Single Source of Truth; kb.json ist deren
generierte, validierte Projektion. Der Gate bricht den Ingest bei Abweichung
ab (Fehlermeldung nennt die Regenerierung der Registry als nächsten Schritt).
"""
from __future__ import annotations
import json
import re
from dataclasses import dataclass, field
from pathlib import Path
import yaml
ID_RE = re.compile(r"^(lb|wk)-[a-z0-9]+-\d+$")
STAND_RE = re.compile(r"^\d{4}-\d{2}$")
REQUIRED_KEYS = (
"id", "batch", "title", "work", "chapter", "topic", "author",
"stand", "source", "legal_bases", "tags", "cross_refs",
)
class KbValidationError(Exception):
"""Wissensbasis oder Registry ist inkonsistent — Ingest wird abgebrochen."""
@dataclass
class Section:
title: str
text: str
@dataclass
class KBEntry:
id: str
title: str
work: str
chapter: str
topic: str
author: str
stand: str
batch: int
source: dict
legal_bases: list
tags: list
cross_refs: list
path: Path
sections: list = field(default_factory=list)
def parse_frontmatter(raw: str) -> tuple[dict, str]:
lines = raw.splitlines()
if not lines or lines[0].strip() != "---":
raise KbValidationError("missing frontmatter delimiter '---'")
for i in range(1, len(lines)):
if lines[i].strip() == "---":
meta = yaml.safe_load("\n".join(lines[1:i]))
body = "\n".join(lines[i + 1:])
break
else:
raise KbValidationError("unterminated frontmatter")
if not isinstance(meta, dict):
raise KbValidationError("frontmatter is not a mapping")
return meta, body
def split_sections(body: str) -> list[Section]:
"""H2-Sektionen als Chunks; H1-Titel und Quellzeile fallen weg."""
sections: list[Section] = []
current_title: str | None = None
current: list[str] = []
for line in body.splitlines():
if line.startswith("## "):
if current_title is not None:
sections.append(Section(current_title, "\n".join(current).strip()))
current_title = line[3:].strip()
current = []
elif line.startswith("# "):
continue
elif current_title is not None:
current.append(line)
if current_title is not None:
sections.append(Section(current_title, "\n".join(current).strip()))
return [s for s in sections if s.text]
def load_entry(path: Path) -> KBEntry:
meta, body = parse_frontmatter(path.read_text(encoding="utf-8"))
missing = [k for k in REQUIRED_KEYS if k not in meta]
if missing:
raise KbValidationError(f"{path.name}: missing frontmatter keys {missing}")
for key in ("id", "title", "work", "chapter", "topic", "author", "stand"):
if not isinstance(meta[key], str) or not meta[key].strip():
raise KbValidationError(f"{path.name}: empty '{key}'")
if not ID_RE.match(meta["id"]):
raise KbValidationError(f"{path.name}: invalid id '{meta['id']}'")
if not STAND_RE.match(meta["stand"]):
raise KbValidationError(f"{path.name}: stand '{meta['stand']}' not YYYY-MM")
if not isinstance(meta["source"], dict) or not {"pdf", "text"} <= set(meta["source"]):
raise KbValidationError(f"{path.name}: source needs pdf+text")
for key in ("legal_bases", "tags", "cross_refs"):
if not isinstance(meta[key], list):
raise KbValidationError(f"{path.name}: '{key}' must be a list")
entry = KBEntry(
id=meta["id"],
title=meta["title"],
work=meta["work"],
chapter=meta["chapter"],
topic=meta["topic"],
author=meta["author"],
stand=meta["stand"],
batch=int(meta["batch"]),
source=dict(meta["source"]),
legal_bases=list(meta["legal_bases"]),
tags=list(meta["tags"]),
cross_refs=list(meta["cross_refs"]),
path=path,
)
entry.sections = split_sections(body)
return entry
def load_kb(kb_dir: str | Path, verify_registry: bool = True) -> list[KBEntry]:
"""Lädt alle Layer-2-Einträge, prüft Cross-Ref-Integrität und das kb.json-Gate."""
root = Path(kb_dir)
doc_dir = root / "dokumente"
paths = sorted(doc_dir.glob("*.md"))
if not paths:
raise KbValidationError(f"no Layer-2 documents under {doc_dir}")
entries: dict[str, KBEntry] = {}
for p in paths:
e = load_entry(p)
if e.id in entries:
raise KbValidationError(
f"duplicate id {e.id}: {entries[e.id].path.name} and {p.name}"
)
entries[e.id] = e
for e in entries.values():
dangling = [ref for ref in e.cross_refs if ref not in entries]
if dangling:
raise KbValidationError(f"{e.id}: dangling cross_refs {dangling}")
if verify_registry:
_registry_gate(root, entries)
return list(entries.values())
def _registry_gate(root: Path, entries: dict[str, KBEntry]) -> None:
reg_path = root / "kb.json"
if not reg_path.is_file():
raise KbValidationError("kb.json missing — regenerate with --registry")
registry = json.loads(reg_path.read_text(encoding="utf-8"))
reg_ids = [e.get("id") for e in registry.get("entries", []) if e.get("id")]
reg_set = set(reg_ids)
kb_set = set(entries)
if reg_set != kb_set:
kb_only = sorted(kb_set - reg_set)[:5]
reg_only = sorted(reg_set - kb_set)[:5]
raise KbValidationError(
"kb.json out of sync with Layer 2 "
f"(docs-only: {kb_only}, registry-only: {reg_only}; "
f"registry n={registry.get('n_entries')}, docs n={len(entries)}) — "
"regenerate the registry first (build_lexis_kb.py --registry)"
)
agent/normalize.py
+43
View File
@@ -0,0 +1,43 @@
"""Deutsch-affine Textnormalisierung für FTS5-Index und -Anfragen.
Konvention (einmalig, konsistent): lowercase, Diakritika via NFKD entfernen
(ä→a, ü→u), ß→ss. Dies gilt für die FTS-Spalte `norm` und die Query gleich.
Die ASCII-Slug-Konvention der Wissensbasis (Umlaute "fallen") betrifft nur
`topic`/`tags`/Dateinamen, nicht die Volltextsuche.
"""
from __future__ import annotations
import re
import unicodedata
# Stopwörter in normalisierter Form (nach Fold), klein halten.
STOPWORDS = frozenset(
"""der die das und oder ein eine einen einem einer eines den dem des ist im in
von fur mit auf zu zum zur an am beim wie was wann wird werden kann muss
sind hat nicht man als auch aus bei sein ihre ihr es sie er doch noch nur
schon sehr mehr hier da durch fuer wird""".split()
)
_TOKEN_RE = re.compile(r"[a-z0-9]+")
def normalize_text(s: str) -> str:
"""Lowercase, Diakritika entfernen (NFKD), ß→ss."""
s = unicodedata.normalize("NFKD", s.casefold())
s = "".join(c for c in s if unicodedata.category(c) != "Mn")
return s.replace("ß", "ss")
def tokenize(s: str) -> list[str]:
return _TOKEN_RE.findall(normalize_text(s))
def fts_query(question: str, min_len: int = 2) -> str:
"""OR-verknüpfte FTS5-Phrasen aus normalisierten Termen; '' wenn leer."""
terms: list[str] = []
seen: set[str] = set()
for t in tokenize(question):
if len(t) >= min_len and t not in STOPWORDS and t not in seen:
seen.add(t)
terms.append(t)
return " OR ".join(f'"{t}"' for t in terms)
agent/ollama_client.py
+108
View File
@@ -0,0 +1,108 @@
"""Schmaler Ollama-HTTP-Client (Chat + Embeddings).
Keine Tools, kein Browsing, keine Web-Hooks — der Agent hat architektonisch
keinen Weg aus der Wissensbasis hinaus (Grounding-Regel 1).
"""
from __future__ import annotations
from typing import Any
import httpx
class OllamaError(RuntimeError):
pass
class OllamaClient:
def __init__(self, base_url: str, embed_timeout_s: float = 240.0,
chat_timeout_s: float = 300.0):
self.base_url = base_url.rstrip("/")
self.embed_timeout_s = embed_timeout_s
self.chat_timeout_s = chat_timeout_s
# Per-Request-Timeouts mit kurzem Connect-Budget — ein unerreichbarer
# Server muss in Sekunden, nicht Minuten scheitern.
self.embed_timeout = httpx.Timeout(embed_timeout_s, connect=10.0)
self.chat_timeout = httpx.Timeout(chat_timeout_s, connect=10.0)
self._client = httpx.Client(timeout=self.chat_timeout)
def close(self) -> None:
self._client.close()
# -- Infrastruktur ------------------------------------------------------
def is_up(self) -> bool:
try:
r = self._client.get(f"{self.base_url}/api/tags", timeout=5.0)
return r.status_code == 200
except httpx.HTTPError:
return False
def list_models(self) -> list[str]:
r = self._client.get(f"{self.base_url}/api/tags")
r.raise_for_status()
return [m.get("model", "") for m in r.json().get("models", [])]
# -- Embeddings ---------------------------------------------------------
def embed(self, model: str, texts: list[str]) -> list[list[float]]:
"""Batch-Embeddings via /api/embed; Fallback auf Legacy /api/embeddings."""
r = self._client.post(
f"{self.base_url}/api/embed",
json={"model": model, "input": texts},
timeout=self.embed_timeout,
)
if r.status_code == 404:
out: list[list[float]] = []
for t in texts:
rr = self._client.post(
f"{self.base_url}/api/embeddings",
json={"model": model, "prompt": t},
timeout=self.embed_timeout,
)
rr.raise_for_status()
emb = rr.json().get("embedding")
if not emb:
raise OllamaError("legacy /api/embeddings returned no embedding")
out.append(emb)
return out
r.raise_for_status()
emb = r.json().get("embeddings")
if not isinstance(emb, list) or len(emb) != len(texts):
raise OllamaError(f"unexpected /api/embed response for {len(texts)} inputs")
return emb
# -- Chat ---------------------------------------------------------------
def chat(self, model: str, messages: list[dict],
temperature: float = 0.1, num_ctx: int = 16384,
num_predict: int = 1024, think: bool = False) -> str:
"""POST /api/chat, stream=False; `think`-Flag mit 404/400-Fallback."""
body: dict[str, Any] = {
"model": model,
"messages": messages,
"stream": False,
"options": {
"temperature": temperature,
"num_ctx": num_ctx,
"num_predict": num_predict,
},
}
if think is not None:
body["think"] = bool(think)
r = self._client.post(
f"{self.base_url}/api/chat", json=body, timeout=self.chat_timeout
)
if r.status_code in (400, 404) and "think" in body:
# Ältere Ollama-Versionen kennen das think-Flag nicht -> Retry ohne.
body.pop("think")
r = self._client.post(
f"{self.base_url}/api/chat", json=body, timeout=self.chat_timeout
)
r.raise_for_status()
data = r.json()
msg = data.get("message") or {}
content = msg.get("content") or ""
if not content.strip():
raise OllamaError(f"empty response from {model} (keys: {list(data.keys())})")
return content
agent/retrieve.py
+252
View File
@@ -0,0 +1,252 @@
"""Hybrid-Retrieval: BM25 (FTS5) + Dense (bge-m3) -> RRF-Fusion,
milde Stand-Aktualitätsgewichtung und kontrollierte cross_ref-Erweiterung.
"""
from __future__ import annotations
import json
import sqlite3
from dataclasses import dataclass, field
from pathlib import Path
import numpy as np
from .config import Config
from .normalize import fts_query
from .ollama_client import OllamaClient
@dataclass
class ChunkResult:
chunk_id: int
entry_id: str
section: str
text: str
title: str
stand: str
work: str
chapter: str
topic: str
tags: list = field(default_factory=list)
legal_bases: list = field(default_factory=list)
cross_refs: list = field(default_factory=list)
batch: int = 0
score: float = 0.0
source: str = "fused" # bm25 | dense | fused | cross_ref
class Retriever:
def __init__(self, cfg: Config, db_path: str | None = None, client=None):
self.cfg = cfg
self.db_path = str(db_path or cfg.db_path)
if not Path(self.db_path).is_file():
raise RuntimeError(
f"index fehlt ({self.db_path}) — zuerst 'python -m agent.cli ingest' ausführen"
)
self._con = sqlite3.connect(self.db_path)
self._con.row_factory = sqlite3.Row
self._client = client
self._owns_client = client is None
self._mat: np.ndarray | None = None
self._mat_chunk_ids: list[int] | None = None
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("-", ""))
def close(self) -> None:
self._con.close()
if self._owns_client and self._client is not None:
self._client.close()
# -- Index-Kennzahlen ---------------------------------------------------
def stats(self) -> dict:
n_chunks = self._con.execute("SELECT COUNT(*) FROM chunks").fetchone()[0]
n_entries = self._con.execute(
"SELECT COUNT(DISTINCT entry_id) FROM chunks"
).fetchone()[0]
n_vec = self._con.execute(
"SELECT COUNT(*) FROM vectors WHERE model = ?",
(self.cfg.embed_model,),
).fetchone()[0]
meta = dict(self._con.execute("SELECT key, value FROM meta").fetchall())
return {
"n_entries": n_entries,
"n_chunks": n_chunks,
"n_vectors": n_vec,
"dense_available": n_vec > 0 and not self.cfg.embed_off,
"stand_min": str(self._stand_min),
"stand_max": str(self._stand_max),
"built_at": meta.get("built_at"),
}
# -- Einzelverfahren ----------------------------------------------------
def _bm25(self, question: str, limit: int) -> dict[int, float]:
q = fts_query(question)
if not q:
return {}
rows = self._con.execute(
"SELECT rowid, bm25(chunks_fts) AS rank FROM chunks_fts "
"WHERE chunks_fts MATCH ? ORDER BY rank LIMIT ?",
(q, limit),
).fetchall()
# 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]:
if self.cfg.embed_off:
return {}
self._ensure_matrix()
if self._mat is None or len(self._mat) == 0:
return {}
if self._client is None:
self._client = OllamaClient(
self.cfg.ollama_url,
embed_timeout_s=self.cfg.embed_timeout_s,
chat_timeout_s=self.cfg.chat_timeout_s,
)
try:
qvec = np.asarray(
self._client.embed(self.cfg.embed_model, [question])[0],
dtype=np.float32,
)
except Exception:
return {} # Ollama nicht erreichbar -> BM25-only weiter
qn = np.linalg.norm(qvec)
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}
def _ensure_matrix(self) -> None:
if self._mat is not None:
return
rows = self._con.execute(
"SELECT c.chunk_id, v.vec, v.dim FROM chunks c "
"JOIN vectors v ON v.content_hash = c.content_hash AND v.model = ?",
(self.cfg.embed_model,),
).fetchall()
if not rows:
self._mat = np.zeros((0, 1), dtype=np.float32)
self._mat_chunk_ids = []
return
ids = [r[0] for r in rows]
mat = np.vstack(
[np.frombuffer(r[1], dtype=np.float32) for r in rows]
)
norms = np.linalg.norm(mat, axis=1, keepdims=True)
self._mat = mat
self._mat_norm = mat / np.where(norms == 0, 1.0, norms)
self._mat_chunk_ids = ids
# -- Metadaten & Fusion -------------------------------------------------
def _stand_factor(self, stand: str) -> float:
if self._stand_max <= self._stand_min:
return 0.0
try:
s = int(stand.replace("-", ""))
except (ValueError, AttributeError):
return 0.0
f = (s - self._stand_min) / (self._stand_max - self._stand_min)
return min(1.0, max(0.0, f))
def _fetch_chunks(self, chunk_ids: list[int]) -> dict[int, sqlite3.Row]:
out: dict[int, sqlite3.Row] = {}
for i in range(0, len(chunk_ids), 500):
part = chunk_ids[i:i + 500]
qm = ",".join("?" * len(part))
for r in self._con.execute(
f"SELECT * FROM chunks WHERE chunk_id IN ({qm})", part
).fetchall():
out[r["chunk_id"]] = r
return out
def _row_to_result(self, row: sqlite3.Row, score: float, source: str) -> ChunkResult:
return ChunkResult(
chunk_id=row["chunk_id"],
entry_id=row["entry_id"],
section=row["section"],
text=row["text"],
title=row["title"],
stand=row["stand"],
work=row["work"],
chapter=row["chapter"],
topic=row["topic"],
tags=json.loads(row["tags"]),
legal_bases=json.loads(row["legal_bases"]),
cross_refs=json.loads(row["cross_refs"]),
batch=row["batch"],
score=score,
source=source,
)
def _best_chunk_of_entry(self, entry_id: str) -> ChunkResult | None:
rows = self._con.execute(
"SELECT * FROM chunks WHERE entry_id = ? "
"ORDER BY CASE WHEN section LIKE 'Zusammenfassung%' THEN 0 ELSE 1 END, "
"chunk_id LIMIT 1",
(entry_id,),
).fetchall()
if not rows:
return None
return self._row_to_result(rows[0], 0.0, "cross_ref")
# -- ö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)
try:
dn = self._dense(question, pool)
except Exception:
dn = {}
fused: dict[int, float] = {}
for ranking in (bm, dn):
ordered = sorted(ranking.items(), key=lambda kv: -kv[1])
for rank, (cid, _) in enumerate(ordered):
fused[cid] = fused.get(cid, 0.0) + 1.0 / (self.cfg.rrf_k + rank)
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"
)
score += self.cfg.recency_boost * self._stand_factor(rows[cid]["stand"])
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)
main: list[ChunkResult] = []
seen: set[str] = set()
for r in results:
if r.entry_id in seen:
continue
seen.add(r.entry_id)
main.append(r)
if len(main) >= n:
break
# cross_ref-Erweiterung (kontrolliert, markiert, begrenzt)
extra: list[ChunkResult] = []
budget = self.cfg.cross_ref_max_extra
for r in main[: self.cfg.cross_ref_expand]:
for ref in r.cross_refs:
if budget <= 0:
break
if ref in seen:
continue
er = self._best_chunk_of_entry(ref)
if er is not None:
extra.append(er)
seen.add(ref)
budget -= 1
return main + extra
planung.md
+16
View File
@@ -230,3 +230,19 @@ tests/ # pytest: Ingest-, Retrieval-, Grounding-Unit-Tests
3. **Layer 1:** bewusst außen vor in Phase A — einverstanden? 3. **Layer 1:** bewusst außen vor in Phase A — einverstanden?
4. **Odoo-Version für Phase B:** Odoo 19 Enterprise (passend zu 4. **Odoo-Version für Phase B:** Odoo 19 Enterprise (passend zu
`l10n_at_hr_payroll*`) — bitte bestätigen. `l10n_at_hr_payroll*`) — bitte bestätigen.
## 13. Umsetzungsstand (2026-09-14)
- **M1 erledigt (offline):** Ingest + Index (601 Einträge → 3.005 Chunks,
FTS5-BM25) + Hybrid-Retrieval (Dense-Code vorhanden, am Host zu messen).
Goldset 31 Fragen (IDs gegen kb.json verifiziert, inkl. ATZ-Konfliktfall
und 4 Verweigerungsfälle). Baseline BM25-only: Hit-Rate 0,871 ·
Recall@8 0,855 · MRR 0,476 — 4 Fehltreffer sind Komposita-/Stamm-
Muster, die die Dense-Suche abdecken soll. 41 Unit-Tests grün.
- **M2 implementiert:** Ollama-Client (embed/chat, think-Fallback),
Systemprompt mit Zitierpflicht, Post-Validierung (1× Regenerierung, dann
Verweigerung), `/ask`-API + CLI + Test-Chat. **Host-Validierung mit
echtem Ollama noch offen** (aus der Zed-Sandbox nicht erreichbar).
- **M3 offen:** Bake-off auf dem Host (qwen3.8:27b vs. qwen3:32b vs.
gemma3:27b vs. mistral-small3.2:24b; qwen3:14b als Latenz-Untergrenze).
- Betrieb/Host-Schritte: `agent/README.md`.
pyproject.toml
+3
View File
@@ -0,0 +1,3 @@
[tool.pytest.ini_options]
pythonpath = ["."]
testpaths = ["tests"]
requirements.txt
+6
View File
@@ -0,0 +1,6 @@
fastapi>=0.115
uvicorn>=0.30
httpx>=0.27
PyYAML>=6.0
numpy>=2.0
pytest>=8.0
tests/conftest.py
+200
View File
@@ -0,0 +1,200 @@
"""Fixtures: Mini-Wissensbasis (3 Einträge), Index, Fake-Ollama-Client.
Alle Tests laufen offline — kein Ollama, kein Netz. Die Mini-KB folgt dem
verbindlichen Layer-2-Schema (Frontmatter + H2-Sektionen + kb.json-Gate).
"""
from __future__ import annotations
import json
from pathlib import Path
import pytest
from agent.config import Config
DOC_ATZ = """---
id: lb-min-01
batch: 1
title: "Altersteilzeit - Überblick"
work: "Lexis Briefings Personalrecht"
chapter: "Beschäftigungsverhältnisse"
topic: altersteilzeit
author: "Marek"
stand: 2026-01
source:
pdf: ".lexis360/Lexis360_test_atz.pdf"
text: ".lexis360/md/test_atz.md"
legal_bases: ["AlVG", "AZG § 19e"]
tags: [altersteilzeit, lohnausgleich]
cross_refs: ["lb-min-02"]
---
# Altersteilzeit – Überblick
*Lexis Briefings Personalrecht, Marek, Stand Jänner 2026 (lb-min-01).*
## Zusammenfassung
Altersteilzeit ist eine Teilzeit, bei der der Arbeitnehmer zusätzlich zum
Teilzeitentgelt einen Lohnausgleich erhält; das AMS ersetzt dem Arbeitgeber
einen Teil der Zusatzkosten (Altersteilzeitgeld).
## Kernwerte & Fristen (Stand 2026-01)
| Wert / Regel | Detail |
|---|---|
| Ersatzquote | 28,5 % der ersetzbaren Kosten (Stand 2026-01) |
## Rechtsgrundlagen
- AZG § 19e (Kündigungszuschlag)
## Verweise
- lb-min-02
"""
DOC_URL = """---
id: lb-min-02
batch: 1
title: "Urlaubsanspruch und Verbrauch"
work: "Lexis Briefings Personalrecht"
chapter: "Urlaub & Karenzierung"
topic: urlaub
author: "Marek"
stand: 2026-07
source:
pdf: ".lexis360/Lexis360_test_urlaub.pdf"
text: ".lexis360/md/test_urlaub.md"
legal_bases: ["UrlG"]
tags: [urlaub, urlaubsentgelt]
cross_refs: ["lb-min-01"]
---
# Urlaubsanspruch und Verbrauch
*Lexis Briefings Personalrecht, Marek, Stand Juli 2026 (lb-min-02).*
## Zusammenfassung
Der gesetzliche Urlaubsanspruch beträgt fünf Werktage Wochenurlaub je
Dienstjahr; das Urlaubsentgelt ist wie Entgeltfortzahlung zu berechnen.
## Kernwerte & Fristen (Stand 2026-07)
| Wert / Regel | Detail |
|---|---|
| Mindesturlaub | 5 Wochen je Dienstjahr (Stand 2026-07) |
## Verweise
- lb-min-01
"""
DOC_WIKU = """---
id: wk-min-01
batch: 1
title: "WIKU Praxisfall: Lohnverrechnung"
work: "WIKU Fachbroschüre"
chapter: "WIKU Fachbroschüre"
topic: lohnverrechnung
author: "Wilhelm Kurzböck"
stand: 2026-03
source:
pdf: ".wiku/test_lohnverrechnung.pdf"
text: ".wiku/md/test_lohnverrechnung.md"
legal_bases: ["EStG § 25"]
tags: [lohnverrechnung, praxisfall]
cross_refs: ["lb-min-01"]
---
# WIKU Praxisfall: Lohnverrechnung
*WIKU Fachbroschüre, Wilhelm Kurzböck, Stand März 2026 (wk-min-01).*
## Zusammenfassung
Praxisfall zur laufenden Lohnverrechnung: Abrechnungsperiode und
Beitragsgrundlagen sind monatlich festzulegen.
## Verweise
- lb-min-01
"""
MIN_DOCS = {
"altersteilzeit_uberblick.md": DOC_ATZ,
"urlaubsanspruch.md": DOC_URL,
"wiku_lohnverrechnung.md": DOC_WIKU,
}
MIN_IDS = ["lb-min-01", "lb-min-02", "wk-min-01"]
def write_mini_kb(root: Path) -> Path:
doc = root / "dokumente"
doc.mkdir(parents=True, exist_ok=True)
for name, content in MIN_DOCS.items():
(doc / name).write_text(content, encoding="utf-8")
kb = {
"n_entries": len(MIN_IDS),
"entries": [{"id": i, "title": i} for i in MIN_IDS],
}
(root / "kb.json").write_text(
json.dumps(kb, ensure_ascii=False, indent=1), encoding="utf-8"
)
return root
class FakeOllama:
"""Skriptbarer Chat-Client; embed() schlägt fehl (Dense bleibt ungetestet)."""
def __init__(self, answers: list[str] | None = None):
self.answers = list(answers or [])
self.calls = 0
self.last_messages: list | None = None
def chat(self, model, messages, **kwargs):
self.calls += 1
self.last_messages = messages
if not self.answers:
raise AssertionError("FakeOllama: keine skriptierte Antwort mehr")
return self.answers.pop(0)
def embed(self, model, texts):
raise RuntimeError("embed nicht verfügbar (offline Test)")
def is_up(self):
return False
def close(self):
pass
@pytest.fixture
def mini_kb(tmp_path):
return write_mini_kb(tmp_path / "kb")
@pytest.fixture
def mini_cfg(tmp_path, mini_kb) -> Config:
return Config(
kb_dir=str(mini_kb),
db_path=str(tmp_path / "index.db"),
embed_off=True,
)
@pytest.fixture
def mini_index(mini_cfg) -> Config:
from agent.ingest import build_index
stats = build_index(mini_cfg)
assert stats.n_entries == 3
assert stats.n_chunks >= 7
assert stats.embed_error is None
return mini_cfg
@pytest.fixture
def fake_ollama():
return FakeOllama
tests/test_generate.py
+146
View File
@@ -0,0 +1,146 @@
"""Tests: Grounding — Post-Validierung, Verweigerungspflicht, Regenerierung.
Der kritische Teil der Pipeline: keine Antwort mit ungültigen Zitaten
verlässt answer_question.
"""
import pytest
from agent.generate import (
REFUSAL_MESSAGE,
UNCERTAIN_MESSAGE,
answer_question,
build_user_content,
looks_like_refusal,
strip_think,
validate_answer,
)
class TestValidateAnswer:
def test_valid_citation_passes(self):
assert validate_answer("ATZ ist Teilzeit [lb-min-01].", ["lb-min-01"]) == []
def test_unknown_id_is_violation(self):
v = validate_answer("ATZ ist xyz [lb-atz-99].", ["lb-min-01"])
assert any("lb-atz-99" in x for x in v)
def test_no_citation_is_violation(self):
v = validate_answer("ATZ ist eine Teilzeit.", ["lb-min-01"])
assert any("keine KB-ID" in x for x in v)
def test_refusal_without_citation_is_ok(self):
assert validate_answer(REFUSAL_MESSAGE, ["lb-min-01"]) == []
def test_bare_id_mention_is_detected(self):
assert validate_answer("Siehe lb-min-02 für Details.", ["lb-min-02"]) == []
class TestRefusalDetection:
def test_refusal_phrase(self):
assert looks_like_refusal("Dazu enthält die Wissensbasis keine Aussage.")
def test_refusal_phrase_with_typos_folds(self):
assert looks_like_refusal("Dazu enthält die Wissensbasis keine Aussage!")
def test_normal_answer_is_no_refusal(self):
assert not looks_like_refusal("Der Anspruch besteht [lb-min-01].")
def test_strip_think_removes_block(self):
open_tag = "<" + "think" + ">"
close_tag = "</" + "think" + ">"
text = open_tag + "Reasoning here" + close_tag + "Antwort [lb-min-01]."
out = strip_think(text)
assert "Reasoning" not in out
assert out.strip().startswith("Antwort [lb-min-01].")
class TestUserContent:
def test_blocks_contain_metadata_header(self, mini_index):
from agent.retrieve import Retriever
r = Retriever(mini_index)
try:
results = r.search("Altersteilzeit Lohnausgleich")
content = build_user_content("Was ist ATZ?", results)
assert "Block 1 — [lb-min-01]" in content
assert "Stand: 2026-01" in content
assert "Frage: Was ist ATZ?" in content
finally:
r.close()
class TestAnswerQuestion:
def test_happy_path_verified(self, mini_index, fake_ollama):
client = fake_ollama(
answers=["Altersteilzeit ist eine Teilzeit mit Lohnausgleich "
"[lb-min-01]. (Stand 2026-01)"]
)
result = answer_question(
"Was ist Altersteilzeit?", mini_index, client=client
)
assert result["verified"] is True
assert result["refused"] is False
assert result["citations"] == ["lb-min-01"]
assert result["sources"][0]["id"] == "lb-min-01"
assert result["sources"][0]["stand"] == "2026-01"
assert client.calls == 1
def test_hallucinated_id_regenerates_then_refuses(self, mini_index, fake_ollama):
client = fake_ollama(answers=[
"ATZ gilt ab 60. Lebensjahr [lb-atz-99].",
"ATZ gilt ab 60. Lebensjahr, siehe [lb-atz-99].",
])
result = answer_question(
"Was ist Altersteilzeit?", mini_index, client=client
)
assert result["refused"] is True
assert result["verified"] is False
assert result["answer"] == UNCERTAIN_MESSAGE
assert result["regenerations"] == 1
assert result["citations"] == []
assert "draft" in result and "lb-atz-99" in result["draft"]
def test_regeneration_can_recover(self, mini_index, fake_ollama):
client = fake_ollama(answers=[
"ATZ gilt ab 60 [lb-atz-99].",
"ATZ ist Teilzeit mit Lohnausgleich [lb-min-01].",
])
result = answer_question(
"Was ist Altersteilzeit?", mini_index, client=client
)
assert result["verified"] is True
assert result["regenerations"] == 1
assert result["citations"] == ["lb-min-01"]
def test_empty_retrieval_refuses_deterministically(self, mini_index, fake_ollama):
client = fake_ollama(answers=["sollte nie aufgerufen werden"])
result = answer_question(
"Wie hoch ist der Wechselkurs von Bermuda-Dollar?", mini_index,
client=client,
)
assert result["refused"] is True
assert result["answer"] == REFUSAL_MESSAGE
assert result["verified"] is True
assert client.calls == 0 # kein LLM-Call bei leerem Retrieval
def test_model_refusal_is_kept(self, mini_index, fake_ollama):
client = fake_ollama(answers=[
f"Zu dieser Frage: {REFUSAL_MESSAGE}"
])
result = answer_question(
"Was ist Altersteilzeit?", mini_index, client=client
)
assert result["refused"] is True
assert result["verified"] is True # Regel-4-konforme Verweigerung
assert client.calls == 1
def test_top_k_limits_context(self, mini_index, fake_ollama):
client = fake_ollama(answers=["Teilzeit [lb-min-01]."])
result = answer_question(
"Altersteilzeit Urlaub Lohnverrechnung", mini_index,
client=client, top_k=1,
)
main = [s for s in result["sources"]]
assert result["n_context"] >= 1
# Haupt-Blöcke auf top_k begrenzt; cross_ref-Erweiterungen dürfen dazu
assert len([s for s in main]) <= result["n_context"]
tests/test_ingest.py
+99
View File
@@ -0,0 +1,99 @@
"""Tests: Index-Bau (Chunking, FTS, Metadaten, Schema-Gates)."""
import sqlite3
import pytest
from agent.ingest import SCHEMA, build_index
from agent.kb import KbValidationError
def test_build_index_chunks_and_fts(mini_index):
con = sqlite3.connect(mini_index.db_path)
try:
n_chunks = con.execute("SELECT COUNT(*) FROM chunks").fetchone()[0]
n_fts = con.execute("SELECT COUNT(*) FROM chunks_fts").fetchone()[0]
n_entries = con.execute(
"SELECT COUNT(DISTINCT entry_id) FROM chunks"
).fetchone()[0]
assert n_entries == 3
assert n_chunks == n_fts and n_chunks >= 7
row = con.execute(
"SELECT entry_id, section, norm FROM chunks WHERE entry_id='lb-min-01' "
"AND section LIKE 'Kernwerte%'"
).fetchone()
assert row is not None
# Umlaut-Folding im FTS-Text: "Lohnausgleich" normalisiert auffindbar
assert "lohnausgleich" in row[2]
meta = dict(con.execute("SELECT key, value FROM meta").fetchall())
assert meta["n_entries"] == "3"
assert meta["embed_model"] == "" # embed_off=True
finally:
con.close()
def test_norm_contains_tags_and_legal_bases(mini_index):
con = sqlite3.connect(mini_index.db_path)
try:
norm = con.execute(
"SELECT norm FROM chunks WHERE entry_id='lb-min-01' "
"AND section='Zusammenfassung'"
).fetchone()[0]
assert "alvg" in norm # legal_bases im FTS-Text
assert "azg" in norm and "19e" in norm
finally:
con.close()
def test_rebuild_is_idempotent(mini_index):
stats = build_index(mini_index)
assert stats.n_entries == 3
con = sqlite3.connect(mini_index.db_path)
try:
assert con.execute("SELECT COUNT(*) FROM chunks").fetchone()[0] == stats.n_chunks
finally:
con.close()
def test_build_index_aborts_on_gate_error(mini_cfg):
"""Gate-Fehler (Registry kaputt) bricht den Ingest ab — kein halber Index."""
from pathlib import Path
(Path(mini_cfg.kb_dir) / "kb.json").write_text(
'{"n_entries": 0, "entries": []}', encoding="utf-8"
)
with pytest.raises(KbValidationError):
build_index(mini_cfg)
def test_vectors_table_cached_across_rebuilds(mini_index):
"""Die Vektoren-Tabelle bleibt beim Rebuild erhalten (Cache-Garantie)."""
con = sqlite3.connect(mini_index.db_path)
try:
con.execute(
"INSERT INTO vectors(content_hash, model, dim, vec) "
"VALUES ('deadbeef', 'bge-m3', 2, x'000000003f800000')"
) # 0.0, 1.0
con.commit()
finally:
con.close()
build_index(mini_index)
con = sqlite3.connect(mini_index.db_path)
try:
assert con.execute(
"SELECT COUNT(*) FROM vectors WHERE content_hash='deadbeef'"
).fetchone()[0] == 1
finally:
con.close()
def test_schema_creates_fts5(tmp_path):
con = sqlite3.connect(tmp_path / "s.db")
try:
con.executescript(SCHEMA)
con.execute("INSERT INTO chunks_fts(rowid, norm) VALUES (1, 'testtext')")
hits = con.execute(
"SELECT rowid FROM chunks_fts WHERE chunks_fts MATCH '\"testtext\"'"
).fetchall()
assert hits == [(1,)]
finally:
con.close()
tests/test_kb.py
+74
View File
@@ -0,0 +1,74 @@
"""Tests: Layer-2-Parsing, Sektionen, Cross-Ref-Integrität, kb.json-Gate."""
import json
from pathlib import Path
import pytest
from agent.kb import KbValidationError, load_entry, load_kb, parse_frontmatter, split_sections
from tests.conftest import DOC_ATZ, MIN_IDS, write_mini_kb
def test_parse_frontmatter_and_sections():
meta, body = parse_frontmatter(DOC_ATZ)
assert meta["id"] == "lb-min-01"
assert meta["stand"] == "2026-01"
sections = split_sections(body)
titles = [s.title for s in sections]
assert "Zusammenfassung" in titles
assert titles[1].startswith("Kernwerte & Fristen")
assert all(s.text for s in sections)
def test_split_sections_ignores_h1_and_intro():
sections = split_sections("# Titel\n\n*Quellzeile*\n\n## A\n\nText A\n\n## B\n\nText B")
assert [s.title for s in sections] == ["A", "B"]
assert sections[0].text == "Text A"
def test_load_entry_minimal_doc(tmp_path):
p = tmp_path / "doc.md"
p.write_text(DOC_ATZ, encoding="utf-8")
e = load_entry(p)
assert e.id == "lb-min-01"
assert e.tags == ["altersteilzeit", "lohnausgleich"]
assert e.cross_refs == ["lb-min-02"]
assert len(e.sections) >= 3
def test_gate_rejects_out_of_sync_registry(tmp_path):
root = write_mini_kb(tmp_path / "kb")
kb = json.loads((root / "kb.json").read_text(encoding="utf-8"))
kb["entries"].append({"id": "lb-min-99"})
(root / "kb.json").write_text(json.dumps(kb), encoding="utf-8")
with pytest.raises(KbValidationError, match="out of sync"):
load_kb(root)
def test_gate_rejects_dangling_cross_refs(tmp_path):
root = write_mini_kb(tmp_path / "kb")
doc = root / "dokumente" / "altersteilzeit_uberblick.md"
doc.write_text(
DOC_ATZ.replace('cross_refs: ["lb-min-02"]', 'cross_refs: ["lb-min-42"]'),
encoding="utf-8",
)
with pytest.raises(KbValidationError, match="dangling cross_refs"):
load_kb(root)
def test_gate_rejects_invalid_stand(tmp_path):
root = write_mini_kb(tmp_path / "kb")
doc = root / "dokumente" / "altersteilzeit_uberblick.md"
doc.write_text(DOC_ATZ.replace("stand: 2026-01", "stand: Jänner 2026"), encoding="utf-8")
with pytest.raises(KbValidationError, match="not YYYY-MM"):
load_kb(root)
def test_real_corpus_loads_and_matches_registry():
"""Integrationstest gegen die echte Wissensbasis (Gate inklusive)."""
entries = load_kb("wissensbasis", verify_registry=True)
ids = {e.id for e in entries}
assert len(entries) == 601
assert "lb-atz-07" in ids and "wk-akt-01" in ids
atz = [e for e in entries if e.id == "lb-atz-07"][0]
assert atz.topic == "altersteilzeit"
assert any(s.title.startswith("Kernwerte") for s in atz.sections)
tests/test_normalize.py
+29
View File
@@ -0,0 +1,29 @@
"""Tests: Textnormalisierung und FTS-Query-Bau."""
from agent.normalize import fts_query, normalize_text, tokenize
def test_normalize_folds_german_diacritics():
assert normalize_text("Gehälter Ärger Größe Übung Ökonomie") == (
"gehalter arger grosse ubung okonomie"
)
def test_normalize_keeps_digits_and_section_sign():
assert normalize_text("AZG § 19e (Stand 2026-01)") == "azg § 19e (stand 2026-01)"
def test_tokenize_splits_alphanumeric():
assert tokenize("Lohnausgleich, AZG §19e") == ["lohnausgleich", "azg", "19e"]
def test_fts_query_drops_stopwords_and_quotes_terms():
q = fts_query("Wie hoch ist die SV-Beitragsgrundlage?")
assert '"beitragsgrundlage"' in q
assert '"sv"' in q
assert '"wie"' not in q
assert '"ist"' not in q
def test_fts_query_empty_and_stopword_only():
assert fts_query("") == ""
assert fts_query("Wie ist der die das?") == ""
tests/test_retrieve.py
+82
View File
@@ -0,0 +1,82 @@
"""Tests: Hybrid-Retrieval (BM25-only offline): Fusion, Entry-Dedup,
cross_ref-Erweiterung, leeres Retrieval."""
import pytest
from agent.retrieve import Retriever
@pytest.fixture
def retriever(mini_index):
r = Retriever(mini_index)
yield r
r.close()
def test_search_finds_expected_entry(retriever):
results = retriever.search("Was ist Altersteilzeit und Lohnausgleich?")
assert results, "Retrieval sollte Treffer liefern"
assert results[0].entry_id == "lb-min-01"
assert results[0].stand == "2026-01"
# Beim Section-Schnitt entscheidet BM25-Längennormalisierung; hier zählt
# der richtige Eintrag, nicht der konkrete Abschnitt.
def test_search_cross_ref_expansion(retriever):
"""Top-Treffer lb-min-02 → cross_ref lb-min-01 wird als Erweiterung ergänzt."""
results = retriever.search("Urlaubsanspruch fünf Werktage")
main = [r for r in results if r.source != "cross_ref"]
extra = [r for r in results if r.source == "cross_ref"]
assert main and main[0].entry_id == "lb-min-02"
assert any(r.entry_id == "lb-min-01" for r in extra)
def test_search_no_match_returns_empty(retriever):
results = retriever.search("kanadische quellensteuer bermuda")
assert results == []
def test_search_dedups_entries(mini_index):
"""Pro Eintrag höchstens ein Haupt-Chunk im Kontext (Entry-Level-Dedup)."""
r = Retriever(mini_index)
try:
results = r.search("Lohnausgleich Urlaubsentgelt Lohnverrechnung",
n_entries=2)
main_ids = [x.entry_id for x in results if x.source != "cross_ref"]
assert len(main_ids) == len(set(main_ids))
assert len(main_ids) <= 2
finally:
r.close()
def test_recency_boost_prefers_newer_stand(retriever):
"""Milde Aktualitätsgewichtung: bei Gleichstand gewinnt der neuere Stand.
'Urlaubsanspruch' (2026-07) sollte vor 'Altersteilzeit' (2026-01)
landen, wenn beide im Kontext sind und der Query beide trifft.
"""
results = retriever.search("Urlaubsanspruch Altersteilzeit")
main = [r for r in results if r.source != "cross_ref"]
if {r.entry_id for r in main} >= {"lb-min-01", "lb-min-02"}:
# Beide im Kontext -> Reihenfolge prüfen ist nur bei Score-Nähe sinnvoll;
# hier reicht die Existenz-Annahme, der Boost ist bewusst minimal.
assert main[0].entry_id in {"lb-min-01", "lb-min-02"}
def test_stats_report(mini_index):
r = Retriever(mini_index)
try:
s = r.stats()
assert s["n_entries"] == 3
assert s["dense_available"] is False # embed_off=True
assert s["stand_min"] == "202601"
assert s["stand_max"] == "202607"
finally:
r.close()
def test_retriever_requires_index(tmp_path):
from agent.config import Config
cfg = Config(kb_dir="wissensbasis", db_path=str(tmp_path / "missing.db"))
with pytest.raises(RuntimeError, match="ingest"):
Retriever(cfg)
web/index.html
+69
View File
@@ -0,0 +1,69 @@
<!DOCTYPE html>
<html lang="de">
<head>
<meta charset="utf-8">
<meta name="viewport" content="width=device-width, initial-scale=1">
<title>PV RAG Agent — Test-Chat</title>
<style>
:root { color-scheme: light dark; }
body { font-family: system-ui, sans-serif; max-width: 780px; margin: 2rem auto; padding: 0 1rem; }
h1 { font-size: 1.2rem; }
#log { display: flex; flex-direction: column; gap: 0.8rem; margin: 1.5rem 0; }
.msg { border: 1px solid rgba(128,128,128,.35); border-radius: 8px; padding: .7rem .9rem; white-space: pre-wrap; }
.q { background: rgba(128,128,128,.12); }
.src { font-size: .78rem; margin-top: .4rem; display: flex; flex-wrap: wrap; gap: .3rem; }
.chip { border-radius: 999px; padding: .1rem .5rem; background: rgba(128,128,128,.15); }
.warn { border-color: #c08020; }
form { display: flex; gap: .5rem; }
input { flex: 1; padding: .5rem; border-radius: 6px; border: 1px solid rgba(128,128,128,.5); }
button { padding: .5rem 1rem; border-radius: 6px; cursor: pointer; }
.meta { font-size: .75rem; opacity: .7; }
</style>
</head>
<body>
<h1>PV RAG Agent — Wissensbasis Personalverrechnung</h1>
<p class="meta">Antworten ausschließlich aus der kuratierten Wissensbasis, mit ID- und Stand-Beleg.</p>
<div id="log"></div>
<form id="f">
<input id="q" placeholder="Frage zur österreichischen Personalverrechnung …" autocomplete="off" required>
<button>Senden</button>
</form>
<script>
const log = document.getElementById("log");
function add(cls, html) {
const d = document.createElement("div");
d.className = "msg " + cls;
d.innerHTML = html;
log.appendChild(d);
}
document.getElementById("f").addEventListener("submit", async (e) => {
e.preventDefault();
const q = document.getElementById("q").value.trim();
if (!q) return;
add("q", q);
document.getElementById("q").value = "";
try {
const r = await fetch("/ask", {
method: "POST",
headers: { "Content-Type": "application/json" },
body: JSON.stringify({ question: q }),
});
if (!r.ok) {
add("warn", "Fehler " + r.status + ": " + (await r.text()));
return;
}
const d = await r.json();
const chips = (d.sources || []).map(s =>
`<span class="chip">${s.id} · ${s.stand || ""}</span>`).join("");
add(d.refused ? "warn" : "",
`${d.answer.replace(/</g, "&lt;")}` +
(d.sources && d.sources.length ? `<div class="src">${chips}</div>` : "") +
`<div class="meta">Modell: ${d.model} · ${d.latency_ms} ms · ` +
`${d.verified ? "zitiergeprüft" : "UNVERIFIZIERT"}${d.refused ? " · verweigert" : ""}</div>`);
} catch (err) {
add("warn", "Netzwerkfehler: " + err);
}
});
</script>
</body>
</html>