README

README.md

@@ -10,10 +10,11 @@ vytvorená RAG vrstva, ktorá pripravuje kontext, zdroje a pravidlá pre
 odpoveď jazykového modelu. API je integrované so školským OpenWebUI ako
 OpenAPI Tool Server.
 
-Súčasná hlavná retrieval vetva používa klasický hybridný RAG:
-FTS5/BM25 + embeddingy + RRF. GraphRAG zatiaľ nie je implementovaný;
-je preň pripravený samostatný evaluačný dataset a bude sa dopĺňať ako
-ďalšia experimentálna vetva.
+Stabilná baseline retrieval vetva používa klasický hybridný RAG:
+FTS5/BM25 + embeddingy + RRF. Nad týmto baseline sa vo vetve GraphRAG
+dopĺňa knowledge graph uložený v Neo4j. Aktuálne je implementovaná
+grafová infraštruktúra, deterministický builder a vizualizácia grafu;
+graph retrieval a jeho spojenie s RAG vrstvou sú ďalší krok.
 
 ## Implementované
 
@@ -60,6 +61,14 @@ je preň pripravený samostatný evaluačný dataset a bude sa dopĺňať ako
 - atomická výmena databázy po úspešnom reindexovaní,
 - persistentná Hugging Face cache v Docker volume,
 - warm-up embeddingového modelu,
+- Neo4j databáza spúšťaná ako samostatná Docker služba,
+- deterministická tvorba knowledge graphu z `documents.json` a
+  `chunks.json`,
+- grafové entity `Person`, `Document`, `Work`, `Topic`, `Category` a
+  `Author`,
+- grafové vzťahy pre dokumenty, práce, témy, kategórie a autorstvo,
+- uniqueness constraints a opakovateľný rebuild GraphRAG grafu,
+- Neo4j Browser na vizualizáciu a kontrolu vytvoreného grafu,
 - retrieval evaluácia pre FTS, embeddings a hybridný retrieval,
 - answer-level RAG evaluácia cez OpenWebUI model a RAG tool,
 - oddelené answer-level overrides bez úpravy frozen retrieval
@@ -113,6 +122,34 @@ jazykový model
 odpoveď + source_url
 ```
 
+GraphRAG vývojová vetva:
+
+```text
+documents.json + chunks.json
+  ↓
+graphrag.py
+  ↓
+build_graphrag.py
+  ↓
+Neo4j
+├── Person
+├── Document
+├── Work
+├── Topic
+├── Category
+└── Author
+  ↓
+grafové vzťahy + vizualizácia
+  ↓
+graph retrieval
+  ↓
+spojenie s klasickým RAG
+```
+
+V aktuálnom stave je implementované vytvorenie a uloženie knowledge
+graphu. Samotný graph retrieval a jeho zapojenie do `/rag` sa ešte
+dopĺňajú.
+
 Synchronizačná vetva:
 
 ```text
@@ -167,6 +204,10 @@ zp-agent/
 │   ├── build_chunks.py
 │   ├── build_sqlite_index.py
 │   ├── embedding_utils.py
+│   ├── graph_db.py
+│   ├── graphrag.py
+│   ├── build_graphrag.py
+│   ├── neo4j_smoke.py
 │   ├── rag_utils.py
 │   ├── rebuild_index.py
 │   ├── search_db.py
@@ -277,6 +318,33 @@ vyhľadávacou vrstvou.
 - normalizuje vektory,
 - poskytuje utility pre vektorové vyhľadávanie.
 
