Deploy TechDocs GraphRAG

.dockerignore

@@ -0,0 +1,11 @@
+.env
+__pycache__/
+*.py[cod]
+.pytest_cache/
+.ruff_cache/
+.venv/
+*.egg-info/
+dist/
+build/
+tests/
+.git/

.env.example

@@ -0,0 +1,8 @@
+OPENAI_API_KEY=sk-...
+CHAT_MODEL=gpt-4o-mini
+EMBEDDING_MODEL=text-embedding-3-small
+
+NEO4J_URI=neo4j+s://xxxxxxxx.databases.neo4j.io
+NEO4J_USERNAME=neo4j
+NEO4J_PASSWORD=...
+NEO4J_DATABASE=neo4j

.gitignore

@@ -0,0 +1,9 @@
+.env
+__pycache__/
+*.py[cod]
+.pytest_cache/
+.ruff_cache/
+.venv/
+*.egg-info/
+dist/
+build/

.streamlit/config.toml

@@ -0,0 +1,9 @@
+[server]
+headless = true
+maxUploadSize = 20
+
+[browser]
+gatherUsageStats = false
+
+[theme]
+primaryColor = "#4F46E5"

Dockerfile

@@ -0,0 +1,39 @@
+FROM python:3.12-slim-bookworm
+
+ARG NEO4J_VERSION=5.26.0
+
+ENV PYTHONDONTWRITEBYTECODE=1 \
+    PYTHONUNBUFFERED=1 \
+    PIP_NO_CACHE_DIR=1
+
+RUN apt-get update \
+    && apt-get install -y --no-install-recommends openjdk-17-jre-headless curl \
+    && rm -rf /var/lib/apt/lists/*
+
+RUN useradd --create-home --uid 1000 user
+USER user
+ENV HOME=/home/user
+ENV NEO4J_HOME=$HOME/neo4j \
+    PATH=$HOME/neo4j/bin:$HOME/.local/bin:$PATH \
+    NEO4J_URI=bolt://localhost:7687 \
+    NEO4J_USERNAME=neo4j
+WORKDIR $HOME
+
+RUN curl -fsSL "https://dist.neo4j.org/neo4j-community-${NEO4J_VERSION}-unix.tar.gz" | tar -xz \
+    && mv "neo4j-community-${NEO4J_VERSION}" neo4j \
+    && printf '%s\n' \
+        'server.memory.heap.initial_size=512m' \
+        'server.memory.heap.max_size=1g' \
+        'server.memory.pagecache.size=512m' \
+        'dbms.usage_report.enabled=false' >> neo4j/conf/neo4j.conf
+
+WORKDIR $HOME/app
+COPY --chown=user pyproject.toml README.md ./
+COPY --chown=user src ./src
+RUN pip install --user --upgrade pip && pip install --user .
+
+COPY --chown=user . .
+RUN chmod +x entrypoint.sh
+
+EXPOSE 7860
+CMD ["./entrypoint.sh"]

README.md

@@ -1,10 +1,111 @@
 ---
-title: Techdocs Graphrag
-emoji: 🌖
-colorFrom: purple
-colorTo: gray
+title: TechDocs GraphRAG
+emoji: 🕸️
+colorFrom: indigo
+colorTo: blue
 sdk: docker
+app_port: 7860
 pinned: false
+license: mit
+short_description: GraphRAG sur documentation technique (Neo4j + OpenAI)
 ---
 
-Check out the configuration reference at https://huggingface.co/docs/hub/spaces-config-reference
+# 🕸️ TechDocs GraphRAG
+
+Assistant de documentation technique qui combine **recherche vectorielle** et **graphe de connaissances Neo4j**
+pour répondre à des questions multi-hop avec des sources traçables.
+
+Un RAG classique retrouve des passages similaires à la question. TechDocs GraphRAG extrait en plus les
+entités et relations du corpus (`StateGraph —PERSISTS_WITH→ Checkpointer —ENABLES→ Persistence`) et
+suit ces liens au moment de répondre.
+
+## Fonctionnalités
+
+- Ingestion Markdown, découpage par paragraphes avec chevauchement.
+- Extraction LLM d'entités typées et de relations avec preuve textuelle.
+- Déduplication des entités par clé canonique (`State Graph`, `state_graph` → `stategraph`).
+- Neo4j Community embarqué dans le conteneur, corpus indexé automatiquement au démarrage.
+- Neo4j comme store unique : documents, chunks, index vectoriel, index plein texte, graphe.
+- Recherche hybride : chunks proches → entités mentionnées + entités citées dans la question → traversée multi-hop.
+- Interface Streamlit : sous-graphe utilisé, relations, extraits, sources.
+- Mode comparaison RAG vectoriel vs GraphRAG.
+
+## Architecture
+
+```mermaid
+flowchart LR
+    A[Markdown] --> B[ParagraphSplitter]
+    B --> C[Embeddings OpenAI]
+    B --> D[Extraction LLM]
+    C --> E[(Neo4j<br/>index vectoriel)]
+    D --> F[(Neo4j<br/>graphe d'entités)]
+    Q[Question] --> R[HybridRetriever]
+    E --> R
+    F --> R
+    R --> G[GraphRAGAssistant]
+    G --> S[Réponse sourcée]
+```
+
+### Modèle de graphe
+
+```text
+(:Document)-[:HAS_CHUNK]->(:Chunk {embedding})-[:MENTIONS]->(:Entity:Component)
+(:Entity)-[:USES | PERSISTS_WITH | DEPENDS_ON | ...]->(:Entity)
+```
+
+Chaque relation porte `evidence` et `documents`, ce qui rend chaque fait vérifiable.
+
+## Structure
+
+```text
+├── app.py
+├── Dockerfile
+├── data/
+├── src/techdocs_graphrag/
+│   ├── config.py
+│   ├── models.py
+│   ├── ports.py
+│   ├── ingestion.py
+│   ├── llm.py
+│   ├── prompts.py
+│   ├── extraction.py
+│   ├── graph_store.py
+│   ├── indexing.py
+│   ├── retrieval.py
+│   ├── answering.py
+│   ├── visualization.py
+│   ├── services.py
+│   └── cli.py
+└── tests/
+```
+
+## Configuration
+
+| Variable | Défaut | Rôle |
+|---|---|---|
+| `OPENAI_API_KEY` | requis | Clé OpenAI |
+| `NEO4J_URI` | `bolt://localhost:7687` | Neo4j embarqué dans le conteneur |
+| `NEO4J_USERNAME` | `neo4j` | Utilisateur Neo4j |
+| `NEO4J_PASSWORD` | `techdocs-local-password` | Mot de passe Neo4j (secret conseillé) |
+| `NEO4J_DATABASE` | `neo4j` | Base Neo4j |
+| `CHAT_MODEL` | `gpt-4o-mini` | Modèle de génération |
+| `EMBEDDING_MODEL` | `text-embedding-3-small` | Modèle d'embeddings |
+| `GRAPH_MAX_HOPS` | `2` | Profondeur de traversée |
+| `VECTOR_TOP_K` | `5` | Nombre de chunks récupérés |
+
+## Lancer en local
+
+```bash
+cp .env.example .env
+pip install -e ".[dev]"
+pytest
+techdocs-index data --reset
+streamlit run app.py
+```
+
+## Feuille de route
+
+- Jeu d'évaluation annoté (30 à 50 questions) et métriques de fidélité.
+- Fusion d'entités par similarité d'embeddings.
+- Résumés de communautés (approche Microsoft GraphRAG) pour les questions globales.
+- API FastAPI `POST /ask`.

app.py

@@ -0,0 +1,182 @@
+from collections.abc import Sequence
+from pathlib import Path
+
+import streamlit as st
+from neo4j.exceptions import DriverError, Neo4jError
+from pydantic import ValidationError
+
+from techdocs_graphrag import (
+    Answer,
+    RetrievalMode,
+    Services,
+    SourceDocument,
+    build_services,
+    load_documents,
+)
+from techdocs_graphrag.visualization import facts_to_dot
+
+SAMPLE_CORPUS = Path(__file__).parent / "data"
+
+MODE_LABELS = {
+    RetrievalMode.GRAPH: "GraphRAG",
+    RetrievalMode.VECTOR: "RAG vectoriel",
+}
+
+DISPLAY_MODES: dict[str, tuple[RetrievalMode, ...]] = {
+    "GraphRAG": (RetrievalMode.GRAPH,),
+    "RAG vectoriel": (RetrievalMode.VECTOR,),
+    "Comparaison côte à côte": (RetrievalMode.VECTOR, RetrievalMode.GRAPH),
+}
+
+EXAMPLE_QUESTIONS = (
+    "Comment persister l'état d'un workflow StateGraph ?",
+    "Quel est le rôle des reducers dans les channels ?",
+    "Comment reprendre un workflow après une interruption humaine ?",
+)
+
+st.set_page_config(page_title="TechDocs GraphRAG", page_icon="🕸️", layout="wide")
+
+
+@st.cache_resource(show_spinner="Connexion à Neo4j et OpenAI…")
+def get_services() -> Services:
+    services = build_services()
+    services.store.verify_connectivity()
+    services.store.ensure_schema()
+    return services
+
+
+def load_services() -> Services:
+    try:
+        return get_services()
+    except ValidationError as error:
+        missing = ", ".join(sorted({str(item["loc"][0]).upper() for item in error.errors()}))
+        st.error(f"Configuration incomplète. Variables à définir dans les secrets : {missing}")
+    except (Neo4jError, DriverError) as error:
+        st.error(f"Connexion à Neo4j impossible : {error}")
+    st.stop()
+
+
+def index_documents(services: Services, documents: Sequence[SourceDocument]) -> None:
+    progress = st.progress(0.0, text="Calcul des embeddings…")
+
+    def report_progress(done: int, total: int) -> None:
+        progress.progress(done / total, text=f"Extraction du graphe : {done}/{total} chunks")
+
+    report = services.indexer.run(documents, on_progress=report_progress)
+    progress.empty()
+    st.success(
+        f"{report.documents} documents · {report.chunks} chunks · "
+        f"{report.entities} entités · {report.relations} relations"
+    )
+
+
+def render_sidebar(services: Services) -> tuple[RetrievalMode, ...]:
+    with st.sidebar:
+        st.header("Base de connaissances")
+
+        uploads = st.file_uploader(
+            "Ajouter des documents",
+            type=["md", "markdown", "txt"],
+            accept_multiple_files=True,
+        )
+        if st.button("Indexer les documents ajoutés", disabled=not uploads):
+            index_documents(
+                services,
+                [SourceDocument(name=file.name, text=file.getvalue().decode("utf-8")) for file in uploads],
+            )
+        if st.button("Indexer le corpus d'exemple"):
+            index_documents(services, load_documents(SAMPLE_CORPUS))
+
+        st.divider()
+        display_mode = st.radio("Mode de recherche", options=list(DISPLAY_MODES))
+
+        st.divider()
+        render_stats(services)
+        if st.button("Effacer la conversation"):
+            st.session_state.turns = []
+            st.rerun()
+
+    return DISPLAY_MODES[display_mode]
+
+
+def render_stats(services: Services) -> None:
+    stats = services.store.stats()
+    first, second = st.columns(2)
+    first.metric("Documents", stats["documents"])
+    second.metric("Chunks", stats["chunks"])
+    first.metric("Entités", stats["entities"])
+    second.metric("Relations", stats["relations"])
+
+
+def render_answer(answer: Answer) -> None:
+    st.markdown(answer.text)
+    context = answer.context
+    if context.sources:
+        st.caption("Sources : " + " · ".join(context.sources))
+
+    with st.expander(f"Contexte · {len(context.chunks)} extraits · {len(context.facts)} relations"):
+        graph_tab, facts_tab, chunks_tab = st.tabs(["Sous-graphe", "Relations", "Extraits"])
+        with graph_tab:
+            if context.facts:
+                st.graphviz_chart(facts_to_dot(context.facts, context.seed_entities))
+            else:
+                st.info("Aucune relation du graphe n'a été utilisée.")
+        with facts_tab:
+            if context.facts:
+                st.dataframe([fact.model_dump() for fact in context.facts], hide_index=True)
+            else:
+                st.info("Aucune relation.")
+        with chunks_tab:
+            for chunk in context.chunks:
+                st.markdown(f"**{chunk.document}** · chunk {chunk.position} · score {chunk.score:.3f}")
+                st.text(chunk.text)
+
+
+def render_turn(answers: Sequence[Answer]) -> None:
+    with st.chat_message("user"):
+        st.markdown(answers[0].question)
+    with st.chat_message("assistant"):
+        for column, answer in zip(st.columns(len(answers)), answers, strict=True):
+            with column:
+                if len(answers) > 1:
+                    st.subheader(MODE_LABELS[answer.mode])
+                render_answer(answer)
+
+
+def pick_example_question() -> str | None:
+    st.markdown("##### Exemples de questions")
+    columns = st.columns(len(EXAMPLE_QUESTIONS))
+    clicked = [
+        question
+        for column, question in zip(columns, EXAMPLE_QUESTIONS, strict=True)
+        if column.button(question)
+    ]
+    return clicked[0] if clicked else None
+
+
+def main() -> None:
+    services = load_services()
+    modes = render_sidebar(services)
+
+    st.title("🕸️ TechDocs GraphRAG")
+    st.caption(
+        "Assistant de documentation technique : recherche vectorielle et graphe de connaissances "
+        "Neo4j pour des réponses multi-hop sourcées."
+    )
+
+    turns: list[list[Answer]] = st.session_state.setdefault("turns", [])
+    for answers in turns:
+        render_turn(answers)
+
+    typed_question = st.chat_input("Pose une question sur la documentation…")
+    question = typed_question or (None if turns else pick_example_question())
+    if not question:
+        return
+
+    with st.spinner("Recherche dans le graphe et les documents…"):
+        answers = [services.assistant.ask(question, mode) for mode in modes]
+    turns.append(answers)
+    render_turn(answers)
+
+
+main()

data/channels.md

@@ -0,0 +1,20 @@
+# Channels et reducers
+
+Un Channel stocke la valeur d'une clé de l'état d'un StateGraph.
+Par défaut, un Channel remplace l'ancienne valeur par la nouvelle valeur renvoyée par un Node.
+
+Un Reducer définit comment un Channel combine l'ancienne valeur et la nouvelle.
+Un Reducer se déclare avec typing.Annotated dans le schéma d'état.
+
+La fonction add_messages est un Reducer fourni par LangGraph.
+add_messages ajoute les nouveaux messages à la liste existante au lieu de la remplacer.
+
+```python
+from typing import Annotated, TypedDict
+from langgraph.graph.message import add_messages
+
+class State(TypedDict):
+    messages: Annotated[list, add_messages]
+```
+
+MessagesState est un schéma d'état prédéfini qui utilise add_messages sur la clé messages.

data/checkpointer.md

@@ -0,0 +1,22 @@
+# Checkpointer
+
+Un Checkpointer sauvegarde un Checkpoint de l'état du graphe après chaque étape d'exécution.
+Le Checkpointer active la Persistence, le Human-in-the-loop, le Time Travel et la mémoire conversationnelle.
+
+Les checkpoints sont regroupés par Thread.
+Un Thread est identifié par le paramètre thread_id passé dans la configuration d'exécution.
+
+```python
+config = {"configurable": {"thread_id": "conversation-42"}}
+graph.invoke({"messages": [("user", "Bonjour")]}, config)
+```
+
+InMemorySaver est un Checkpointer qui stocke les checkpoints en mémoire. Il convient aux tests.
+SqliteSaver est un Checkpointer qui stocke les checkpoints dans une base SQLite.
+PostgresSaver est un Checkpointer qui stocke les checkpoints dans PostgreSQL. Il est recommandé en production.
+
+SqliteSaver dépend du package langgraph-checkpoint-sqlite.
+PostgresSaver dépend du package langgraph-checkpoint-postgres.
+
+La méthode get_state d'un CompiledGraph lit le dernier Checkpoint d'un Thread.
+La méthode get_state_history liste tous les checkpoints d'un Thread.

data/nodes.md

@@ -0,0 +1,12 @@
+# Nodes et Edges
+
+Un Node est une fonction Python qui reçoit l'état courant et renvoie une mise à jour partielle de l'état.
+Les mises à jour renvoyées par un Node sont appliquées aux Channels via leurs Reducers.
+
+Un Edge relie deux Nodes et définit l'ordre d'exécution.
+START et END sont des Nodes virtuels qui marquent l'entrée et la sortie du graphe.
+
+Un Conditional Edge utilise une fonction de routage pour choisir dynamiquement le Node suivant.
+L'API Send permet à un Conditional Edge de lancer plusieurs Nodes en parallèle (map-reduce).
+
+La classe Command permet à un Node de mettre à jour l'état et de choisir le Node suivant en une seule instruction.

data/stategraph.md

@@ -0,0 +1,27 @@
+# StateGraph
+
+StateGraph est la classe centrale de LangGraph pour construire des workflows orientés état.
+Un StateGraph est paramétré par un schéma d'état, généralement un TypedDict ou un modèle Pydantic.
+
+StateGraph utilise des Channels pour stocker et transmettre l'état entre les nodes.
+Chaque clé du schéma d'état correspond à un Channel.
+
+Un StateGraph contient des Nodes reliés par des Edges.
+La méthode add_node enregistre un Node, la méthode add_edge crée un Edge fixe
+et la méthode add_conditional_edges crée un routage dynamique.
+
+Avant l'exécution, un StateGraph doit être compilé avec la méthode compile.
+La méthode compile retourne un CompiledGraph qui expose invoke, stream et get_state.
+
+La méthode compile accepte un paramètre checkpointer.
+Un StateGraph compilé avec un Checkpointer persiste son état entre plusieurs exécutions.
+
+```python
+from langgraph.graph import StateGraph, START, END
+
+builder = StateGraph(State)
+builder.add_node("agent", call_model)
+builder.add_edge(START, "agent")
+builder.add_edge("agent", END)
+graph = builder.compile()
+```

data/tutorials/human_in_the_loop.md

@@ -0,0 +1,22 @@
+# Tutoriel : Human-in-the-loop
+
+Le Human-in-the-loop permet de mettre un workflow en pause pour obtenir une validation humaine.
+
+La fonction interrupt suspend l'exécution d'un Node et renvoie une valeur à l'appelant.
+La fonction interrupt nécessite un Checkpointer, car l'état doit être sauvegardé pendant la pause.
+
+Pour reprendre, on invoque le graphe avec Command(resume=valeur) et le même thread_id.
+Le Node interrompu redémarre et interrupt renvoie la valeur fournie par l'humain.
+
+```python
+from langgraph.types import interrupt, Command
+
+def review(state):
+    decision = interrupt({"draft": state["draft"]})
+    return {"approved": decision == "ok"}
+
+graph.invoke(Command(resume="ok"), config)
+```
+
+Les paramètres interrupt_before et interrupt_after de compile permettent aussi de s'arrêter
+avant ou après des Nodes précis, principalement pour le débogage.

data/tutorials/multi_agent.md

@@ -0,0 +1,13 @@
+# Tutoriel : systèmes multi-agents
+
+Un système multi-agents LangGraph combine plusieurs agents, chacun implémenté comme un Node ou un Subgraph.
+
+Un Subgraph est un StateGraph compilé utilisé comme Node d'un autre StateGraph.
+Un Subgraph partage les clés d'état communes avec le graphe parent.
+
+Dans l'architecture Supervisor, un agent Supervisor route les tâches vers des agents spécialisés.
+Le Supervisor utilise la classe Command pour désigner l'agent suivant.
+
+Dans l'architecture Swarm, les agents se transmettent le contrôle directement par handoff.
+
+Un Checkpointer défini sur le graphe parent est propagé automatiquement aux Subgraphs.

data/tutorials/persistence.md

@@ -0,0 +1,22 @@
+# Tutoriel : persister un workflow
+
+Ce tutoriel montre comment rendre un StateGraph persistant.
+
+Étape 1 : choisir un Checkpointer. En développement, InMemorySaver suffit.
+En production, PostgresSaver garantit que l'état survit au redémarrage du serveur.
+
+Étape 2 : passer le Checkpointer à la méthode compile.
+
+```python
+from langgraph.checkpoint.memory import InMemorySaver
+
+graph = builder.compile(checkpointer=InMemorySaver())
+```
+
+Étape 3 : fournir un thread_id à chaque appel.
+Deux appels avec le même thread_id partagent le même état ; le Channel messages conserve l'historique
+grâce au Reducer add_messages.
+
+Étape 4 : inspecter l'état avec get_state pour vérifier ce qui a été sauvegardé.
+
+Sans Checkpointer, un StateGraph repart d'un état vide à chaque invocation.

entrypoint.sh

@@ -0,0 +1,12 @@
+#!/usr/bin/env bash
+set -euo pipefail
+
+export NEO4J_PASSWORD="${NEO4J_PASSWORD:-techdocs-local-password}"
+
+neo4j-admin dbms set-initial-password "$NEO4J_PASSWORD" > /dev/null 2>&1 || true
+neo4j start
+
+techdocs-index data --wait 120 --if-empty \
+    || echo "Indexation initiale impossible : l'application démarre sans corpus."
+
+exec streamlit run app.py --server.port=7860 --server.address=0.0.0.0

pyproject.toml

@@ -0,0 +1,40 @@
+[build-system]
+requires = ["hatchling>=1.24"]
+build-backend = "hatchling.build"
+
+[project]
+name = "techdocs-graphrag"
+version = "0.1.0"
+description = "Assistant de documentation technique : recherche hybride vectorielle et graphe de connaissances Neo4j."
+readme = "README.md"
+requires-python = ">=3.12"
+license = { text = "MIT" }
+dependencies = [
+    "neo4j>=5.20",
+    "openai>=1.40",
+    "pydantic>=2.7",
+    "pydantic-settings>=2.3",
+    "streamlit>=1.38",
+    "tenacity>=8.3",
+]
+
+[project.optional-dependencies]
+dev = ["pytest>=8.2", "ruff>=0.6"]
+
+[project.scripts]
+techdocs-index = "techdocs_graphrag.cli:main"
+
+[tool.hatch.build.targets.wheel]
+packages = ["src/techdocs_graphrag"]
+
+[tool.pytest.ini_options]
+pythonpath = ["src"]
+testpaths = ["tests"]
+
+[tool.ruff]
+line-length = 100
+target-version = "py312"
+
+[tool.ruff.lint]
+select = ["E", "F", "I", "B", "UP", "SIM", "RUF"]
+ignore = ["E501", "RUF001", "RUF002", "RUF003"]

src/techdocs_graphrag/__init__.py

@@ -0,0 +1,14 @@
+from techdocs_graphrag.ingestion import SourceDocument, load_documents
+from techdocs_graphrag.models import Answer, RetrievalMode
+from techdocs_graphrag.services import Services, build_services
+
+__version__ = "0.1.0"
+
+__all__ = [
+    "Answer",
+    "RetrievalMode",
+    "Services",
+    "SourceDocument",
+    "build_services",
+    "load_documents",
+]

src/techdocs_graphrag/answering.py

@@ -0,0 +1,24 @@
+from techdocs_graphrag.models import Answer, RetrievalMode
+from techdocs_graphrag.ports import LanguageModel
+from techdocs_graphrag.prompts import ANSWER_SYSTEM_PROMPT, render_answer_prompt
+from techdocs_graphrag.retrieval import HybridRetriever
+
+NO_CONTEXT_ANSWER = (
+    "Je ne trouve aucune information pertinente dans la base indexée. "
+    "Indexe d'abord un corpus ou reformule la question."
+)
+
+
+class GraphRAGAssistant:
+    def __init__(self, *, retriever: HybridRetriever, llm: LanguageModel) -> None:
+        self._retriever = retriever
+        self._llm = llm
+
+    def ask(self, question: str, mode: RetrievalMode = RetrievalMode.GRAPH) -> Answer:
+        context = self._retriever.retrieve(question, mode)
+        text = (
+            NO_CONTEXT_ANSWER
+            if context.is_empty
+            else self._llm.complete(ANSWER_SYSTEM_PROMPT, render_answer_prompt(question, context))
+        )
+        return Answer(question=question, text=text, mode=mode, context=context)

src/techdocs_graphrag/cli.py

@@ -0,0 +1,56 @@
+import argparse
+import logging
+import time
+from pathlib import Path
+
+from neo4j.exceptions import DriverError, Neo4jError
+
+from techdocs_graphrag.ingestion import load_documents
+from techdocs_graphrag.services import Services, build_services
+
+RETRY_DELAY_SECONDS = 2
+
+
+def parse_arguments() -> argparse.Namespace:
+    parser = argparse.ArgumentParser(description="Indexe un corpus Markdown dans Neo4j.")
+    parser.add_argument("data_dir", type=Path, nargs="?", default=Path("data"))
+    parser.add_argument("--reset", action="store_true", help="Vide la base avant indexation.")
+    parser.add_argument("--if-empty", action="store_true", help="N'indexe que si la base est vide.")
+    parser.add_argument("--wait", type=int, default=0, metavar="SECONDS", help="Attend Neo4j.")
+    return parser.parse_args()
+
+
+def wait_for_database(services: Services, timeout_seconds: int) -> None:
+    deadline = time.monotonic() + timeout_seconds
+    while True:
+        try:
+            services.store.verify_connectivity()
+            return
+        except (DriverError, Neo4jError):
+            if time.monotonic() >= deadline:
+                raise
+            time.sleep(RETRY_DELAY_SECONDS)
+
+
+def main() -> None:
+    logging.basicConfig(level=logging.INFO, format="%(levelname)s %(name)s: %(message)s")
+    arguments = parse_arguments()
+    services = build_services()
+    try:
+        wait_for_database(services, arguments.wait)
+        if arguments.if_empty and services.store.stats()["chunks"]:
+            print("Base déjà indexée.")
+            return
+        if arguments.reset:
+            services.store.reset()
+        report = services.indexer.run(
+            load_documents(arguments.data_dir),
+            on_progress=lambda done, total: print(f"\r{done}/{total} chunks", end="", flush=True),
+        )
+        print(f"\n{report}")
+    finally:
+        services.close()
+
+
+if __name__ == "__main__":
+    main()

src/techdocs_graphrag/config.py

@@ -0,0 +1,32 @@
+from functools import lru_cache
+
+from pydantic import Field, SecretStr
+from pydantic_settings import BaseSettings, SettingsConfigDict
+
+
+class Settings(BaseSettings):
+    model_config = SettingsConfigDict(env_file=".env", env_file_encoding="utf-8", extra="ignore")
+
+    openai_api_key: SecretStr
+    chat_model: str = "gpt-4o-mini"
+    chat_temperature: float | None = 0.0
+    embedding_model: str = "text-embedding-3-small"
+    embedding_dimensions: int = Field(default=1536, gt=0)
+
+    neo4j_uri: str
+    neo4j_username: str = "neo4j"
+    neo4j_password: SecretStr
+    neo4j_database: str = "neo4j"
+
+    chunk_size: int = Field(default=900, gt=0)
+    chunk_overlap: int = Field(default=150, ge=0)
+    extraction_workers: int = Field(default=4, ge=1, le=16)
+
+    vector_top_k: int = Field(default=5, ge=1, le=50)
+    graph_max_hops: int = Field(default=2, ge=1, le=4)
+    graph_max_facts: int = Field(default=40, ge=1, le=200)
+
+
+@lru_cache
+def get_settings() -> Settings:
+    return Settings()

src/techdocs_graphrag/extraction.py

@@ -0,0 +1,24 @@
+import logging
+
+from pydantic import ValidationError
+
+from techdocs_graphrag.models import Chunk, GraphExtraction
+from techdocs_graphrag.ports import LanguageModel
+from techdocs_graphrag.prompts import EXTRACTION_SYSTEM_PROMPT
+
+logger = logging.getLogger(__name__)
+
+
+class GraphExtractor:
+    def __init__(self, llm: LanguageModel) -> None:
+        self._llm = llm
+
+    def extract(self, chunk: Chunk) -> GraphExtraction:
+        payload = self._llm.complete(EXTRACTION_SYSTEM_PROMPT, chunk.text, json_mode=True)
+        try:
+            return GraphExtraction.model_validate_json(payload)
+        except ValidationError as error:
+            logger.warning(
+                "Graph extraction rejected for chunk %s (%s errors)", chunk.id, error.error_count()
+            )
+            return GraphExtraction()

src/techdocs_graphrag/graph_store.py

@@ -0,0 +1,226 @@
+import re
+from collections import defaultdict
+from collections.abc import Sequence
+from itertools import batched
+from typing import Any, Self
+
+from neo4j import GraphDatabase
+
+from techdocs_graphrag.config import Settings
+from techdocs_graphrag.models import Chunk, GraphExtraction, GraphFact, RetrievedChunk
+
+VECTOR_INDEX = "chunk_embeddings"
+FULLTEXT_INDEX = "entity_names"
+WRITE_BATCH_SIZE = 200
+MIN_SEARCH_TERM_LENGTH = 3
+INDEX_TIMEOUT_SECONDS = 300
+CYPHER_IDENTIFIER = re.compile(r"^[A-Za-z][A-Za-z0-9_]*$")
+
+
+def cypher_identifier(value: str) -> str:
+    if not CYPHER_IDENTIFIER.fullmatch(value):
+        raise ValueError(f"Unsafe Cypher identifier: {value!r}")
+    return value
+
+
+class Neo4jGraphStore:
+    def __init__(self, settings: Settings) -> None:
+        self._driver = GraphDatabase.driver(
+            settings.neo4j_uri,
+            auth=(settings.neo4j_username, settings.neo4j_password.get_secret_value()),
+        )
+        self._database = settings.neo4j_database
+        self._embedding_dimensions = settings.embedding_dimensions
+
+    def __enter__(self) -> Self:
+        return self
+
+    def __exit__(self, *exc_info: object) -> None:
+        self.close()
+
+    def close(self) -> None:
+        self._driver.close()
+
+    def verify_connectivity(self) -> None:
+        self._driver.verify_connectivity()
+
+    def ensure_schema(self) -> None:
+        statements = (
+            "CREATE CONSTRAINT document_name IF NOT EXISTS "
+            "FOR (d:Document) REQUIRE d.name IS UNIQUE",
+            "CREATE CONSTRAINT chunk_id IF NOT EXISTS FOR (c:Chunk) REQUIRE c.id IS UNIQUE",
+            "CREATE CONSTRAINT entity_key IF NOT EXISTS FOR (e:Entity) REQUIRE e.key IS UNIQUE",
+            f"CREATE FULLTEXT INDEX {FULLTEXT_INDEX} IF NOT EXISTS "
+            "FOR (e:Entity) ON EACH [e.name]",
+            f"""
+            CREATE VECTOR INDEX {VECTOR_INDEX} IF NOT EXISTS
+            FOR (c:Chunk) ON c.embedding
+            OPTIONS {{indexConfig: {{
+                `vector.dimensions`: {int(self._embedding_dimensions)},
+                `vector.similarity_function`: 'cosine'
+            }}}}
+            """,
+        )
+        for statement in statements:
+            self._run(statement)
+
+    def await_indexes(self) -> None:
+        self._run("CALL db.awaitIndexes($timeout)", timeout=INDEX_TIMEOUT_SECONDS)
+
+    def upsert_chunks(self, chunks: Sequence[Chunk], embeddings: Sequence[list[float]]) -> None:
+        rows = [
+            {**chunk.model_dump(), "embedding": embedding}
+            for chunk, embedding in zip(chunks, embeddings, strict=True)
+        ]
+        for batch in batched(rows, WRITE_BATCH_SIZE):
+            self._run(
+                """
+                UNWIND $rows AS row
+                MERGE (d:Document {name: row.document})
+                MERGE (c:Chunk {id: row.id})
+                SET c.document = row.document,
+                    c.position = row.position,
+                    c.text = row.text,
+                    c.embedding = row.embedding
+                MERGE (d)-[:HAS_CHUNK]->(c)
+                """,
+                rows=list(batch),
+            )
+
+    def upsert_extraction(self, chunk: Chunk, extraction: GraphExtraction) -> None:
+        self._upsert_entities(chunk, extraction)
+        self._upsert_relations(chunk, extraction)
+
+    def _upsert_entities(self, chunk: Chunk, extraction: GraphExtraction) -> None:
+        rows_by_type: dict[str, list[dict[str, str]]] = defaultdict(list)
+        for entity in extraction.entities:
+            rows_by_type[entity.type.value].append({"key": entity.key, "name": entity.name})
+        for entity_type, rows in rows_by_type.items():
+            self._run(
+                f"""
+                MATCH (c:Chunk {{id: $chunk_id}})
+                UNWIND $rows AS row
+                MERGE (e:Entity {{key: row.key}})
+                ON CREATE SET e.name = row.name, e.type = $entity_type
+                SET e:{cypher_identifier(entity_type)}
+                MERGE (c)-[:MENTIONS]->(e)
+                """,
+                chunk_id=chunk.id,
+                rows=rows,
+                entity_type=entity_type,
+            )
+
+    def _upsert_relations(self, chunk: Chunk, extraction: GraphExtraction) -> None:
+        rows_by_type: dict[str, list[dict[str, str]]] = defaultdict(list)
+        for relation in extraction.relations:
+            rows_by_type[relation.relation].append(
+                {
+                    "source": relation.source_key,
+                    "target": relation.target_key,
+                    "evidence": relation.evidence,
+                    "document": chunk.document,
+                }
+            )
+        for relation_type, rows in rows_by_type.items():
+            self._run(
+                f"""
+                UNWIND $rows AS row
+                MATCH (s:Entity {{key: row.source}})
+                MATCH (t:Entity {{key: row.target}})
+                MERGE (s)-[r:{cypher_identifier(relation_type)}]->(t)
+                ON CREATE SET r.evidence = [], r.documents = []
+                SET r.evidence = CASE
+                        WHEN row.evidence = '' OR row.evidence IN r.evidence THEN r.evidence
+                        ELSE r.evidence + row.evidence
+                    END,
+                    r.documents = CASE
+                        WHEN row.document IN r.documents THEN r.documents
+                        ELSE r.documents + row.document
+                    END
+                """,
+                rows=rows,
+            )
+
+    def vector_search(self, embedding: list[float], top_k: int) -> list[RetrievedChunk]:
+        rows = self._run(
+            """
+            CALL db.index.vector.queryNodes($index, $top_k, $embedding) YIELD node, score
+            OPTIONAL MATCH (node)-[:MENTIONS]->(entity:Entity)
+            RETURN node.id AS id,
+                   node.document AS document,
+                   node.position AS position,
+                   node.text AS text,
+                   score,
+                   collect(entity.key) AS entity_keys
+            ORDER BY score DESC
+            """,
+            index=VECTOR_INDEX,
+            top_k=top_k,
+            embedding=embedding,
+        )
+        return [RetrievedChunk.model_validate(row) for row in rows]
+
+    def search_entities(self, text: str, limit: int) -> list[str]:
+        terms = sorted(
+            {term.lower() for term in re.findall(r"\w+", text) if len(term) >= MIN_SEARCH_TERM_LENGTH}
+        )
+        if not terms:
+            return []
+        rows = self._run(
+            """
+            CALL db.index.fulltext.queryNodes($index, $query) YIELD node, score
+            RETURN node.key AS key
+            ORDER BY score DESC
+            LIMIT $limit
+            """,
+            index=FULLTEXT_INDEX,
+            query=" OR ".join(terms),
+            limit=limit,
+        )
+        return [row["key"] for row in rows]
+
+    def expand(self, seed_keys: Sequence[str], max_hops: int, limit: int) -> list[GraphFact]:
+        if not seed_keys:
+            return []
+        rows = self._run(
+            f"""
+            MATCH (seed:Entity) WHERE seed.key IN $keys
+            MATCH path = (seed)-[*1..{int(max_hops)}]-(:Entity)
+            WHERE all(node IN nodes(path) WHERE node:Entity)
+            UNWIND relationships(path) AS rel
+            WITH rel, min(length(path)) AS distance
+            RETURN startNode(rel).name AS source,
+                   type(rel) AS relation,
+                   endNode(rel).name AS target,
+                   coalesce(rel.evidence, []) AS evidence,
+                   coalesce(rel.documents, []) AS documents,
+                   distance
+            ORDER BY distance, source, relation, target
+            LIMIT $limit
+            """,
+            keys=list(seed_keys),
+            limit=limit,
+        )
+        return [GraphFact.model_validate(row) for row in rows]
+
+    def stats(self) -> dict[str, int]:
+        rows = self._run(
+            """
+            RETURN COUNT { (:Document) } AS documents,
+                   COUNT { (:Chunk) } AS chunks,
+                   COUNT { (:Entity) } AS entities,
+                   COUNT { (:Entity)-[]->(:Entity) } AS relations
+            """
+        )
+        return rows[0]
+
+    def reset(self) -> None:
+        self._run("MATCH (n) DETACH DELETE n")
+
+    def _run(self, query: str, **parameters: Any) -> list[dict[str, Any]]:
+        records, _, _ = self._driver.execute_query(
+            query,
+            parameters_=parameters,
+            database_=self._database,
+        )
+        return [record.data() for record in records]

src/techdocs_graphrag/indexing.py

@@ -0,0 +1,66 @@
+from collections.abc import Callable, Sequence
+from concurrent.futures import ThreadPoolExecutor
+from dataclasses import dataclass
+
+from techdocs_graphrag.extraction import GraphExtractor
+from techdocs_graphrag.ingestion import ParagraphSplitter, SourceDocument, chunk_documents
+from techdocs_graphrag.ports import KnowledgeGraph, LanguageModel
+
+ProgressCallback = Callable[[int, int], None]
+
+
+@dataclass(frozen=True, slots=True)
+class IndexingReport:
+    documents: int
+    chunks: int
+    entities: int
+    relations: int
+
+
+class IndexingPipeline:
+    def __init__(
+        self,
+        *,
+        splitter: ParagraphSplitter,
+        llm: LanguageModel,
+        extractor: GraphExtractor,
+        store: KnowledgeGraph,
+        workers: int,
+    ) -> None:
+        self._splitter = splitter
+        self._llm = llm
+        self._extractor = extractor
+        self._store = store
+        self._workers = workers
+
+    def run(
+        self,
+        documents: Sequence[SourceDocument],
+        on_progress: ProgressCallback | None = None,
+    ) -> IndexingReport:
+        chunks = chunk_documents(documents, self._splitter)
+        if not chunks:
+            return IndexingReport(documents=len(documents), chunks=0, entities=0, relations=0)
+
+        self._store.ensure_schema()
+        self._store.upsert_chunks(chunks, self._llm.embed([chunk.text for chunk in chunks]))
+
+        entities = relations = 0
+        with ThreadPoolExecutor(max_workers=self._workers) as executor:
+            extractions = executor.map(self._extractor.extract, chunks)
+            for completed, (chunk, extraction) in enumerate(
+                zip(chunks, extractions, strict=True), start=1
+            ):
+                self._store.upsert_extraction(chunk, extraction)
+                entities += len(extraction.entities)
+                relations += len(extraction.relations)
+                if on_progress:
+                    on_progress(completed, len(chunks))
+
+        self._store.await_indexes()
+        return IndexingReport(
+            documents=len(documents),
+            chunks=len(chunks),
+            entities=entities,
+            relations=relations,
+        )

src/techdocs_graphrag/ingestion.py

@@ -0,0 +1,77 @@
+import hashlib
+import re
+from collections.abc import Iterable
+from dataclasses import dataclass
+from pathlib import Path
+
+from techdocs_graphrag.models import Chunk
+
+SUPPORTED_SUFFIXES = frozenset({".md", ".markdown", ".txt"})
+PARAGRAPH_SEPARATOR = "\n\n"
+
+
+@dataclass(frozen=True, slots=True)
+class SourceDocument:
+    name: str
+    text: str
+
+
+def load_documents(root: Path) -> list[SourceDocument]:
+    return [
+        SourceDocument(name=path.relative_to(root).as_posix(), text=path.read_text(encoding="utf-8"))
+        for path in sorted(root.rglob("*"))
+        if path.is_file() and path.suffix.lower() in SUPPORTED_SUFFIXES
+    ]
+
+
+class ParagraphSplitter:
+    def __init__(self, chunk_size: int, chunk_overlap: int) -> None:
+        if not 0 <= chunk_overlap < chunk_size:
+            raise ValueError("chunk_overlap must be in [0, chunk_size)")
+        self._chunk_size = chunk_size
+        self._chunk_overlap = chunk_overlap
+
+    def split(self, text: str) -> list[str]:
+        chunks: list[str] = []
+        buffer = ""
+        for paragraph in self._paragraphs(text):
+            candidate = self._join(buffer, paragraph)
+            if len(candidate) <= self._chunk_size:
+                buffer = candidate
+                continue
+            if buffer:
+                chunks.append(buffer)
+                candidate = self._join(self._overlap_of(buffer), paragraph)
+            while len(candidate) > self._chunk_size:
+                chunks.append(candidate[: self._chunk_size])
+                candidate = candidate[self._chunk_size - self._chunk_overlap :]
+            buffer = candidate
+        if buffer:
+            chunks.append(buffer)
+        return chunks
+
+    @staticmethod
+    def _paragraphs(text: str) -> Iterable[str]:
+        return (block.strip() for block in re.split(r"\n\s*\n", text) if block.strip())
+
+    @staticmethod
+    def _join(head: str, tail: str) -> str:
+        return f"{head}{PARAGRAPH_SEPARATOR}{tail}" if head else tail
+
+    def _overlap_of(self, text: str) -> str:
+        if not self._chunk_overlap or len(text) <= self._chunk_overlap:
+            return ""
+        tail = text[-self._chunk_overlap :]
+        return tail.partition(" ")[2] or tail
+
+
+def chunk_id(document: str, position: int, text: str) -> str:
+    return hashlib.sha256(f"{document}\x00{position}\x00{text}".encode()).hexdigest()[:16]
+
+
+def chunk_documents(documents: Iterable[SourceDocument], splitter: ParagraphSplitter) -> list[Chunk]:
+    return [
+        Chunk(id=chunk_id(document.name, position, text), document=document.name, position=position, text=text)
+        for document in documents
+        for position, text in enumerate(splitter.split(document.text))
+    ]

src/techdocs_graphrag/llm.py

@@ -0,0 +1,60 @@
+from collections.abc import Sequence
+from itertools import batched
+from typing import Any
+
+from openai import APIConnectionError, APITimeoutError, InternalServerError, OpenAI, RateLimitError
+from tenacity import retry, retry_if_exception_type, stop_after_attempt, wait_random_exponential
+
+from techdocs_graphrag.config import Settings
+
+EMBEDDING_BATCH_SIZE = 96
+TRANSIENT_ERRORS = (APIConnectionError, APITimeoutError, InternalServerError, RateLimitError)
+
+with_retry = retry(
+    retry=retry_if_exception_type(TRANSIENT_ERRORS),
+    wait=wait_random_exponential(min=1, max=30),
+    stop=stop_after_attempt(5),
+    reraise=True,
+)
+
+
+class OpenAIGateway:
+    def __init__(self, settings: Settings) -> None:
+        self._client = OpenAI(api_key=settings.openai_api_key.get_secret_value())
+        self._chat_model = settings.chat_model
+        self._chat_temperature = settings.chat_temperature
+        self._embedding_model = settings.embedding_model
+        self._embedding_dimensions = settings.embedding_dimensions
+
+    @with_retry
+    def complete(self, system_prompt: str, user_prompt: str, *, json_mode: bool = False) -> str:
+        options: dict[str, Any] = {}
+        if json_mode:
+            options["response_format"] = {"type": "json_object"}
+        if self._chat_temperature is not None:
+            options["temperature"] = self._chat_temperature
+        response = self._client.chat.completions.create(
+            model=self._chat_model,
+            messages=[
+                {"role": "system", "content": system_prompt},
+                {"role": "user", "content": user_prompt},
+            ],
+            **options,
+        )
+        return response.choices[0].message.content or ""
+
+    def embed(self, texts: Sequence[str]) -> list[list[float]]:
+        return [
+            vector
+            for batch in batched(texts, EMBEDDING_BATCH_SIZE)
+            for vector in self._embed_batch(batch)
+        ]
+
+    @with_retry
+    def _embed_batch(self, texts: Sequence[str]) -> list[list[float]]:
+        response = self._client.embeddings.create(
+            model=self._embedding_model,
+            input=list(texts),
+            dimensions=self._embedding_dimensions,
+        )
+        return [item.embedding for item in sorted(response.data, key=lambda item: item.index)]

src/techdocs_graphrag/models.py

@@ -0,0 +1,144 @@
+import re
+import unicodedata
+from enum import StrEnum
+from typing import Annotated
+
+from pydantic import BaseModel, ConfigDict, Field, StringConstraints, field_validator, model_validator
+
+NonEmptyStr = Annotated[str, StringConstraints(strip_whitespace=True, min_length=1)]
+
+RESERVED_RELATIONS = frozenset({"MENTIONS", "HAS_CHUNK"})
+FALLBACK_RELATION = "RELATED_TO"
+
+
+def to_ascii(value: str) -> str:
+    return unicodedata.normalize("NFKD", value).encode("ascii", "ignore").decode()
+
+
+def canonical_key(name: str) -> str:
+    return re.sub(r"[^a-z0-9]", "", to_ascii(name).lower()) or name.strip().casefold()
+
+
+class EntityType(StrEnum):
+    COMPONENT = "Component"
+    CLASS = "Class"
+    FUNCTION = "Function"
+    LIBRARY = "Library"
+    CONCEPT = "Concept"
+    FEATURE = "Feature"
+    ERROR = "Error"
+    DATABASE = "Database"
+    API = "API"
+
+
+ENTITY_TYPES_BY_NAME = {entity_type.value.lower(): entity_type for entity_type in EntityType}
+
+
+class RetrievalMode(StrEnum):
+    VECTOR = "vector"
+    GRAPH = "graphrag"
+
+
+class Chunk(BaseModel):
+    model_config = ConfigDict(frozen=True)
+
+    id: str
+    document: str
+    position: int
+    text: str
+
+
+class RetrievedChunk(Chunk):
+    score: float
+    entity_keys: list[str] = Field(default_factory=list)
+
+
+class Entity(BaseModel):
+    name: NonEmptyStr
+    type: EntityType = EntityType.CONCEPT
+
+    @field_validator("type", mode="before")
+    @classmethod
+    def coerce_type(cls, value: object) -> EntityType:
+        return ENTITY_TYPES_BY_NAME.get(str(value).strip().lower(), EntityType.CONCEPT)
+
+    @property
+    def key(self) -> str:
+        return canonical_key(self.name)
+
+
+class Relation(BaseModel):
+    source: NonEmptyStr
+    relation: str
+    target: NonEmptyStr
+    evidence: str = ""
+
+    @field_validator("relation", mode="before")
+    @classmethod
+    def normalize_relation(cls, value: object) -> str:
+        normalized = re.sub(r"[^A-Z0-9]+", "_", to_ascii(str(value)).upper()).strip("_")
+        if not normalized or not normalized[0].isalpha() or normalized in RESERVED_RELATIONS:
+            return FALLBACK_RELATION
+        return normalized
+
+    @field_validator("evidence", mode="before")
+    @classmethod
+    def normalize_evidence(cls, value: object) -> str:
+        return " ".join(str(value or "").split())
+
+    @property
+    def source_key(self) -> str:
+        return canonical_key(self.source)
+
+    @property
+    def target_key(self) -> str:
+        return canonical_key(self.target)
+
+
+class GraphExtraction(BaseModel):
+    entities: list[Entity] = Field(default_factory=list)
+    relations: list[Relation] = Field(default_factory=list)
+
+    @model_validator(mode="after")
+    def reconcile(self) -> "GraphExtraction":
+        self.relations = [
+            relation for relation in self.relations if relation.source_key != relation.target_key
+        ]
+        entities = {entity.key: entity for entity in self.entities}
+        for relation in self.relations:
+            entities.setdefault(relation.source_key, Entity(name=relation.source))
+            entities.setdefault(relation.target_key, Entity(name=relation.target))
+        self.entities = list(entities.values())
+        return self
+
+
+class GraphFact(BaseModel):
+    source: str
+    relation: str
+    target: str
+    evidence: list[str] = Field(default_factory=list)
+    documents: list[str] = Field(default_factory=list)
+    distance: int = 1
+
+
+class RetrievalContext(BaseModel):
+    chunks: list[RetrievedChunk] = Field(default_factory=list)
+    facts: list[GraphFact] = Field(default_factory=list)
+    seed_entities: list[str] = Field(default_factory=list)
+
+    @property
+    def is_empty(self) -> bool:
+        return not self.chunks and not self.facts
+
+    @property
+    def sources(self) -> list[str]:
+        documents = {chunk.document for chunk in self.chunks}
+        documents.update(document for fact in self.facts for document in fact.documents)
+        return sorted(documents)
+
+
+class Answer(BaseModel):
+    question: str
+    text: str
+    mode: RetrievalMode
+    context: RetrievalContext

src/techdocs_graphrag/ports.py

@@ -0,0 +1,26 @@
+from collections.abc import Sequence
+from typing import Protocol
+
+from techdocs_graphrag.models import Chunk, GraphExtraction, GraphFact, RetrievedChunk
+
+
+class LanguageModel(Protocol):
+    def complete(self, system_prompt: str, user_prompt: str, *, json_mode: bool = False) -> str: ...
+
+    def embed(self, texts: Sequence[str]) -> list[list[float]]: ...
+
+
+class KnowledgeGraph(Protocol):
+    def ensure_schema(self) -> None: ...
+
+    def await_indexes(self) -> None: ...
+
+    def upsert_chunks(self, chunks: Sequence[Chunk], embeddings: Sequence[list[float]]) -> None: ...
+
+    def upsert_extraction(self, chunk: Chunk, extraction: GraphExtraction) -> None: ...
+
+    def vector_search(self, embedding: list[float], top_k: int) -> list[RetrievedChunk]: ...
+
+    def search_entities(self, text: str, limit: int) -> list[str]: ...
+
+    def expand(self, seed_keys: Sequence[str], max_hops: int, limit: int) -> list[GraphFact]: ...

src/techdocs_graphrag/prompts.py

@@ -0,0 +1,78 @@
+from collections.abc import Sequence
+
+from techdocs_graphrag.models import EntityType, GraphFact, RetrievalContext, RetrievedChunk
+
+ENTITY_TYPES = ", ".join(entity_type.value for entity_type in EntityType)
+
+EXTRACTION_SYSTEM_PROMPT = f"""\
+Tu es un système d'extraction de graphe de connaissances pour de la documentation technique.
+
+À partir du texte fourni par l'utilisateur, extrais :
+1. les entités techniques importantes (composants, classes, fonctions, bibliothèques, concepts...) ;
+2. le type de chaque entité, choisi parmi : {ENTITY_TYPES} ;
+3. les relations factuelles explicitement affirmées par le texte ;
+4. une preuve textuelle courte (25 mots maximum) recopiée du texte pour chaque relation.
+
+Règles :
+- Nomme chaque entité sous sa forme canonique, telle qu'écrite dans le code (ex. StateGraph, MemorySaver).
+- Utilise le singulier pour les concepts (Channel et non Channels).
+- Écris les relations en anglais, en UPPER_SNAKE_CASE, orientées source -> cible
+  (ex. USES, PERSISTS_WITH, PART_OF, DEPENDS_ON, CONFIGURES, RETURNS, RAISES, EXTENDS, ENABLES).
+- N'invente aucune relation absente du texte.
+- Si le texte ne contient rien d'exploitable, renvoie des listes vides.
+
+Réponds uniquement avec un objet JSON valide de cette forme :
+{{
+  "entities": [{{"name": "StateGraph", "type": "Component"}}],
+  "relations": [
+    {{
+      "source": "StateGraph",
+      "relation": "USES",
+      "target": "Channel",
+      "evidence": "StateGraph utilise des Channels pour gérer l'état"
+    }}
+  ]
+}}
+"""
+
+ANSWER_SYSTEM_PROMPT = """\
+Tu es un assistant expert en documentation technique.
+
+Règles :
+- Réponds uniquement à partir du contexte fourni : faits du graphe et extraits documentaires.
+- Réponds dans la langue de la question.
+- Explique clairement comment les composants sont reliés entre eux lorsque c'est utile.
+- Cite tes sources en ligne sous la forme [nom_du_document].
+- Ajoute un court exemple de code uniquement s'il est présent ou directement déductible du contexte.
+- Si le contexte ne suffit pas, dis-le explicitement au lieu d'inventer.
+"""
+
+NO_FACTS = "Aucun fait pertinent."
+NO_CHUNKS = "Aucun extrait pertinent."
+
+
+def render_answer_prompt(question: str, context: RetrievalContext) -> str:
+    return "\n\n".join(
+        (
+            f"Question :\n{question}",
+            f"Faits du graphe de connaissances :\n{render_facts(context.facts)}",
+            f"Extraits documentaires :\n{render_chunks(context.chunks)}",
+        )
+    )
+
+
+def render_facts(facts: Sequence[GraphFact]) -> str:
+    if not facts:
+        return NO_FACTS
+    return "\n".join(
+        f"- {fact.source} --{fact.relation}--> {fact.target}"
+        f" | preuve : {' / '.join(fact.evidence[:2]) or 'n/a'}"
+        f" | sources : {', '.join(fact.documents) or 'n/a'}"
+        for fact in facts
+    )
+
+
+def render_chunks(chunks: Sequence[RetrievedChunk]) -> str:
+    if not chunks:
+        return NO_CHUNKS
+    return "\n\n".join(f"[{chunk.document}]\n{chunk.text}" for chunk in chunks)

src/techdocs_graphrag/retrieval.py

@@ -0,0 +1,41 @@
+from collections import Counter
+from collections.abc import Sequence
+
+from techdocs_graphrag.models import RetrievalContext, RetrievalMode, RetrievedChunk
+from techdocs_graphrag.ports import KnowledgeGraph, LanguageModel
+
+MAX_LEXICAL_SEEDS = 5
+MAX_MENTION_SEEDS = 8
+
+
+class HybridRetriever:
+    def __init__(
+        self,
+        *,
+        llm: LanguageModel,
+        store: KnowledgeGraph,
+        top_k: int,
+        max_hops: int,
+        max_facts: int,
+    ) -> None:
+        self._llm = llm
+        self._store = store
+        self._top_k = top_k
+        self._max_hops = max_hops
+        self._max_facts = max_facts
+
+    def retrieve(self, question: str, mode: RetrievalMode = RetrievalMode.GRAPH) -> RetrievalContext:
+        embedding = self._llm.embed([question])[0]
+        chunks = self._store.vector_search(embedding, self._top_k)
+        if mode is RetrievalMode.VECTOR:
+            return RetrievalContext(chunks=chunks)
+
+        seeds = self._seed_entities(question, chunks)
+        facts = self._store.expand(seeds, self._max_hops, self._max_facts)
+        return RetrievalContext(chunks=chunks, facts=facts, seed_entities=seeds)
+
+    def _seed_entities(self, question: str, chunks: Sequence[RetrievedChunk]) -> list[str]:
+        lexical_matches = self._store.search_entities(question, MAX_LEXICAL_SEEDS)
+        mentions = Counter(key for chunk in chunks for key in chunk.entity_keys)
+        most_mentioned = [key for key, _ in mentions.most_common(MAX_MENTION_SEEDS)]
+        return list(dict.fromkeys([*lexical_matches, *most_mentioned]))

src/techdocs_graphrag/services.py

@@ -0,0 +1,47 @@
+from dataclasses import dataclass
+
+from techdocs_graphrag.answering import GraphRAGAssistant
+from techdocs_graphrag.config import Settings, get_settings
+from techdocs_graphrag.extraction import GraphExtractor
+from techdocs_graphrag.graph_store import Neo4jGraphStore
+from techdocs_graphrag.indexing import IndexingPipeline
+from techdocs_graphrag.ingestion import ParagraphSplitter
+from techdocs_graphrag.llm import OpenAIGateway
+from techdocs_graphrag.retrieval import HybridRetriever
+
+
+@dataclass(frozen=True, slots=True)
+class Services:
+    settings: Settings
+    store: Neo4jGraphStore
+    assistant: GraphRAGAssistant
+    indexer: IndexingPipeline
+
+    def close(self) -> None:
+        self.store.close()
+
+
+def build_services(settings: Settings | None = None) -> Services:
+    settings = settings or get_settings()
+    llm = OpenAIGateway(settings)
+    store = Neo4jGraphStore(settings)
+    retriever = HybridRetriever(
+        llm=llm,
+        store=store,
+        top_k=settings.vector_top_k,
+        max_hops=settings.graph_max_hops,
+        max_facts=settings.graph_max_facts,
+    )
+    indexer = IndexingPipeline(
+        splitter=ParagraphSplitter(settings.chunk_size, settings.chunk_overlap),
+        llm=llm,
+        extractor=GraphExtractor(llm),
+        store=store,
+        workers=settings.extraction_workers,
+    )
+    return Services(
+        settings=settings,
+        store=store,
+        assistant=GraphRAGAssistant(retriever=retriever, llm=llm),
+        indexer=indexer,
+    )

src/techdocs_graphrag/visualization.py

@@ -0,0 +1,35 @@
+from collections.abc import Iterable, Sequence
+
+from techdocs_graphrag.models import GraphFact, canonical_key
+
+SEED_COLOR = "#C7D2FE"
+NODE_COLOR = "#F1F5F9"
+
+
+def facts_to_dot(facts: Sequence[GraphFact], highlighted_keys: Iterable[str] = ()) -> str:
+    highlighted = set(highlighted_keys)
+    names = sorted({name for fact in facts for name in (fact.source, fact.target)})
+    nodes = [
+        f"  {quote(name)} [fillcolor={quote(SEED_COLOR if canonical_key(name) in highlighted else NODE_COLOR)}];"
+        for name in names
+    ]
+    edges = [
+        f"  {quote(fact.source)} -> {quote(fact.target)} [label={quote(fact.relation)}];"
+        for fact in facts
+    ]
+    return "\n".join(
+        (
+            "digraph KnowledgeGraph {",
+            "  rankdir=LR;",
+            '  node [shape=box, style="rounded,filled", fontname="Helvetica", fontcolor="#0F172A"];',
+            '  edge [fontname="Helvetica", fontsize=10, color="#64748B", fontcolor="#64748B"];',
+            *nodes,
+            *edges,
+            "}",
+        )
+    )
+
+
+def quote(value: str) -> str:
+    escaped = value.replace("\\", "\\\\").replace('"', '\\"').replace("\n", " ")
+    return f'"{escaped}"'

tests/__init__.py

@@ -0,0 +1 @@
+ 

tests/conftest.py

@@ -0,0 +1,71 @@
+from collections.abc import Sequence
+
+import pytest
+
+from techdocs_graphrag.models import Chunk, GraphExtraction, GraphFact, RetrievedChunk
+
+
+class FakeLanguageModel:
+    def __init__(self, completion: str = "réponse") -> None:
+        self.completion = completion
+        self.completions: list[tuple[str, str]] = []
+
+    def complete(self, system_prompt: str, user_prompt: str, *, json_mode: bool = False) -> str:
+        self.completions.append((system_prompt, user_prompt))
+        return self.completion
+
+    def embed(self, texts: Sequence[str]) -> list[list[float]]:
+        return [[float(len(text))] for text in texts]
+
+
+class FakeKnowledgeGraph:
+    def __init__(
+        self,
+        chunks: list[RetrievedChunk] | None = None,
+        lexical_matches: list[str] | None = None,
+        facts: list[GraphFact] | None = None,
+    ) -> None:
+        self.chunks = chunks or []
+        self.lexical_matches = lexical_matches or []
+        self.facts = facts or []
+        self.expanded_with: list[str] | None = None
+        self.extractions: list[tuple[Chunk, GraphExtraction]] = []
+        self.indexes_awaited = False
+
+    def ensure_schema(self) -> None:
+        pass
+
+    def await_indexes(self) -> None:
+        self.indexes_awaited = True
+
+    def upsert_chunks(self, chunks: Sequence[Chunk], embeddings: Sequence[list[float]]) -> None:
+        pass
+
+    def upsert_extraction(self, chunk: Chunk, extraction: GraphExtraction) -> None:
+        self.extractions.append((chunk, extraction))
+
+    def vector_search(self, embedding: list[float], top_k: int) -> list[RetrievedChunk]:
+        return self.chunks[:top_k]
+
+    def search_entities(self, text: str, limit: int) -> list[str]:
+        return self.lexical_matches[:limit]
+
+    def expand(self, seed_keys: Sequence[str], max_hops: int, limit: int) -> list[GraphFact]:
+        self.expanded_with = list(seed_keys)
+        return self.facts[:limit]
+
+
+def retrieved_chunk(document: str, entity_keys: list[str]) -> RetrievedChunk:
+    return RetrievedChunk(
+        id=f"{document}-0",
+        document=document,
+        position=0,
+        text=f"contenu de {document}",
+        score=0.9,
+        entity_keys=entity_keys,
+    )
+
+
+@pytest.fixture
+def fake_llm() -> FakeLanguageModel:
+    return FakeLanguageModel()

tests/test_ingestion.py

@@ -0,0 +1,51 @@
+from pathlib import Path
+
+import pytest
+
+from techdocs_graphrag.ingestion import (
+    ParagraphSplitter,
+    SourceDocument,
+    chunk_documents,
+    load_documents,
+)
+
+
+def test_splitter_respects_chunk_size_and_covers_every_paragraph() -> None:
+    paragraphs = [f"P{index:02d} " + "mot " * 40 for index in range(10)]
+    chunks = ParagraphSplitter(chunk_size=120, chunk_overlap=20).split("\n\n".join(paragraphs))
+    joined = " ".join(chunks)
+
+    assert all(len(chunk) <= 120 for chunk in chunks)
+    assert all(f"P{index:02d}" in joined for index in range(10))
+
+
+def test_splitter_keeps_short_paragraphs_together() -> None:
+    chunks = ParagraphSplitter(chunk_size=100, chunk_overlap=10).split("Un.\n\nDeux.\n\nTrois.")
+    assert chunks == ["Un.\n\nDeux.\n\nTrois."]
+
+
+def test_splitter_rejects_overlap_larger_than_chunk() -> None:
+    with pytest.raises(ValueError):
+        ParagraphSplitter(chunk_size=10, chunk_overlap=10)
+
+
+def test_load_documents_reads_supported_files_recursively(tmp_path: Path) -> None:
+    (tmp_path / "guides").mkdir()
+    (tmp_path / "intro.md").write_text("Intro", encoding="utf-8")
+    (tmp_path / "guides" / "setup.md").write_text("Setup", encoding="utf-8")
+    (tmp_path / "image.png").write_bytes(b"\x89PNG")
+
+    documents = load_documents(tmp_path)
+
+    assert [document.name for document in documents] == ["guides/setup.md", "intro.md"]
+
+
+def test_chunk_ids_are_deterministic_and_unique() -> None:
+    documents = [SourceDocument("a.md", "Alpha.\n\nBeta."), SourceDocument("b.md", "Alpha.")]
+    splitter = ParagraphSplitter(chunk_size=8, chunk_overlap=0)
+
+    first_run = chunk_documents(documents, splitter)
+    second_run = chunk_documents(documents, splitter)
+
+    assert [chunk.id for chunk in first_run] == [chunk.id for chunk in second_run]
+    assert len({chunk.id for chunk in first_run}) == len(first_run)

tests/test_models.py

@@ -0,0 +1,60 @@
+import pytest
+
+from techdocs_graphrag.models import (
+    FALLBACK_RELATION,
+    Entity,
+    EntityType,
+    GraphExtraction,
+    GraphFact,
+    Relation,
+    RetrievalContext,
+    canonical_key,
+)
+
+
+@pytest.mark.parametrize("name", ["StateGraph", "State Graph", "state_graph", " stategraph "])
+def test_canonical_key_merges_spelling_variants(name: str) -> None:
+    assert canonical_key(name) == "stategraph"
+
+
+@pytest.mark.parametrize(
+    ("raw", "expected"),
+    [
+        ("persists with", "PERSISTS_WITH"),
+        ("dépend-de", "DEPEND_DE"),
+        ("MENTIONS", FALLBACK_RELATION),
+        ("42_uses", FALLBACK_RELATION),
+        ("", FALLBACK_RELATION),
+    ],
+)
+def test_relation_type_is_normalized_to_a_safe_identifier(raw: str, expected: str) -> None:
+    relation = Relation(source="A", relation=raw, target="B")
+    assert relation.relation == expected
+
+
+def test_unknown_entity_type_falls_back_to_concept() -> None:
+    assert Entity(name="Reducer", type="Pattern").type is EntityType.CONCEPT
+    assert Entity(name="StateGraph", type="component").type is EntityType.COMPONENT
+
+
+def test_extraction_adds_missing_endpoints_and_drops_self_loops() -> None:
+    extraction = GraphExtraction.model_validate(
+        {
+            "entities": [{"name": "StateGraph", "type": "Component"}],
+            "relations": [
+                {"source": "StateGraph", "relation": "USES", "target": "Channel"},
+                {"source": "StateGraph", "relation": "IS", "target": "State Graph"},
+            ],
+        }
+    )
+
+    assert {entity.key for entity in extraction.entities} == {"stategraph", "channel"}
+    assert [relation.relation for relation in extraction.relations] == ["USES"]
+
+
+def test_context_sources_combine_chunks_and_facts() -> None:
+    context = RetrievalContext(
+        facts=[GraphFact(source="A", relation="USES", target="B", documents=["b.md", "a.md"])]
+    )
+    assert context.sources == ["a.md", "b.md"]
+    assert not context.is_empty

tests/test_pipeline.py

@@ -0,0 +1,101 @@
+from techdocs_graphrag.answering import NO_CONTEXT_ANSWER, GraphRAGAssistant
+from techdocs_graphrag.extraction import GraphExtractor
+from techdocs_graphrag.indexing import IndexingPipeline
+from techdocs_graphrag.ingestion import ParagraphSplitter, SourceDocument
+from techdocs_graphrag.models import Chunk, GraphFact, RetrievalMode
+from techdocs_graphrag.retrieval import HybridRetriever
+from techdocs_graphrag.visualization import facts_to_dot
+
+from .conftest import FakeKnowledgeGraph, FakeLanguageModel, retrieved_chunk
+
+EXTRACTION_PAYLOAD = """
+{
+  "entities": [{"name": "StateGraph", "type": "Component"}],
+  "relations": [{"source": "StateGraph", "relation": "uses", "target": "Channel", "evidence": "x"}]
+}
+"""
+
+
+def build_retriever(llm: FakeLanguageModel, store: FakeKnowledgeGraph) -> HybridRetriever:
+    return HybridRetriever(llm=llm, store=store, top_k=5, max_hops=2, max_facts=10)
+
+
+def test_graph_mode_seeds_expansion_with_lexical_then_mentioned_entities(
+    fake_llm: FakeLanguageModel,
+) -> None:
+    store = FakeKnowledgeGraph(
+        chunks=[
+            retrieved_chunk("a.md", ["checkpointer", "stategraph"]),
+            retrieved_chunk("b.md", ["checkpointer"]),
+        ],
+        lexical_matches=["stategraph"],
+    )
+
+    context = build_retriever(fake_llm, store).retrieve("question", RetrievalMode.GRAPH)
+
+    assert store.expanded_with == ["stategraph", "checkpointer"]
+    assert context.seed_entities == ["stategraph", "checkpointer"]
+
+
+def test_vector_mode_never_touches_the_graph(fake_llm: FakeLanguageModel) -> None:
+    store = FakeKnowledgeGraph(chunks=[retrieved_chunk("a.md", ["stategraph"])])
+
+    context = build_retriever(fake_llm, store).retrieve("question", RetrievalMode.VECTOR)
+
+    assert store.expanded_with is None
+    assert context.facts == []
+
+
+def test_assistant_does_not_call_the_llm_without_context(fake_llm: FakeLanguageModel) -> None:
+    assistant = GraphRAGAssistant(retriever=build_retriever(fake_llm, FakeKnowledgeGraph()), llm=fake_llm)
+
+    answer = assistant.ask("question")
+
+    assert answer.text == NO_CONTEXT_ANSWER
+    assert fake_llm.completions == []
+
+
+def test_assistant_sends_graph_facts_to_the_llm(fake_llm: FakeLanguageModel) -> None:
+    store = FakeKnowledgeGraph(
+        chunks=[retrieved_chunk("a.md", ["stategraph"])],
+        facts=[GraphFact(source="StateGraph", relation="USES", target="Channel", documents=["a.md"])],
+    )
+    assistant = GraphRAGAssistant(retriever=build_retriever(fake_llm, store), llm=fake_llm)
+
+    answer = assistant.ask("Comment fonctionne StateGraph ?")
+
+    _, user_prompt = fake_llm.completions[0]
+    assert "StateGraph --USES--> Channel" in user_prompt
+    assert answer.context.sources == ["a.md"]
+
+
+def test_extractor_returns_empty_graph_on_invalid_json() -> None:
+    chunk = Chunk(id="1", document="a.md", position=0, text="texte")
+    assert GraphExtractor(FakeLanguageModel("pas du json")).extract(chunk).entities == []
+
+
+def test_indexing_pipeline_reports_extracted_graph() -> None:
+    llm = FakeLanguageModel(EXTRACTION_PAYLOAD)
+    store = FakeKnowledgeGraph()
+    pipeline = IndexingPipeline(
+        splitter=ParagraphSplitter(chunk_size=50, chunk_overlap=0),
+        llm=llm,
+        extractor=GraphExtractor(llm),
+        store=store,
+        workers=2,
+    )
+    progress: list[int] = []
+
+    report = pipeline.run(
+        [SourceDocument("a.md", "Premier paragraphe.\n\nSecond paragraphe assez long pour couper.")],
+        on_progress=lambda done, _: progress.append(done),
+    )
+
+    assert (report.chunks, report.entities, report.relations) == (2, 4, 2)
+    assert progress == [1, 2]
+    assert store.indexes_awaited
+
+
+def test_dot_output_escapes_quotes() -> None:
+    dot = facts_to_dot([GraphFact(source='Say "hi"', relation="USES", target="B")])
+    assert '"Say \\"hi\\""' in dot