Initial production-ready EBM Mentor

.github/workflows/ci.yml

@@ -0,0 +1,21 @@
+name: CI
+
+on:
+  push:
+  pull_request:
+
+jobs:
+  test:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v4
+      - uses: actions/setup-python@v5
+        with:
+          python-version: "3.11"
+      - name: Install dependencies
+        run: |
+          python -m pip install --upgrade pip
+          pip install -r requirements.txt
+      - name: Run tests
+        run: pytest
+

.gitignore

@@ -0,0 +1,20 @@
+__pycache__/
+.pytest_cache/
+.mypy_cache/
+.ruff_cache/
+.venv/
+venv/
+env/
+dist/
+build/
+*.pyc
+*.pyo
+*.pyd
+*.egg-info/
+data/vector_store/
+data/*.parquet
+data/*.jsonl
+data/*.npy
+data/*.faiss
+*.log
+

Dockerfile

@@ -0,0 +1,22 @@
+FROM python:3.11-slim
+
+ENV PYTHONDONTWRITEBYTECODE=1 \
+    PYTHONUNBUFFERED=1 \
+    PIP_NO_CACHE_DIR=1
+
+WORKDIR /app
+
+RUN apt-get update && apt-get install -y --no-install-recommends \
+    build-essential \
+    git \
+    && rm -rf /var/lib/apt/lists/*
+
+COPY requirements.txt /app/requirements.txt
+RUN pip install --upgrade pip && pip install -r requirements.txt
+
+COPY . /app
+
+EXPOSE 7860
+
+CMD ["python", "app.py"]
+

README.md

@@ -0,0 +1,162 @@
+---
+title: EBM Mentor
+emoji: 🩺
+colorFrom: teal
+colorTo: blue
+sdk: gradio
+app_file: app.py
+pinned: false
+---
+
+# EBM Mentor
+
+EBM Mentor is a Retrieval-Augmented Generation assistant for German EBM billing research. It helps physicians, coders, and practice staff search and understand EBM code descriptions, points, exclusions, notes, and eligibility details.
+
+The system is RAG-only:
+
+- `CohereLabs/tiny-aya-water` is used only for answer generation.
+- All EBM knowledge comes from retrieval over locally indexed EBM XML data.
+- If the answer is not in the retrieved context, the app must say:
+  `Diese Information ist nicht in den bereitgestellten EBM-Daten enthalten.`
+
+## Architecture
+
+```mermaid
+flowchart LR
+    A[Official EBM XML] --> B[src/parser.py]
+    B --> C[Structured records]
+    C --> D[src/chunking.py]
+    D --> E[Searchable documents]
+    E --> F[src/embeddings.py]
+    F --> G[FAISS index]
+    G --> H[src/retriever.py]
+    H --> I[src/prompts.py]
+    I --> J[src/model.py]
+    J --> K[Gradio app]
+```
+
+## Features
+
+- Free-form EBM question answering
+- Code explanation with structured metadata
+- Random quiz mode
+- Search and chapter browsing
+- Source citations and retrieved document viewer
+- Confidence score for retrieval results
+
+## Repository Layout
+
+```text
+ebm-rag-trainer/
+├── app.py
+├── requirements.txt
+├── README.md
+├── Dockerfile
+├── data/
+│   └── ebm.xml
+├── src/
+│   ├── parser.py
+│   ├── chunking.py
+│   ├── embeddings.py
+│   ├── vector_store.py
+│   ├── retriever.py
+│   ├── prompts.py
+│   ├── rag_pipeline.py
+│   └── model.py
+├── scripts/
+│   ├── build_database.py
+│   └── ingest_ebm.py
+├── tests/
+└── .github/
+```
+
+## Setup
+
+### 1. Create an environment
+
+```bash
+python -m venv .venv
+.venv\Scripts\activate
+```
+
+### 2. Install dependencies
+
+```bash
+pip install -r requirements.txt
+```
+
+### 3. Add the official EBM XML
+
+Place the official XML file at:
+
+```text
+data/ebm.xml
+```
+
+The repository includes a tiny demo XML so the codebase is runnable out of the box, but production use should replace it with the official current EBM source.
+
+### 4. Build the local database
+
+```bash
+python scripts/ingest_ebm.py --xml data/ebm.xml --output data/processed
+python scripts/build_database.py --xml data/ebm.xml --store data/vector_store
+```
+
+### 5. Run the app
+
+```bash
+python app.py
+```
+
+## Hugging Face Spaces Deployment
+
+1. Create a new Hugging Face Space.
+2. Choose `Gradio` as the SDK.
+3. Upload this repository.
+4. Ensure `data/ebm.xml` is included or mounted in the Space.
+5. Let the Space build the local FAISS index on first launch or prebuild it with the scripts above.
+
+### Space Notes
+
+- SDK: `gradio`
+- Hardware: `CPU Basic / Zero`
+- No external database
+- No paid APIs
+- Local FAISS index and local metadata files only
+
+## Example Screenshots
+
+Add screenshots here after deployment:
+
+```text
+docs/screenshots/ask-ebm.png
+docs/screenshots/explain-code.png
+docs/screenshots/quiz-mode.png
+docs/screenshots/explore-ebm.png
+```
+
+## Sample Questions
+
+- Was bedeutet EBM-Code 01100?
+- Welche Punkte hat die Leistung?
+- Gibt es Ausschlüsse für diesen Code?
+- Welche Fachgruppen sind berechtigt?
+- Welche Hinweise stehen in den Anmerkungen?
+
+## Testing
+
+Run the local test suite:
+
+```bash
+pytest
+```
+
+The tests cover:
+
+- XML parsing
+- document chunking
+- retrieval
+
+## Safety and Scope
+
+This project is designed as a research assistant, not a legal authority. Always verify billing decisions against the official EBM source material and current practice guidance.

app.py

@@ -0,0 +1,249 @@
+from __future__ import annotations
+
+from pathlib import Path
+
+import gradio as gr
+import pandas as pd
+
+from src.parser import parse_ebm_xml_to_dataframe
+from src.rag_pipeline import EbmRAGPipeline, build_pipeline_from_paths
+
+
+ROOT = Path(__file__).resolve().parent
+DATA_XML = ROOT / "data" / "ebm.xml"
+STORE_DIR = ROOT / "data" / "vector_store"
+
+
+PIPELINE: EbmRAGPipeline | None = None
+
+
+def get_pipeline() -> EbmRAGPipeline:
+    global PIPELINE
+    if PIPELINE is None:
+        PIPELINE = build_pipeline_from_paths(DATA_XML, STORE_DIR)
+    return PIPELINE
+
+
+def format_retrieved(results: list[dict]) -> str:
+    if not results:
+        return "No retrieved documents."
+    lines = []
+    for item in results:
+        lines.append(
+            f"### {item['code']} - {item.get('title') or 'Unbenannt'}\n"
+            f"Score: {item['score']:.3f}\n\n"
+            f"{item['text']}"
+        )
+    return "\n\n".join(lines)
+
+
+def ask_ebm(question: str) -> tuple[str, str, float]:
+    pipeline = get_pipeline()
+    result = pipeline.answer(question)
+    return result["answer"], format_retrieved(result["retrieved_documents"]), result["confidence"]
+
+
+def explain_code(code: str) -> tuple[str, str]:
+    pipeline = get_pipeline()
+    result = pipeline.explain_code(code)
+    return result["answer"], format_retrieved(result["retrieved_documents"])
+
+
+def quiz_me() -> tuple[str, str, str]:
+    pipeline = get_pipeline()
+    doc = pipeline.random_document()
+    prompt = f"What does this code describe?\n\nEBM code: {doc.code}"
+    return prompt, gr.update(value="", visible=False), doc.code
+
+
+def reveal_quiz_answer(code: str) -> tuple[str, str]:
+    pipeline = get_pipeline()
+    if not code:
+        return gr.update(value="No code selected.", visible=True), ""
+    result = pipeline.explain_code(code)
+    return gr.update(value=result["answer"], visible=True), format_retrieved(result["retrieved_documents"])
+
+
+def explore_ebm(query: str, chapter: str) -> tuple[str, str, str]:
+    pipeline = get_pipeline()
+    results = pipeline.search(query=query, chapter=chapter, top_k=10)
+    if not results:
+        empty = pd.DataFrame(columns=["code", "title", "points", "exclusions", "notes"])
+        return empty, "No matches found.", ""
+
+    rows = []
+    for item in results:
+        rows.append(
+            {
+                "code": item["code"],
+                "title": item.get("title") or "",
+                "points": item.get("points") or "",
+                "exclusions": ", ".join(item.get("exclusions_text", [])),
+                "notes": " | ".join(item.get("notes", [])),
+            }
+        )
+
+    table = pd.DataFrame(rows)
+    details = format_retrieved(results)
+    chapters = "\n".join(sorted({item.get("chapter_name") or "" for item in results if item.get("chapter_name")}))
+    return table, details, chapters
+
+
+def browse_chapters() -> list[str]:
+    if STORE_DIR.exists() and (STORE_DIR / "metadata.jsonl").exists():
+        pipeline = get_pipeline()
+        chapters = pipeline.list_chapters()
+    elif DATA_XML.exists():
+        df = parse_ebm_xml_to_dataframe(str(DATA_XML))
+        chapters = sorted(
+            {
+                str(value)
+                for value in df.get("chapter_name", pd.Series(dtype=str)).dropna().tolist()
+                if str(value).strip()
+            }
+        )
+    else:
+        chapters = []
+    return ["All"] + chapters
+
+
+CSS = """
+:root {
+  --bg: #f4f7f6;
+  --panel: rgba(255,255,255,0.85);
+  --ink: #183a37;
+  --muted: #4e6964;
+  --accent: #2e7d6b;
+  --accent-2: #165d50;
+  --border: rgba(24,58,55,0.12);
+}
+
+body, .gradio-container {
+  background:
+    radial-gradient(circle at top left, rgba(46,125,107,0.12), transparent 32%),
+    radial-gradient(circle at top right, rgba(24,58,55,0.08), transparent 28%),
+    linear-gradient(180deg, #f7fbfa 0%, #eef4f2 100%);
+  color: var(--ink);
+  font-family: "IBM Plex Sans", "Segoe UI", sans-serif;
+}
+
+.hero {
+  background: linear-gradient(135deg, rgba(24,58,55,0.95), rgba(46,125,107,0.92));
+  color: white;
+  border-radius: 24px;
+  padding: 28px;
+  box-shadow: 0 20px 60px rgba(24,58,55,0.18);
+}
+
+.hero h1 {
+  margin: 0;
+  font-size: 2.4rem;
+  letter-spacing: -0.03em;
+}
+
+.hero p {
+  margin: 8px 0 0 0;
+  opacity: 0.92;
+  font-size: 1.05rem;
+}
+
+.card {
+  background: var(--panel);
+  border: 1px solid var(--border);
+  border-radius: 20px;
+  backdrop-filter: blur(10px);
+  box-shadow: 0 10px 30px rgba(24,58,55,0.08);
+}
+
+.gr-button-primary {
+  background: linear-gradient(135deg, var(--accent), var(--accent-2)) !important;
+  border: none !important;
+}
+"""
+
+
+def build_app() -> gr.Blocks:
+    with gr.Blocks(css=CSS, theme=gr.themes.Base()) as demo:
+        gr.HTML(
+            """
+            <div class="hero">
+              <h1>EBM Mentor</h1>
+              <p>Learn and explore the German EBM interactively.</p>
+            </div>
+            """
+        )
+
+        chapter_choices = browse_chapters()
+
+        with gr.Tabs():
+            with gr.Tab("Ask EBM"):
+                with gr.Row():
+                    question = gr.Textbox(
+                        label="Ask a question",
+                        placeholder="Was bedeutet EBM-Code 01100?",
+                        lines=3,
+                    )
+                ask_btn = gr.Button("Ask EBM", variant="primary")
+                answer = gr.Markdown()
+                confidence = gr.Slider(0, 1, value=0, label="Retrieval confidence", interactive=False)
+                ask_sources = gr.Markdown()
+                ask_btn.click(ask_ebm, inputs=[question], outputs=[answer, ask_sources, confidence])
+
+            with gr.Tab("Explain a Code"):
+                code_input = gr.Textbox(label="EBM code", placeholder="01100")
+                explain_btn = gr.Button("Explain", variant="primary")
+                explanation = gr.Markdown()
+                explanation_sources = gr.Markdown()
+                explain_btn.click(explain_code, inputs=[code_input], outputs=[explanation, explanation_sources])
+
+            with gr.Tab("Quiz Me"):
+                quiz_prompt = gr.Markdown()
+                quiz_answer = gr.Markdown(visible=False)
+                quiz_sources = gr.Markdown()
+                quiz_code = gr.State("")
+                quiz_btn = gr.Button("Random code", variant="primary")
+                reveal_btn = gr.Button("Reveal answer")
+                quiz_btn.click(
+                    quiz_me,
+                    inputs=None,
+                    outputs=[quiz_prompt, quiz_answer, quiz_code],
+                )
+                reveal_btn.click(
+                    reveal_quiz_answer,
+                    inputs=[quiz_code],
+                    outputs=[quiz_answer, quiz_sources],
+                )
+
+            with gr.Tab("Explore EBM"):
+                with gr.Row():
+                    search_query = gr.Textbox(label="Search", placeholder="points, exclusions, title, notes...")
+                    chapter = gr.Dropdown(
+                        label="Chapter",
+                        choices=chapter_choices,
+                        value="All",
+                        interactive=True,
+                    )
+                search_btn = gr.Button("Search", variant="primary")
+                table = gr.Dataframe(
+                    headers=["code", "title", "points", "exclusions", "notes"],
+                    datatype=["str", "str", "str", "str", "str"],
+                    interactive=False,
+                    wrap=True,
+                    row_count=(5, "dynamic"),
+                )
+                explore_sources = gr.Markdown()
+                explore_chapters = gr.Markdown()
+                search_btn.click(
+                    explore_ebm,
+                    inputs=[search_query, chapter],
+                    outputs=[table, explore_sources, explore_chapters],
+                )
+
+    return demo
+
+
+app = build_app()
+
+
+if __name__ == "__main__":
+    app.launch()

data/ebm.xml

@@ -0,0 +1,82 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<ehd:root xmlns:ehd="urn:ehd/001" xmlns:go="urn:ehd/go/001">
+  <ehd:body>
+    <go:gnr_liste>
+      <go:gnr V="01100" USE="1" VT="20240101">
+        <go:allgemein>
+          <go:legende>
+            <go:kurztext V="Unvorhergesehene Inanspruchnahme I"/>
+            <go:quittungstext V="Unvorhergesehene Inanspruchnahme I"/>
+            <go:langtext>Notdienstliche Erstversorgung und kurzfristige Inanspruchnahme außerhalb der regulären Sprechzeiten.</go:langtext>
+            <go:kap_bez V="01" DN="Allgemeine Leistungen">
+              <go:bereich DN="Allgemeinmedizin"/>
+              <go:kapitel DN="Hausärztliche Versorgung"/>
+              <go:abschnitt DN="Unvorhergesehene Inanspruchnahme"/>
+            </go:kap_bez>
+          </go:legende>
+          <go:anmerkungen_liste>
+            <go:anmerkung>Nur einmal je Behandlungstag berechnungsfähig.</go:anmerkung>
+          </go:anmerkungen_liste>
+          <go:bewertung_liste>
+            <go:bewertung V="196" U="PUNKTE">
+              <go:leistung_typ V="GKV"/>
+            </go:bewertung>
+          </go:bewertung_liste>
+        </go:allgemein>
+        <go:regel>
+          <go:ausschluss_liste>
+            <go:bezugsraum>
+              <go:gnr_liste>
+                <go:gnr V="01101" DN="Unvorhergesehene Inanspruchnahme II"/>
+                <go:gnr V="01200" DN="Weitere Inanspruchnahme"/>
+              </go:gnr_liste>
+            </go:bezugsraum>
+          </go:ausschluss_liste>
+        </go:regel>
+        <go:vdx>
+          <go:gkv_kontenart_liste>
+            <go:gkv_kontenart V="A"/>
+            <go:gkv_kontenart V="B"/>
+          </go:gkv_kontenart_liste>
+        </go:vdx>
+      </go:gnr>
+      <go:gnr V="01732" USE="1" VT="20240101">
+        <go:allgemein>
+          <go:legende>
+            <go:kurztext V="Gesundheitsuntersuchung"/>
+            <go:quittungstext V="Gesundheitsuntersuchung"/>
+            <go:langtext>Vorsorgeorientierte Untersuchung bei asymptomatischen Patienten.</go:langtext>
+            <go:kap_bez V="02" DN="Vorsorgeleistungen">
+              <go:bereich DN="Prävention"/>
+              <go:kapitel DN="Vorsorge"/>
+              <go:abschnitt DN="Gesundheitsuntersuchungen"/>
+            </go:kap_bez>
+          </go:legende>
+          <go:anmerkungen_liste>
+            <go:anmerkung>Einmal innerhalb definierter Fristen abrechenbar.</go:anmerkung>
+          </go:anmerkungen_liste>
+          <go:bewertung_liste>
+            <go:bewertung V="300" U="PUNKTE">
+              <go:leistung_typ V="GKV"/>
+            </go:bewertung>
+          </go:bewertung_liste>
+        </go:allgemein>
+        <go:regel>
+          <go:ausschluss_liste>
+            <go:bezugsraum>
+              <go:gnr_liste>
+                <go:gnr V="01733" DN="Weiteres Vorsorgepaket"/>
+              </go:gnr_liste>
+            </go:bezugsraum>
+          </go:ausschluss_liste>
+        </go:regel>
+        <go:vdx>
+          <go:gkv_kontenart_liste>
+            <go:gkv_kontenart V="A"/>
+          </go:gkv_kontenart_liste>
+        </go:vdx>
+      </go:gnr>
+    </go:gnr_liste>
+  </ehd:body>
+</ehd:root>
+

requirements.txt

@@ -0,0 +1,11 @@
+gradio>=4.44.0
+pandas>=2.2.2
+numpy>=1.26.4
+sentence-transformers>=3.0.1
+transformers>=4.44.2
+accelerate>=0.33.0
+torch>=2.3.1
+faiss-cpu>=1.8.0.post1
+pytest>=8.3.2
+pyarrow>=17.0.0
+

scripts/build_database.py

@@ -0,0 +1,31 @@
+from __future__ import annotations
+
+import argparse
+from pathlib import Path
+
+from src.chunking import dataframe_to_documents
+from src.embeddings import EmbeddingModel
+from src.parser import parse_ebm_xml_to_dataframe
+from src.vector_store import EbmVectorStore
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser(description="Build the local FAISS database from EBM XML.")
+    parser.add_argument("--xml", required=True, help="Path to the official EBM XML.")
+    parser.add_argument("--store", required=True, help="Output directory for the FAISS store.")
+    parser.add_argument("--model", default=None, help="Optional sentence-transformers model name.")
+    args = parser.parse_args()
+
+    xml_path = Path(args.xml)
+    store_dir = Path(args.store)
+    embedding_model = EmbeddingModel(args.model) if args.model else EmbeddingModel()
+
+    df = parse_ebm_xml_to_dataframe(str(xml_path))
+    documents = dataframe_to_documents(df)
+    store, embeddings = EbmVectorStore.build(documents, embedding_model=embedding_model)
+    store.save(store_dir, embeddings=embeddings)
+
+
+if __name__ == "__main__":
+    main()
+

scripts/ingest_ebm.py

@@ -0,0 +1,33 @@
+from __future__ import annotations
+
+import argparse
+import json
+from pathlib import Path
+
+from src.chunking import dataframe_to_search_corpus
+from src.parser import parse_ebm_xml_to_dataframe
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser(description="Parse the EBM XML into structured artifacts.")
+    parser.add_argument("--xml", required=True, help="Path to the official EBM XML.")
+    parser.add_argument("--output", required=True, help="Directory for generated artifacts.")
+    args = parser.parse_args()
+
+    xml_path = Path(args.xml)
+    output_dir = Path(args.output)
+    output_dir.mkdir(parents=True, exist_ok=True)
+
+    df = parse_ebm_xml_to_dataframe(str(xml_path))
+    df.to_parquet(output_dir / "ebm.parquet", index=False)
+    df.to_json(output_dir / "ebm.jsonl", orient="records", lines=True, force_ascii=False)
+
+    corpus = dataframe_to_search_corpus(df)
+    (output_dir / "ebm_documents.jsonl").write_text(
+        "\n".join(json.dumps(item, ensure_ascii=False) for item in corpus),
+        encoding="utf-8",
+    )
+
+
+if __name__ == "__main__":
+    main()

src/__init__.py

@@ -0,0 +1,2 @@
+"""EBM RAG package."""
+

src/chunking.py

@@ -0,0 +1,136 @@
+from __future__ import annotations
+
+from dataclasses import asdict, dataclass
+from typing import Any
+
+import pandas as pd
+
+
+@dataclass(frozen=True)
+class EbmDocument:
+    code: str
+    title: str
+    short_text: str | None
+    receipt_text: str | None
+    long_text: str | None
+    chapter_code: str | None
+    chapter_name: str | None
+    bereich: str | None
+    kapitel: str | None
+    abschnitt: str | None
+    notes: list[str]
+    points: int | None
+    fachgruppen: list[str]
+    exclusions: list[dict[str, str | None]]
+    gkv_account_types: list[str]
+    raw: dict[str, Any] | None = None
+
+
+def _coerce_points(value: Any) -> int | None:
+    if value in (None, ""):
+        return None
+    try:
+        return int(float(value))
+    except (TypeError, ValueError):
+        return None
+
+
+def _safe_list(value: Any) -> list[Any]:
+    if isinstance(value, list):
+        return value
+    return []
+
+
+def dataframe_to_documents(df: pd.DataFrame) -> list[EbmDocument]:
+    documents: list[EbmDocument] = []
+    for _, row in df.iterrows():
+        data = row.to_dict()
+        title = data.get("short_text") or data.get("receipt_text") or data.get("code") or ""
+        documents.append(
+            EbmDocument(
+                code=str(data.get("code") or ""),
+                title=str(title),
+                short_text=data.get("short_text"),
+                receipt_text=data.get("receipt_text"),
+                long_text=data.get("long_text"),
+                chapter_code=data.get("chapter_code"),
+                chapter_name=data.get("chapter_name"),
+                bereich=data.get("bereich"),
+                kapitel=data.get("kapitel"),
+                abschnitt=data.get("abschnitt"),
+                notes=[str(item) for item in _safe_list(data.get("notes")) if item],
+                points=_coerce_points(data.get("points")),
+                fachgruppen=[str(item) for item in _safe_list(data.get("fachgruppen")) if item],
+                exclusions=[
+                    {
+                        "code": item.get("code"),
+                        "description": item.get("description"),
+                    }
+                    for item in _safe_list(data.get("exclusions"))
+                    if isinstance(item, dict)
+                ],
+                gkv_account_types=[str(item) for item in _safe_list(data.get("gkv_account_types")) if item],
+                raw=data,
+            )
+        )
+    return documents
+
+
+def _format_bullets(items: list[str]) -> str:
+    return "\n".join(f"- {item}" for item in items) if items else "Nicht angegeben."
+
+
+def _format_exclusions(items: list[dict[str, str | None]]) -> str:
+    if not items:
+        return "Keine Ausschlüsse angegeben."
+    formatted = []
+    for item in items:
+        code = item.get("code") or ""
+        description = item.get("description") or ""
+        if description:
+            formatted.append(f"- {code}: {description}")
+        else:
+            formatted.append(f"- {code}")
+    return "\n".join(formatted)
+
+
+def document_to_search_text(doc: EbmDocument) -> str:
+    parts = [
+        f"EBM Code: {doc.code}",
+        f"Title: {doc.title}",
+    ]
+    if doc.short_text:
+        parts.append(f"Short text: {doc.short_text}")
+    if doc.receipt_text:
+        parts.append(f"Receipt text: {doc.receipt_text}")
+    if doc.long_text:
+        parts.append(f"Description: {doc.long_text}")
+    if doc.points is not None:
+        parts.append(f"Points: {doc.points}")
+    if doc.notes:
+        parts.append("Notes:\n" + _format_bullets(doc.notes))
+    if doc.exclusions:
+        parts.append("Exclusions:\n" + _format_exclusions(doc.exclusions))
+    if doc.fachgruppen:
+        parts.append("Fachgruppen:\n" + _format_bullets(doc.fachgruppen))
+    if doc.gkv_account_types:
+        parts.append("GKV account types:\n" + _format_bullets(doc.gkv_account_types))
+    if doc.chapter_name:
+        parts.append(f"Chapter: {doc.chapter_name}")
+    if doc.kapitel:
+        parts.append(f"Kapitel: {doc.kapitel}")
+    if doc.abschnitt:
+        parts.append(f"Abschnitt: {doc.abschnitt}")
+    return "\n\n".join(parts)
+
+
+def document_to_structured_dict(doc: EbmDocument) -> dict[str, Any]:
+    payload = asdict(doc)
+    payload["search_text"] = document_to_search_text(doc)
+    return payload
+
+
+def dataframe_to_search_corpus(df: pd.DataFrame) -> list[dict[str, Any]]:
+    docs = dataframe_to_documents(df)
+    return [document_to_structured_dict(doc) for doc in docs]
+

src/embeddings.py

@@ -0,0 +1,37 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import Iterable
+
+import numpy as np
+from sentence_transformers import SentenceTransformer
+
+
+DEFAULT_EMBEDDING_MODEL = "sentence-transformers/paraphrase-multilingual-MiniLM-L12-v2"
+
+
+@dataclass
+class EmbeddingConfig:
+    model_name: str = DEFAULT_EMBEDDING_MODEL
+
+
+class EmbeddingModel:
+    def __init__(self, model_name: str = DEFAULT_EMBEDDING_MODEL):
+        self.model_name = model_name
+        self._model: SentenceTransformer | None = None
+
+    @property
+    def model(self) -> SentenceTransformer:
+        if self._model is None:
+            self._model = SentenceTransformer(self.model_name)
+        return self._model
+
+    def encode(self, texts: Iterable[str]) -> np.ndarray:
+        embeddings = self.model.encode(
+            list(texts),
+            normalize_embeddings=True,
+            convert_to_numpy=True,
+            show_progress_bar=False,
+        )
+        return np.asarray(embeddings, dtype=np.float32)
+

src/model.py

@@ -0,0 +1,30 @@
+from __future__ import annotations
+
+import torch
+from transformers import AutoModelForCausalLM, AutoTokenizer, pipeline
+
+
+DEFAULT_GENERATION_MODEL = "CohereLabs/tiny-aya-water"
+
+
+def load_generation_pipeline(model_name: str = DEFAULT_GENERATION_MODEL):
+    tokenizer = AutoTokenizer.from_pretrained(model_name, trust_remote_code=True)
+    dtype = torch.bfloat16 if torch.cuda.is_available() and torch.cuda.is_bf16_supported() else torch.float32
+    model = AutoModelForCausalLM.from_pretrained(
+        model_name,
+        trust_remote_code=True,
+        torch_dtype=dtype,
+        device_map="auto",
+        low_cpu_mem_usage=True,
+    )
+    return pipeline(
+        "text-generation",
+        model=model,
+        tokenizer=tokenizer,
+        max_new_tokens=256,
+        do_sample=False,
+        temperature=0.0,
+        return_full_text=False,
+        pad_token_id=tokenizer.eos_token_id,
+    )
+

src/parser.py

@@ -0,0 +1,109 @@
+from __future__ import annotations
+
+from typing import Any
+import xml.etree.ElementTree as ET
+
+import pandas as pd
+
+
+NAMESPACES = {"ehd": "urn:ehd/001", "go": "urn:ehd/go/001"}
+
+
+def get_text_content(elem: ET.Element | None) -> str | None:
+    """Extract all text recursively from an XML element."""
+    if elem is None:
+        return None
+
+    text = " ".join(t.strip() for t in elem.itertext() if t.strip())
+    return text if text else None
+
+
+def parse_ebm_xml_to_dataframe(xml_path: str) -> pd.DataFrame:
+    """Parse EBM XML into a pandas DataFrame with one row per GNR."""
+    tree = ET.parse(xml_path)
+    root = tree.getroot()
+
+    rows: list[dict[str, Any]] = []
+
+    for gnr in root.findall("./ehd:body/go:gnr_liste/go:gnr", namespaces=NAMESPACES):
+        row: dict[str, Any] = {
+            "code": gnr.get("V"),
+            "use": gnr.get("USE"),
+            "valid_from": gnr.get("VT"),
+        }
+
+        legende = gnr.find("./go:allgemein/go:legende", namespaces=NAMESPACES)
+        if legende is not None:
+            kurztext = legende.find("go:kurztext", namespaces=NAMESPACES)
+            quittungstext = legende.find("go:quittungstext", namespaces=NAMESPACES)
+            langtext = legende.find("go:langtext", namespaces=NAMESPACES)
+            kap_bez = legende.find("go:kap_bez", namespaces=NAMESPACES)
+
+            row["short_text"] = kurztext.get("V") if kurztext is not None else None
+            row["receipt_text"] = quittungstext.get("V") if quittungstext is not None else None
+            row["long_text"] = get_text_content(langtext)
+
+            if kap_bez is not None:
+                bereich = kap_bez.find("go:bereich", namespaces=NAMESPACES)
+                kapitel = kap_bez.find("go:kapitel", namespaces=NAMESPACES)
+                abschnitt = kap_bez.find("go:abschnitt", namespaces=NAMESPACES)
+                row["chapter_code"] = kap_bez.get("V")
+                row["chapter_name"] = kap_bez.get("DN")
+                row["bereich"] = bereich.get("DN") if bereich is not None else None
+                row["kapitel"] = kapitel.get("DN") if kapitel is not None else None
+                row["abschnitt"] = abschnitt.get("DN") if abschnitt is not None else None
+
+        row["service_period"] = (
+            gnr.find("./go:allgemein/go:gueltigkeit/go:service_tmr", namespaces=NAMESPACES).get("V")
+            if gnr.find("./go:allgemein/go:gueltigkeit/go:service_tmr", namespaces=NAMESPACES) is not None
+            else None
+        )
+        row["effective_period"] = (
+            gnr.find("./go:allgemein/go:gueltigkeit/go:effective_tmr", namespaces=NAMESPACES).get("V")
+            if gnr.find("./go:allgemein/go:gueltigkeit/go:effective_tmr", namespaces=NAMESPACES) is not None
+            else None
+        )
+
+        notes: list[str] = []
+        for note in gnr.findall("./go:allgemein/go:anmerkungen_liste/go:anmerkung", namespaces=NAMESPACES):
+            txt = get_text_content(note)
+            if txt:
+                notes.append(txt)
+        row["notes"] = notes
+
+        bewertung = gnr.find("./go:allgemein/go:bewertung_liste/go:bewertung", namespaces=NAMESPACES)
+        if bewertung is not None:
+            row["points"] = bewertung.get("V")
+            row["unit"] = bewertung.get("U")
+            lt = bewertung.find("go:leistung_typ", namespaces=NAMESPACES)
+            row["leistung_typ"] = lt.get("V") if lt is not None else None
+
+        fachgruppen: list[str] = []
+        for fg in gnr.findall(".//go:fachgruppe_liste//go:fachgruppe", namespaces=NAMESPACES):
+            value = fg.get("V")
+            if value:
+                fachgruppen.append(value)
+        row["fachgruppen"] = fachgruppen
+
+        exclusions: list[dict[str, str | None]] = []
+        for ex in gnr.findall("./go:regel/go:ausschluss_liste/go:bezugsraum", namespaces=NAMESPACES):
+            for ex_gnr in ex.findall("./go:gnr_liste/go:gnr", namespaces=NAMESPACES):
+                exclusions.append(
+                    {
+                        "code": ex_gnr.get("V"),
+                        "description": ex_gnr.get("DN"),
+                    }
+                )
+        row["exclusions"] = exclusions
+
+        gkv_types: list[str] = []
+        for gkv in gnr.findall("./go:vdx/go:gkv_kontenart_liste/go:gkv_kontenart", namespaces=NAMESPACES):
+            value = gkv.get("V")
+            if value:
+                gkv_types.append(value)
+        row["gkv_account_types"] = gkv_types
+
+        rows.append(row)
+
+    return pd.DataFrame(rows)
+

src/prompts.py

@@ -0,0 +1,43 @@
+from __future__ import annotations
+
+
+NO_ANSWER_TEXT = "Diese Information ist nicht in den bereitgestellten EBM-Daten enthalten."
+
+ANSWER_PROMPT = """You are an EBM billing tutor.
+Answer ONLY using the provided context.
+If the answer cannot be found in the context, say:
+
+"Diese Information ist nicht in den bereitgestellten EBM-Daten enthalten."
+
+Context:
+{retrieved_documents}
+
+Question:
+{user_question}
+
+Answer:
+"""
+
+CODE_EXPLANATION_PROMPT = """You are an EBM billing tutor.
+Explain this code ONLY using the provided context.
+Return a concise, structured explanation with the fields:
+- Code
+- Title
+- Description
+- Points
+- Notes
+- Exclusions
+- Fachgruppen
+- GKV account types
+
+If a field is missing, say it is not provided in the context.
+
+Context:
+{retrieved_documents}
+
+Question:
+{user_question}
+
+Answer:
+"""
+

src/rag_pipeline.py

@@ -0,0 +1,141 @@
+from __future__ import annotations
+
+from pathlib import Path
+from typing import Any
+
+from src.chunking import dataframe_to_documents
+from src.embeddings import EmbeddingModel
+from src.model import load_generation_pipeline
+from src.prompts import ANSWER_PROMPT, CODE_EXPLANATION_PROMPT, NO_ANSWER_TEXT
+from src.parser import parse_ebm_xml_to_dataframe
+from src.retriever import EbmRetriever
+from src.vector_store import EbmVectorStore
+
+
+def _format_context(results: list[dict[str, Any]]) -> str:
+    blocks = []
+    for item in results:
+        notes = "\n".join(f"- {note}" for note in item.get("notes", [])) or "Keine."
+        exclusions = "\n".join(
+            f"- {ex['code']}: {ex.get('description') or ''}".strip()
+            for ex in item.get("exclusions", [])
+            if ex.get("code")
+        ) or "Keine."
+        blocks.append(
+            "\n".join(
+                [
+                    f"EBM Code: {item.get('code')}",
+                    f"Title: {item.get('title') or ''}",
+                    f"Points: {item.get('points') if item.get('points') is not None else 'Nicht angegeben'}",
+                    f"Chapter: {item.get('chapter_name') or ''}",
+                    f"Description: {item.get('long_text') or item.get('short_text') or ''}",
+                    f"Notes:\n{notes}",
+                    f"Exclusions:\n{exclusions}",
+                    f"Fachgruppen: {', '.join(item.get('fachgruppen', [])) or 'Nicht angegeben'}",
+                    f"GKV account types: {', '.join(item.get('gkv_account_types', [])) or 'Nicht angegeben'}",
+                ]
+            )
+        )
+    return "\n\n---\n\n".join(blocks)
+
+
+def _extract_citations(results: list[dict[str, Any]]) -> list[str]:
+    citations = []
+    for item in results:
+        code = item.get("code")
+        title = item.get("title")
+        if code:
+            citations.append(f"{code} - {title}".strip())
+    return citations
+
+
+class EbmRAGPipeline:
+    def __init__(self, retriever: EbmRetriever, generator=None):
+        self.retriever = retriever
+        self._generator = generator
+
+    @property
+    def generator(self):
+        if self._generator is None:
+            self._generator = load_generation_pipeline()
+        return self._generator
+
+    def answer(self, question: str, top_k: int = 5, chapter: str | None = None) -> dict[str, Any]:
+        retrieved = self.retriever.retrieve(question, top_k=top_k, chapter=chapter)
+        if not retrieved:
+            return {
+                "answer": NO_ANSWER_TEXT,
+                "retrieved_documents": [],
+                "confidence": 0.0,
+                "citations": [],
+            }
+
+        context = _format_context(retrieved)
+        prompt = ANSWER_PROMPT.format(retrieved_documents=context, user_question=question)
+        try:
+            generated = self.generator(prompt)[0]["generated_text"].strip()
+        except Exception:
+            generated = ""
+        answer = generated or NO_ANSWER_TEXT
+        confidence = max(0.0, min(1.0, float(retrieved[0].get("confidence", 0.0))))
+        return {
+            "answer": answer,
+            "retrieved_documents": retrieved,
+            "confidence": confidence,
+            "citations": _extract_citations(retrieved),
+        }
+
+    def explain_code(self, code: str) -> dict[str, Any]:
+        code = code.strip()
+        document = self.retriever.get_by_code(code)
+        if not document:
+            return {
+                "answer": NO_ANSWER_TEXT,
+                "retrieved_documents": [],
+                "confidence": 0.0,
+                "citations": [],
+            }
+
+        retrieved = [dict(document)]
+        context = _format_context(retrieved)
+        prompt = CODE_EXPLANATION_PROMPT.format(retrieved_documents=context, user_question=f"Explain EBM code {code}.")
+        try:
+            generated = self.generator(prompt)[0]["generated_text"].strip()
+        except Exception:
+            generated = ""
+        confidence = 1.0
+        return {
+            "answer": generated or NO_ANSWER_TEXT,
+            "retrieved_documents": retrieved,
+            "confidence": confidence,
+            "citations": _extract_citations(retrieved),
+        }
+
+    def search(self, query: str, top_k: int = 10, chapter: str | None = None) -> list[dict[str, Any]]:
+        return self.retriever.search(query=query, top_k=top_k, chapter=chapter)
+
+    def random_document(self):
+        from types import SimpleNamespace
+
+        doc = self.retriever.random_document()
+        return SimpleNamespace(**doc)
+
+    def list_chapters(self) -> list[str]:
+        return self.retriever.list_chapters()
+
+
+def build_pipeline_from_paths(xml_path: str | Path, store_dir: str | Path, embedding_model: EmbeddingModel | None = None) -> EbmRAGPipeline:
+    xml_path = Path(xml_path)
+    store_dir = Path(store_dir)
+    embedding_model = embedding_model or EmbeddingModel()
+
+    if store_dir.exists() and (store_dir / "index.faiss").exists() and (store_dir / "metadata.jsonl").exists():
+        store = EbmVectorStore.load(store_dir)
+    else:
+        df = parse_ebm_xml_to_dataframe(str(xml_path))
+        documents = dataframe_to_documents(df)
+        store, embeddings = EbmVectorStore.build(documents, embedding_model=embedding_model)
+        store.save(store_dir, embeddings=embeddings)
+
+    retriever = EbmRetriever(store, embedding_model=embedding_model)
+    return EbmRAGPipeline(retriever=retriever)

src/retriever.py

@@ -0,0 +1,73 @@
+from __future__ import annotations
+
+import re
+from typing import Any
+
+from src.embeddings import EmbeddingModel
+from src.vector_store import EbmVectorStore, RetrievalResult
+
+
+CODE_PATTERN = re.compile(r"\b\d{5}\b")
+
+
+class EbmRetriever:
+    def __init__(self, store: EbmVectorStore, embedding_model: EmbeddingModel | None = None):
+        self.store = store
+        self.embedding_model = embedding_model or EmbeddingModel(store.embedding_model_name)
+
+    def retrieve(self, query: str, top_k: int = 5, chapter: str | None = None) -> list[dict[str, Any]]:
+        if not query.strip():
+            return []
+
+        embeddings = self.embedding_model.encode([query])
+        results = self.store.search(embeddings, top_k=top_k * 3 if chapter and chapter != "All" else top_k)
+
+        payloads = [self._to_payload(result) for result in results]
+        if chapter and chapter != "All":
+            payloads = [item for item in payloads if item.get("chapter_name") == chapter]
+        return payloads[:top_k]
+
+    def get_by_code(self, code: str) -> dict[str, Any] | None:
+        code = code.strip()
+        for doc in self.store.documents:
+            if str(doc.get("code") or "") == code:
+                return dict(doc)
+        return None
+
+    def random_document(self) -> dict[str, Any]:
+        import random
+
+        if not self.store.documents:
+            raise ValueError("No documents available.")
+        return dict(random.choice(self.store.documents))
+
+    def list_chapters(self) -> list[str]:
+        chapters = sorted(
+            {
+                str(doc.get("chapter_name"))
+                for doc in self.store.documents
+                if doc.get("chapter_name")
+            }
+        )
+        return chapters
+
+    def search(self, query: str, top_k: int = 10, chapter: str | None = None) -> list[dict[str, Any]]:
+        return self.retrieve(query=query, top_k=top_k, chapter=chapter)
+
+    def code_from_text(self, text: str) -> str | None:
+        match = CODE_PATTERN.search(text or "")
+        return match.group(0) if match else None
+
+    @staticmethod
+    def _to_payload(result: RetrievalResult) -> dict[str, Any]:
+        payload = dict(result.structured)
+        payload["score"] = result.score
+        payload["title"] = result.title
+        payload["text"] = result.text
+        payload["confidence"] = max(0.0, min(1.0, (result.score + 1.0) / 2.0))
+        payload["exclusions_text"] = [
+            item.get("code")
+            for item in payload.get("exclusions", [])
+            if isinstance(item, dict) and item.get("code")
+        ]
+        return payload

src/vector_store.py

@@ -0,0 +1,117 @@
+from __future__ import annotations
+
+import json
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Any, Iterable
+
+import faiss
+import numpy as np
+
+from src.chunking import EbmDocument, dataframe_to_documents, document_to_search_text, document_to_structured_dict
+from src.embeddings import EmbeddingModel, DEFAULT_EMBEDDING_MODEL
+
+
+@dataclass
+class RetrievalResult:
+    code: str
+    title: str
+    score: float
+    text: str
+    structured: dict[str, Any]
+
+
+class EbmVectorStore:
+    def __init__(self, index: faiss.Index | None, documents: list[dict[str, Any]], embedding_model_name: str):
+        self.index = index
+        self.documents = documents
+        self.embedding_model_name = embedding_model_name
+
+    @classmethod
+    def build(
+        cls,
+        documents: Iterable[EbmDocument],
+        embedding_model: EmbeddingModel | None = None,
+    ) -> tuple["EbmVectorStore", np.ndarray]:
+        embedding_model = embedding_model or EmbeddingModel()
+        docs = [
+            document_to_structured_dict(doc) if hasattr(doc, "__dataclass_fields__") else dict(doc)
+            for doc in documents
+        ]
+        texts = [
+            document_to_search_text(EbmDocument(**{k: v for k, v in doc.items() if k != "search_text"}))
+            for doc in docs
+        ]
+        embeddings = embedding_model.encode(texts)
+        index = faiss.IndexFlatIP(embeddings.shape[1])
+        index.add(embeddings)
+        store = cls(index=index, documents=docs, embedding_model_name=embedding_model.model_name)
+        return store, embeddings
+
+    @classmethod
+    def from_dataframe(cls, df, embedding_model: EmbeddingModel | None = None) -> tuple["EbmVectorStore", np.ndarray]:
+        return cls.build(dataframe_to_documents(df), embedding_model=embedding_model)
+
+    def save(self, directory: str | Path, embeddings: np.ndarray | None = None) -> None:
+        path = Path(directory)
+        path.mkdir(parents=True, exist_ok=True)
+
+        if self.index is None:
+            raise ValueError("Cannot save a store without an index.")
+
+        faiss.write_index(self.index, str(path / "index.faiss"))
+        (path / "metadata.jsonl").write_text(
+            "\n".join(json.dumps(doc, ensure_ascii=False) for doc in self.documents),
+            encoding="utf-8",
+        )
+        (path / "config.json").write_text(
+            json.dumps({"embedding_model_name": self.embedding_model_name}, ensure_ascii=False, indent=2),
+            encoding="utf-8",
+        )
+        if embeddings is not None:
+            np.save(path / "embeddings.npy", embeddings)
+
+    @classmethod
+    def load(cls, directory: str | Path) -> "EbmVectorStore":
+        path = Path(directory)
+        index = faiss.read_index(str(path / "index.faiss"))
+        metadata_path = path / "metadata.jsonl"
+        documents = [json.loads(line) for line in metadata_path.read_text(encoding="utf-8").splitlines() if line.strip()]
+        config_path = path / "config.json"
+        if config_path.exists():
+            config = json.loads(config_path.read_text(encoding="utf-8"))
+            embedding_model_name = config.get("embedding_model_name", DEFAULT_EMBEDDING_MODEL)
+        else:
+            embedding_model_name = DEFAULT_EMBEDDING_MODEL
+        return cls(index=index, documents=documents, embedding_model_name=embedding_model_name)
+
+    def search(self, query_embedding: np.ndarray, top_k: int = 5) -> list[RetrievalResult]:
+        if self.index is None:
+            return []
+
+        query = np.asarray(query_embedding, dtype=np.float32)
+        if query.ndim == 1:
+            query = query[None, :]
+        scores, indices = self.index.search(query, top_k)
+
+        results: list[RetrievalResult] = []
+        for score, idx in zip(scores[0], indices[0]):
+            if idx < 0 or idx >= len(self.documents):
+                continue
+            doc = self.documents[idx]
+            structured = dict(doc)
+            search_text = structured.get("search_text")
+            if not search_text:
+                search_text = document_to_search_text(
+                    EbmDocument(**{k: v for k, v in structured.items() if k != "search_text"})
+                )
+            results.append(
+                RetrievalResult(
+                    code=str(doc.get("code") or ""),
+                    title=str(doc.get("title") or doc.get("short_text") or doc.get("code") or ""),
+                    score=float(score),
+                    text=str(search_text or ""),
+                    structured=structured,
+                )
+            )
+        return results

tests/fixtures/ebm_sample.xml

@@ -0,0 +1,44 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<ehd:root xmlns:ehd="urn:ehd/001" xmlns:go="urn:ehd/go/001">
+  <ehd:body>
+    <go:gnr_liste>
+      <go:gnr V="01100" USE="1" VT="20240101">
+        <go:allgemein>
+          <go:legende>
+            <go:kurztext V="Unvorhergesehene Inanspruchnahme I"/>
+            <go:quittungstext V="QText"/>
+            <go:langtext>Langtext der Leistung.</go:langtext>
+            <go:kap_bez V="01" DN="Kapitel 01">
+              <go:bereich DN="Bereich A"/>
+              <go:kapitel DN="Kapitelname"/>
+              <go:abschnitt DN="Abschnittname"/>
+            </go:kap_bez>
+          </go:legende>
+          <go:anmerkungen_liste>
+            <go:anmerkung>Anmerkung 1</go:anmerkung>
+          </go:anmerkungen_liste>
+          <go:bewertung_liste>
+            <go:bewertung V="196" U="PUNKTE">
+              <go:leistung_typ V="GKV"/>
+            </go:bewertung>
+          </go:bewertung_liste>
+        </go:allgemein>
+        <go:regel>
+          <go:ausschluss_liste>
+            <go:bezugsraum>
+              <go:gnr_liste>
+                <go:gnr V="01101" DN="Ausschluss eins"/>
+              </go:gnr_liste>
+            </go:bezugsraum>
+          </go:ausschluss_liste>
+        </go:regel>
+        <go:vdx>
+          <go:gkv_kontenart_liste>
+            <go:gkv_kontenart V="A"/>
+          </go:gkv_kontenart_liste>
+        </go:vdx>
+      </go:gnr>
+    </go:gnr_liste>
+  </ehd:body>
+</ehd:root>
+

tests/test_chunking.py

@@ -0,0 +1,36 @@
+from __future__ import annotations
+
+import pandas as pd
+
+from src.chunking import dataframe_to_documents, document_to_search_text
+
+
+def test_dataframe_to_documents_and_search_text() -> None:
+    df = pd.DataFrame(
+        [
+            {
+                "code": "01100",
+                "short_text": "Unvorhergesehene Inanspruchnahme I",
+                "receipt_text": "Receipt",
+                "long_text": "Beschreibung.",
+                "chapter_code": "01",
+                "chapter_name": "Kapitel 01",
+                "bereich": "Bereich A",
+                "kapitel": "Kapitelname",
+                "abschnitt": "Abschnittname",
+                "notes": ["Note 1", "Note 2"],
+                "points": "196",
+                "fachgruppen": ["A", "B"],
+                "exclusions": [{"code": "01101", "description": "Ausschluss"}],
+                "gkv_account_types": ["A"],
+            }
+        ]
+    )
+
+    docs = dataframe_to_documents(df)
+    assert len(docs) == 1
+    text = document_to_search_text(docs[0])
+    assert "EBM Code: 01100" in text
+    assert "Points: 196" in text
+    assert "Exclusions:" in text
+

tests/test_parser.py

@@ -0,0 +1,67 @@
+from __future__ import annotations
+
+from pathlib import Path
+
+from src.parser import parse_ebm_xml_to_dataframe
+
+
+def test_parse_ebm_xml_to_dataframe(tmp_path: Path) -> None:
+    xml = tmp_path / "sample.xml"
+    xml.write_text(
+        """<?xml version="1.0" encoding="UTF-8"?>
+<ehd:root xmlns:ehd="urn:ehd/001" xmlns:go="urn:ehd/go/001">
+  <ehd:body>
+    <go:gnr_liste>
+      <go:gnr V="01100" USE="1" VT="20240101">
+        <go:allgemein>
+          <go:legende>
+            <go:kurztext V="Unvorhergesehene Inanspruchnahme I"/>
+            <go:quittungstext V="QText"/>
+            <go:langtext>Langtext der Leistung.</go:langtext>
+            <go:kap_bez V="01" DN="Kapitel 01">
+              <go:bereich DN="Bereich A"/>
+              <go:kapitel DN="Kapitelname"/>
+              <go:abschnitt DN="Abschnittname"/>
+            </go:kap_bez>
+          </go:legende>
+          <go:anmerkungen_liste>
+            <go:anmerkung>Anmerkung 1</go:anmerkung>
+          </go:anmerkungen_liste>
+          <go:bewertung_liste>
+            <go:bewertung V="196" U="PUNKTE">
+              <go:leistung_typ V="GKV"/>
+            </go:bewertung>
+          </go:bewertung_liste>
+        </go:allgemein>
+        <go:regel>
+          <go:ausschluss_liste>
+            <go:bezugsraum>
+              <go:gnr_liste>
+                <go:gnr V="01101" DN="Ausschluss eins"/>
+              </go:gnr_liste>
+            </go:bezugsraum>
+          </go:ausschluss_liste>
+        </go:regel>
+        <go:vdx>
+          <go:gkv_kontenart_liste>
+            <go:gkv_kontenart V="A"/>
+          </go:gkv_kontenart_liste>
+        </go:vdx>
+      </go:gnr>
+    </go:gnr_liste>
+  </ehd:body>
+</ehd:root>
+""",
+        encoding="utf-8",
+    )
+
+    df = parse_ebm_xml_to_dataframe(str(xml))
+    assert len(df) == 1
+    row = df.iloc[0]
+    assert row["code"] == "01100"
+    assert row["short_text"] == "Unvorhergesehene Inanspruchnahme I"
+    assert row["points"] == "196"
+    assert row["notes"] == ["Anmerkung 1"]
+    assert row["fachgruppen"] == []
+    assert row["exclusions"][0]["code"] == "01101"
+

tests/test_retrieval.py

@@ -0,0 +1,75 @@
+from __future__ import annotations
+
+import numpy as np
+
+from src.chunking import EbmDocument
+from src.vector_store import EbmVectorStore
+from src.retriever import EbmRetriever
+
+
+class DummyEmbeddingModel:
+    model_name = "dummy"
+
+    def encode(self, texts):
+        vectors = []
+        for text in texts:
+            lower = text.lower()
+            vectors.append(
+                np.array(
+                    [
+                        1.0 if "01100" in lower else 0.0,
+                        1.0 if "inanspruchnahme" in lower else 0.0,
+                        1.0 if "vorsorge" in lower else 0.0,
+                    ],
+                    dtype=np.float32,
+                )
+            )
+        arr = np.vstack(vectors)
+        norm = np.linalg.norm(arr, axis=1, keepdims=True)
+        norm[norm == 0] = 1.0
+        return arr / norm
+
+
+def test_retrieval_ranks_relevant_code() -> None:
+    docs = [
+        EbmDocument(
+            code="01100",
+            title="Unvorhergesehene Inanspruchnahme I",
+            short_text="Unvorhergesehene Inanspruchnahme I",
+            receipt_text=None,
+            long_text="Notfallversorgung",
+            chapter_code=None,
+            chapter_name="Kapitel A",
+            bereich=None,
+            kapitel=None,
+            abschnitt=None,
+            notes=[],
+            points=196,
+            fachgruppen=[],
+            exclusions=[],
+            gkv_account_types=[],
+        ),
+        EbmDocument(
+            code="01732",
+            title="Vorsorge",
+            short_text="Vorsorge",
+            receipt_text=None,
+            long_text="Vorsorgeleistung",
+            chapter_code=None,
+            chapter_name="Kapitel B",
+            bereich=None,
+            kapitel=None,
+            abschnitt=None,
+            notes=[],
+            points=100,
+            fachgruppen=[],
+            exclusions=[],
+            gkv_account_types=[],
+        ),
+    ]
+
+    store, _ = EbmVectorStore.build(docs, embedding_model=DummyEmbeddingModel())
+    retriever = EbmRetriever(store, embedding_model=DummyEmbeddingModel())
+    results = retriever.retrieve("Was bedeutet 01100?", top_k=1)
+    assert results[0]["code"] == "01100"
+