+`graph_db.py`
+
+- načítava konfiguráciu Neo4j,
+- vytvára Neo4j driver,
+- poskytuje kontrolu spojenia s grafovou databázou.
+
+`graphrag.py`
+
+- pripravuje dátový model knowledge graphu,
+- normalizuje grafové identifikátory,
+- extrahuje osoby, práce, roky, témy, kategórie a súvisiace metadata,
+- pripravuje uzly a vzťahy pre zápis do Neo4j.
+
+`build_graphrag.py`
+
+- načíta `documents.json` a `chunks.json`,
+- vytvorí GraphRAG payload,
+- vytvorí Neo4j uniqueness constraints,
+- zapíše uzly a vzťahy,
+- podporuje opakovateľný rebuild spravovanej časti grafu,
+- vypíše základné počty uzlov, vzťahov a kvalitu extrakcie názvov prác.
+
+`neo4j_smoke.py`
+
+Jednoduchý smoke test pripojenia k Neo4j používaný pri kontrole Docker
+prostredia.
+
 `search_core.py`
 
 Nízkoúrovňové jadro vyhľadávania:
@@ -420,7 +488,7 @@ ladenie frozen benchmarku.
 
 `graphrag_questions.json`
 
-Samostatný benchmark pripravený pre budúcu GraphRAG vetvu.
+Samostatný benchmark pripravený pre vývoj GraphRAG vetvy.
 
 Obsahuje 500 otázok zameraných na:
 
@@ -445,8 +513,9 @@ graph.expected_relations
 graph.expected_paths
 ```
 
-Dôležité: existencia tohto datasetu neznamená, že je GraphRAG už
-implementovaný. Dataset je pripravený na neskorší vývoj a porovnanie.
+Dataset sa používa pre vývoj a neskoršie porovnanie GraphRAG vetvy.
+Knowledge graph a jeho Neo4j vrstva sú už implementované; samotný
+graph retrieval a answer-level GraphRAG tok sa ešte dopĺňajú.
 
 `robustness_questions.json`
 
@@ -517,6 +586,12 @@ WEBHOOK_PULL_GIT=false
 # Voliteľné
 EMBEDDING_MODEL=intfloat/multilingual-e5-small
 EMBEDDING_BATCH_SIZE=32
+
+# GraphRAG / Neo4j
+NEO4J_URI=bolt://localhost:7687
+NEO4J_USER=neo4j
+NEO4J_PASSWORD=<silné lokálne heslo>
+NEO4J_DATABASE=neo4j
 ```
 
 Tajomstvá je možné vygenerovať príkazom:
@@ -569,6 +644,26 @@ docker compose down
 Hugging Face cache je uložená v persistentnom Docker volume, takže sa
 embeddingový model pri bežnom reštarte kontajnera nemusí znova sťahovať.
 
+Neo4j je spustené ako samostatná služba v rovnakom Docker Compose
+prostredí. Stav služieb je možné skontrolovať:
+
+```bash
+docker compose ps
+```
+
+Neo4j Browser je lokálne dostupný na:
+
+```text
+http://127.0.0.1:7474/browser/
+```
+
+Kontrola spojenia z API kontajnera:
+
+```bash
+docker compose exec zp-agent-api \
+  python -m scripts.neo4j_smoke
+```
+
 ## Reindexovanie
 
 Celý proces načíta dokumenty, vytvorí chunky, obnoví FTS5 index a
@@ -699,6 +794,51 @@ curl -X POST http://127.0.0.1:8000/rag \
   }'
 ```
 
+## GraphRAG
+
+GraphRAG vetva rozširuje existujúci klasický RAG o knowledge graph.
+Graf sa vytvára deterministicky z už pripravených dokumentov a chunkov
+a ukladá sa do Neo4j.
+
+Aktuálne používané hlavné typy uzlov:
+
+```text
+Person
+Document
+Work
+Topic
+Category
+Author
+```
+
+Hlavné vzťahy zahŕňajú napríklad:
+
+```text
+Person   -[:HAS_DOCUMENT]-> Document
+Person   -[:HAS_WORK]-> Work
+Work     -[:EVIDENCED_BY]-> Document
+Document -[:HAS_TOPIC]-> Topic
+Document -[:IN_CATEGORY]-> Category
+Author   -[:AUTHORED]-> Document
+Document -[:DESCRIBES]-> Topic
+```
+
+Knowledge graph sa vytvorí príkazom:
+
+```bash
+docker compose exec zp-agent-api \
+  python -m scripts.build_graphrag
+```
+
+Builder vytvára constraints, zapisuje uzly a vzťahy a pri štandardnom
+spustení obnoví iba časť grafu spravovanú ZP Agentom. Opakované
+spustenie preto nevytvára duplicitné uzly.
+
+Graf je možné vizuálne kontrolovať v Neo4j Browseri. GraphRAG zatiaľ
+nezasahuje do stabilného klasického `/rag` toku. Nasledujúca etapa
+doplní graph retrieval a následne spojenie grafového a klasického
+retrievalu.
+
 ## OpenWebUI
 
 ZP Agent je pripojený do OpenWebUI ako OpenAPI Tool Server.
@@ -891,10 +1031,9 @@ Pevný počet `pytest` výsledkov nie je v README uvádzaný, pretože sa
 s vývojom mení. Aktuálny stav sa má vždy potvrdiť novým spustením testov
 na konkrétnom commite.
 
-## Pripravené, ale zatiaľ neaktívne experimenty
+## Experimentálne datasety
 
-Nasledujúce datasety sú pripravené, ale nemajú sa teraz používať na
-priebežné ladenie aktuálneho frozen benchmarku:
+Pre ďalšie experimenty sú pripravené:
 
 ```text
 questions_extended_2000.json
@@ -903,104 +1042,53 @@ robustness_questions.json
 performance_queries.json
 ```
 
-Odporúčané poradie ich neskoršieho použitia:
-
-```text
-stabilný klasický RAG
-  ↓
-extended benchmark
-  ↓
-robustness benchmark
-  ↓
-GraphRAG implementácia
-  ↓
-GraphRAG benchmark
-  ↓
-performance benchmark
-  ↓
-stress test
-  ↓
-finálne experimenty
-```
+`graphrag_questions.json` je aktuálne určený pre vývoj GraphRAG vetvy.
+Ostatné datasety zostávajú oddelené od priebežného ladenia frozen
+klasického baseline.
 
 ## Najbližší postup
 
-Najbližšia vývojová etapa je zámerne menšia a má uzavrieť existujúci
-klasický RAG pred otvorením ďalších experimentov.
+Klasický RAG baseline je zmrazený a GraphRAG infraštruktúra s Neo4j je
+funkčná. Ďalší vývoj pokračuje nad samostatnou GraphRAG vetvou.
 
-### 1. Uzavretie retrieval/regresie
+### 1. Dokončenie kvality knowledge graphu
 
-Najprv sa má potvrdiť, že aktuálne zmeny neovplyvnili frozen retrieval
-baseline.
+- doplniť bezpečné fallback pravidlá pre explicitné názvy prác,
+- ponechať `null` tam, kde zdroj názov skutočne neuvádza,
+- doplniť regresné testy pre graph builder.
 
-Postup:
+### 2. Graph retrieval
+
+Vytvoriť vyhľadávaciu vrstvu nad Neo4j, ktorá z otázky identifikuje
+relevantné osoby, témy, kategórie, práce a roky a vráti:
 
 ```text
-py_compile
-↓
-pytest
-↓
-frozen retrieval DEV strict
-↓
-porovnanie s uloženým baseline
-↓
-git diff --check
-↓
-retrieval lock
+grafové entity
+→ grafové cesty
+→ relevantné dokumenty
+→ zdrojové chunky
 ```
 
-Do retrieval scoringu sa potom nemá zasahovať bez nového
-experimentálneho dôvodu.
+### 3. Spojenie s klasickým RAG
 
-### 2. Regresia section-lead RAG kontextu
-
-Treba potvrdiť, že section-lead expanzia:
-
-- opravuje prípady, kde primárny chunk chýba o názov/tému/rok,
-- nepridáva duplicitný chunk,
-- zostáva v rovnakom dokumente a sekcii,
-- nemení retrieval `max_per_document=1`,
-- nevytvára zbytočne veľký kontext.
-
-### 3. `retry + backoff + resume` pre answer evaluator
-
-Pred spustením veľkého DEV answer benchmarku sa má evaluator doplniť
-tak, aby dlhý beh nebol znehodnotený jedným timeoutom alebo dočasnou
-chybou API.
-
-Plánované správanie:
+Grafový retrieval sa následne spojí s existujúcim FTS5 + embedding
+retrievalom. Klasický baseline zostane zachovaný ako samostatný
+referenčný bod.
 
 ```text
-request
+query
   ↓
-úspech ───────────────→ uložiť výsledok
-  │
-  └─ timeout / 429 / 5xx
-           ↓
-       retry
-           ↓
- exponential backoff
-           ↓
-       retry limit
+├── classic retrieval
+└── graph retrieval
+       ↓
+   context fusion
+       ↓
+      RAG
 ```
 
-Resume mechanizmus má:
+### 4. GraphRAG evaluácia
 
-- priebežne ukladať `.partial.json`,
-- pri novom spustení načítať existujúci partial výsledok,
-- overiť kompatibilitu modelu/datasetu/splitu,
-- preskočiť už úspešne dokončené otázky,
-- pokračovať od ďalšej otázky,
-- neprepisovať hotové výsledky bez explicitnej voľby,
-- po úspešnom dokončení vytvoriť finálny JSON/CSV výstup.
-
-Zároveň je vhodné:
-
-- rozlišovať retryable a permanentné chyby,
-- logovať číslo pokusu a dôvod retry,
-- mať konfigurovateľný maximálny počet pokusov,
-- mať konfigurovateľný počiatočný backoff,
-- používať mierny delay medzi otázkami,
-- validovať typy answer-level datasetových polí,
-- nenačítavať osobný OpenWebUI API kľúč z fallback súboru, ak má byť
-  podľa bezpečnostnej politiky dostupný iba cez environment.
+GraphRAG sa bude priebežne ladiť na DEV dátach a vyhodnocovať aj na
+samostatnom `graphrag_questions.json` datasete. Až po stabilizovaní
+celého systému budú nasledovať širšie embeddingové, robustness a
+performance experimenty.

docker-compose.yml

@@ -15,6 +15,10 @@ services:
       CHUNK_MIN_TOKENS: "80"
       CHUNK_TOKEN_ENCODING: cl100k_base
       HF_HOME: /cache/huggingface
+      NEO4J_URI: bolt://neo4j:7687
+      NEO4J_USER: ${NEO4J_USER:-neo4j}
+      NEO4J_PASSWORD: ${NEO4J_PASSWORD}
+      NEO4J_DATABASE: ${NEO4J_DATABASE:-neo4j}
 
     volumes:
       - ./data:/app/data
@@ -23,5 +27,35 @@ services:
 
     restart: unless-stopped
 
+  neo4j:
+    image: neo4j:2026.09.0
+    container_name: zp-agent-neo4j
+
+    ports:
+      - "127.0.0.1:7474:7474"
+      - "127.0.0.1:7687:7687"
+
+    environment:
+      NEO4J_AUTH: ${NEO4J_USER:-neo4j}/${NEO4J_PASSWORD}
+
+    volumes:
+      - neo4j-data:/data
+      - neo4j-logs:/logs
+
+    healthcheck:
+      test:
+        [
+          "CMD-SHELL",
+          "u=$${NEO4J_AUTH%%/*}; p=$${NEO4J_AUTH#*/}; cypher-shell -u \"$$u\" -p \"$$p\" 'RETURN 1' >/dev/null 2>&1"
+        ]
+      interval: 10s
+      timeout: 5s
+      retries: 10
+      start_period: 20s
+
+    restart: unless-stopped
+
 volumes:
   hf-cache:
+  neo4j-data:
+  neo4j-logs:

requirements.txt

@@ -6,3 +6,4 @@ tiktoken>=0.8,<1
 uvicorn[standard]==0.48.0
 numpy==2.2.6
 sentence-transformers==5.7.0
+neo4j==6.3.1

scripts/build_graphrag.py

@@ -0,0 +1,463 @@
+from __future__ import annotations
+
+import argparse
+import json
+from pathlib import Path
+from typing import Any
+
+from scripts.graph_db import (
+    create_neo4j_driver,
+    get_neo4j_settings,
+)
+from scripts.graphrag import (
+    GRAPH_SCOPE,
+    build_graph_payload,
+)
+
+
+PROJECT_ROOT = Path(
+    __file__
+).resolve().parents[1]
+
+DOCUMENTS_FILE = (
+    PROJECT_ROOT
+    / "data"
+    / "documents.json"
+)
+
+CHUNKS_FILE = (
+    PROJECT_ROOT
+    / "data"
+    / "chunks.json"
+)
+
+
+CONSTRAINTS = (
+    (
+        "zpwiki_document_path_unique",
+        "Document",
+        "path",
+    ),
+    (
+        "zpwiki_person_id_unique",
+        "Person",
+        "id",
+    ),
+    (
+        "zpwiki_author_id_unique",
+        "Author",
+        "id",
+    ),
+    (
+        "zpwiki_category_name_unique",
+        "Category",
+        "name",
+    ),
+    (
+        "zpwiki_topic_id_unique",
+        "Topic",
+        "id",
+    ),
+    (
+        "zpwiki_work_id_unique",
+        "Work",
+        "id",
+    ),
+)
+
+
+NODE_QUERIES = {
+    "people": """
+        UNWIND $rows AS row
+        MERGE (n:Person {id: row.id})
+        SET n += row
+    """,
+    "authors": """
+        UNWIND $rows AS row
+        MERGE (n:Author {id: row.id})
+        SET n += row
+    """,
+    "documents": """
+        UNWIND $rows AS row
+        MERGE (n:Document {path: row.path})
+        SET n += row
+    """,
+    "categories": """
+        UNWIND $rows AS row
+        MERGE (n:Category {name: row.name})
+        SET n += row
+    """,
+    "topics": """
+        UNWIND $rows AS row
+        MERGE (n:Topic {id: row.id})
+        SET n += row
+    """,
+    "works": """
+        UNWIND $rows AS row
+        MERGE (n:Work {id: row.id})
+        SET n += row
+    """,
+}
+
+
+RELATIONSHIP_QUERIES = {
+    "person_documents": """
+        UNWIND $rows AS row
+        MATCH (a:Person {id: row.person_id})
+        MATCH (b:Document {
+            path: row.document_path
+        })
+        MERGE (a)-[:HAS_DOCUMENT]->(b)
+    """,
+    "author_documents": """
+        UNWIND $rows AS row
+        MATCH (a:Author {id: row.author_id})
+        MATCH (b:Document {
+            path: row.document_path
+        })
+        MERGE (a)-[:AUTHORED]->(b)
+    """,
+    "document_categories": """
+        UNWIND $rows AS row
+        MATCH (a:Document {
+            path: row.document_path
+        })
+        MATCH (b:Category {
+            name: row.category_name
+        })
+        MERGE (a)-[:IN_CATEGORY]->(b)
+    """,
+    "document_topics": """
+        UNWIND $rows AS row
+        MATCH (a:Document {
+            path: row.document_path
+        })
+        MATCH (b:Topic {
+            id: row.topic_id
+        })
+        MERGE (a)-[:HAS_TOPIC]->(b)
+    """,
+    "document_describes_topics": """
+        UNWIND $rows AS row
+        MATCH (a:Document {
+            path: row.document_path
+        })
+        MATCH (b:Topic {
+            id: row.topic_id
+        })
+        MERGE (a)-[:DESCRIBES]->(b)
+    """,
+    "person_works": """
+        UNWIND $rows AS row
+        MATCH (a:Person {
+            id: row.person_id
+        })
+        MATCH (b:Work {
+            id: row.work_id
+        })
+        MERGE (a)-[:HAS_WORK]->(b)
+    """,
+    "work_documents": """
+        UNWIND $rows AS row
+        MATCH (a:Work {
+            id: row.work_id
+        })
+        MATCH (b:Document {
+            path: row.document_path
+        })
+        MERGE (a)-[:EVIDENCED_BY]->(b)
+    """,
+}
+
+
+def load_json(
+    path: Path,
+) -> list[dict[str, Any]]:
+    if not path.exists():
+        raise FileNotFoundError(
+            f"Missing file: {path}"
+        )
+
+    data = json.loads(
+        path.read_text(
+            encoding="utf-8"
+        )
+    )
+
+    if not isinstance(
+        data,
+        list,
+    ):
+        raise ValueError(
+            f"Expected list in {path}"
+        )
+
+    return data
+
+
+def create_constraints(
+    session,
+) -> None:
+    for (
+        name,
+        label,
+        property_name,
+    ) in CONSTRAINTS:
+        session.run(
+            (
+                f"CREATE CONSTRAINT "
+                f"{name} IF NOT EXISTS "
+                f"FOR (n:{label}) "
+                f"REQUIRE n.{property_name} "
+                f"IS UNIQUE"
+            )
+        ).consume()
+
+
+def clear_managed_graph(
+    session,
+) -> None:
+    session.run(
+        """
+        MATCH (n)
+        WHERE n.graph_scope = $scope
+        DETACH DELETE n
+        """,
+        scope=GRAPH_SCOPE,
+    ).consume()
+
+
+def write_rows(
+    session,
+    query: str,
+    rows: list[dict[str, Any]],
+) -> None:
+    if not rows:
+        return
+
+    session.run(
+        query,
+        rows=rows,
+    ).consume()
+
+
+def graph_counts(
+    session,
+) -> tuple[
+    list[tuple[str, int]],
+    list[tuple[str, int]],
+]:
+    node_rows = session.run(
+        """
+        MATCH (n)
+        WHERE n.graph_scope = $scope
+        UNWIND labels(n) AS label
+        RETURN label, count(*) AS count
+        ORDER BY label
+        """,
+        scope=GRAPH_SCOPE,
+    )
+
+    nodes = [
+        (
+            str(record["label"]),
+            int(record["count"]),
+        )
+        for record in node_rows
+    ]
+
+    relationship_rows = session.run(
+        """
+        MATCH (a)-[r]->(b)
+        WHERE
+            a.graph_scope = $scope
+            AND b.graph_scope = $scope
+        RETURN
+            type(r) AS relationship,
+            count(*) AS count
+        ORDER BY relationship
+        """,
+        scope=GRAPH_SCOPE,
+    )
+
+    relationships = [
+        (
+            str(
+                record[
+                    "relationship"
+                ]
+            ),
+            int(record["count"]),
+        )
+        for record in relationship_rows
+    ]
+
+    return nodes, relationships
+
+
+def work_quality_counts(
+    session,
+) -> tuple[int, int]:
+    record = session.run(
+        """
+        MATCH (w:Work)
+        WHERE w.graph_scope = $scope
+        RETURN
+            count(w) AS total,
+            count(w.title) AS with_title
+        """,
+        scope=GRAPH_SCOPE,
+    ).single()
+
+    if record is None:
+        return 0, 0
+
+    return (
+        int(record["total"]),
+        int(record["with_title"]),
+    )
+
+
+def print_payload_summary(
+    payload: dict[
+        str,
+        list[dict[str, Any]],
+    ],
+) -> None:
+    print()
+    print("Prepared GraphRAG payload")
+    print("=" * 60)
+
+    for key in (
+        "people",
+        "authors",
+        "documents",
+        "categories",
+        "topics",
+        "works",
+    ):
+        print(
+            f"{key:<24}"
+            f"{len(payload[key]):>6}"
+        )
+
+    print("=" * 60)
+
+
+def print_database_summary(
+    session,
+) -> None:
+    nodes, relationships = (
+        graph_counts(
+            session
+        )
+    )
+
+    print()
+    print("Neo4j GraphRAG knowledge graph")
+    print("=" * 60)
+
+    print("Nodes")
+
+    for label, count in nodes:
+        print(
+            f"  {label:<22}"
+            f"{count:>6}"
+        )
+
+    print()
+    print("Relationships")
+
+    for relationship, count in relationships:
+        print(
+            f"  {relationship:<22}"
+            f"{count:>6}"
+        )
+
+    total, with_title = (
+        work_quality_counts(
+            session
+        )
+    )
+
+    print()
+    print(
+        "Works with extracted title: "
+        f"{with_title}/{total}"
+    )
+
+    print("=" * 60)
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser()
+
+    parser.add_argument(
+        "--replace",
+        action=argparse.BooleanOptionalAction,
+        default=True,
+    )
+
+    args = parser.parse_args()
+
+    documents = load_json(
+        DOCUMENTS_FILE
+    )
+
+    chunks = load_json(
+        CHUNKS_FILE
+    )
+
+    payload = build_graph_payload(
+        documents,
+        chunks,
+    )
+
+    print_payload_summary(
+        payload
+    )
+
+    settings = get_neo4j_settings()
+
+    with create_neo4j_driver(
+        settings
+    ) as driver:
+        driver.verify_connectivity()
+
+        with driver.session(
+            database=settings.database
+        ) as session:
+            create_constraints(
+                session
+            )
+
+            if args.replace:
+                clear_managed_graph(
+                    session
+                )
+
+            for key, query in (
+                NODE_QUERIES.items()
+            ):
+                write_rows(
+                    session,
+                    query,
+                    payload[key],
+                )
+
+            for key, query in (
+                RELATIONSHIP_QUERIES.items()
+            ):
+                write_rows(
+                    session,
+                    query,
+                    payload[key],
+                )
+
+            print_database_summary(
+                session
+            )
+
+
+if __name__ == "__main__":
+    main()

scripts/graph_db.py

@@ -0,0 +1,75 @@
+import os
+from dataclasses import dataclass
+
+from neo4j import GraphDatabase
+
+
+@dataclass(frozen=True)
+class Neo4jSettings:
+    uri: str
+    user: str
+    password: str
+    database: str
+
+
+def get_neo4j_settings() -> Neo4jSettings:
+    password = os.getenv("NEO4J_PASSWORD")
+
+    if not password:
+        raise RuntimeError(
+            "NEO4J_PASSWORD is not configured."
+        )
+
+    return Neo4jSettings(
+        uri=os.getenv(
+            "NEO4J_URI",
+            "bolt://localhost:7687",
+        ),
+        user=os.getenv(
+            "NEO4J_USER",
+            "neo4j",
+        ),
+        password=password,
+        database=os.getenv(
+            "NEO4J_DATABASE",
+            "neo4j",
+        ),
+    )
+
+
+def create_neo4j_driver(
+    settings: Neo4jSettings | None = None,
+):
+    resolved = settings or get_neo4j_settings()
+
+    return GraphDatabase.driver(
+        resolved.uri,
+        auth=(
+            resolved.user,
+            resolved.password,
+        ),
+    )
+
+
+def verify_neo4j_connection() -> Neo4jSettings:
+    settings = get_neo4j_settings()
+
+    with create_neo4j_driver(settings) as driver:
+        driver.verify_connectivity()
+
+        with driver.session(
+            database=settings.database
+        ) as session:
+            record = session.run(
+                "RETURN 1 AS value"
+            ).single()
+
+            if (
+                record is None
+                or record["value"] != 1
+            ):
+                raise RuntimeError(
+                    "Neo4j connectivity check failed."
+                )
+
+    return settings

scripts/neo4j_smoke.py

@@ -0,0 +1,14 @@
+from scripts.graph_db import verify_neo4j_connection
+
+
+def main() -> None:
+    settings = verify_neo4j_connection()
+
+    print("Neo4j connection: OK")
+    print(f"URI: {settings.uri}")
+    print(f"Database: {settings.database}")
+    print(f"User: {settings.user}")
+
+
+if __name__ == "__main__":
+    main()