📚 Learning Hub

English

Français

English

Français

Appearance

GraphRAG — quand les relations tuent le vector RAG ​

TL;DR Vector RAG est un marteau qui voit tout en clous. Dès que ta question implique des relations transitives ("qui contrôle qui ?", "quels composants partagent un fournisseur risqué ?", "quels dirigeants sont liés à plus de 3 liquidations ?"), le cosine sim s'écroule. GraphRAG transforme tes docs en knowledge graph (entités + relations), puis interroge en mélangeant traversal Cypher + retrieval vector. Microsoft GraphRAG, LangChain GraphCypherQAChain, LlamaIndex PropertyGraphIndex, Neo4j sont les briques. Sur une mission BODACC + Infogreffe (registres légaux français) pour un cabinet de due diligence, GraphRAG répond à des questions impossibles en vector pur — et se facture 18-30 k€ par mission spécialisée.

🧠 Mental model ​

Vector RAG = recherche par proximité sémantique. GraphRAG = recherche par relations explicites.

   VECTOR RAG                          GRAPH RAG
   ──────────                          ─────────
   doc ────► embed ────► top-k         doc ──► entities ──► graph
                                                            │
   query ──► embed ──► cosine          query ──► Cypher ────┤
                                                            │
   ◀──── flat chunks                   ◀── subgraph + ctx ──┘

   forte sur "ressemblance"            forte sur "qui est lié à quoi"

Analogie : tu enquêtes sur un scandale financier. Vector RAG = tu cherches par mots-clés dans des articles de presse. Graph RAG = tu as la généalogie des sociétés écrans, tu remontes "qui détient qui, qui dirige quoi". Pour des questions du type "trouve-moi tous les dirigeants liés à 3+ liquidations dans le BTP en 2024", le graphe est invincible.

Pour ton cerveau de dev SQL : graph DB = JOIN ... ON ... JOIN ... ON ... à profondeur variable, optimisé pour la traversée. Cypher = SQL pour graphes. Vector RAG = SELECT ... ORDER BY embedding <-> $1. GraphRAG = les deux ensemble : Cypher pour la structure, vector pour le contenu textuel.

🛠️ Code minimal ​

Le "Hello World" GraphRAG en LangChain + Neo4j, en 25 lignes :

python
from langchain_community.graphs import Neo4jGraph
from langchain.chains import GraphCypherQAChain
from langchain_anthropic import ChatAnthropic

graph = Neo4jGraph(
    url="bolt://localhost:7687",
    username="neo4j",
    password=os.environ["NEO4J_PASSWORD"],
)

# Seed quelques nœuds
graph.query("""
MERGE (acme:Company {name: 'Acme SAS', siren: '123456789'})
MERGE (john:Person {name: 'Jean Dupont'})
MERGE (john)-[:DIRIGE {depuis: '2022-01-01'}]->(acme)
MERGE (acme)-[:LIQUIDEE {date: '2024-03-15'}]->(:Status {label: 'liquidation_judiciaire'})
""")

chain = GraphCypherQAChain.from_llm(
    llm=ChatAnthropic(model="claude-sonnet-4-6", max_tokens=1024),
    graph=graph,
    verbose=True,
    allow_dangerous_requests=True,  # set False en prod, voir pitfalls
)

print(chain.invoke({"query": "Qui a dirigé Acme SAS et quand a-t-elle été liquidée ?"}))

Sous le capot : le LLM génère du Cypher, on l'exécute, on renvoie le résultat au LLM qui formule la réponse en FR.

🎬 Cas d'usage concrets ​

Scénario 1 — Cabinet d'avocats parisien, due diligence M&A ​

Qui : cabinet 28 avocats, pratique M&A, mission de due diligence sur cibles d'acquisition.

Problème : pour qualifier une cible, il faut tracer les relations entre la société, ses dirigeants, leurs autres mandats, les contentieux liés, les actionnaires, les filiales. Vector RAG sur les rapports annuels donne du contenu, mais ne répond pas à "le directeur financier est-il actionnaire d'un fournisseur stratégique ?".

Solution : extraction d'entités via Claude → graphe Neo4j → questions multi-hop.

python
EXTRACT_PROMPT = """Extrait du texte les entités et relations au format JSON.
Entités : Company, Person, Court, Contract, Litigation.
Relations : DIRIGE, ACTIONNAIRE_DE, FILIALE_DE, PARTIE_A, EMPLOYE_DE.

Texte :
{text}

JSON :"""

async def extract_and_graph(text: str, neo: Neo4jGraph):
    raw = await claude_json(EXTRACT_PROMPT.format(text=text))
    for e in raw["entities"]:
        neo.query(
            f"MERGE (n:{e['type']} {{name: $name}}) SET n += $props",
            params={"name": e["name"], "props": e.get("attrs", {})},
        )
    for r in raw["relations"]:
        neo.query(
            f"""MATCH (a {{name: $a}}), (b {{name: $b}})
                MERGE (a)-[rel:{r['type']}]->(b) SET rel += $props""",
            params={"a": r["from"], "b": r["to"], "props": r.get("attrs", {})},
        )

Gains : sur une due diligence type (4 semaines à 6 avocats), l'agent GraphRAG réduit le temps de cartographie à 3-4 jours. Économie estimée par le cabinet : 42 k€ par mission. Le freelance facture la mise en place : 22 k€ (16 jours × 1 400 €/j) + retainer mensuel 3 k€.

Scénario 2 — Industriel aéronautique, supply chain risk ​

Qui : ETI aéro à Toulouse (920 employés, 180 M€ CA), 1 200 composants critiques, 340 fournisseurs.

Problème : "quels avions livrables Q3 dépendent d'un composant dont le fournisseur a un risque crédit > 80 ?" — impossible en vector RAG. Les données existent (ERP, fiches fournisseurs, Coface), mais la relation est implicite.

Solution : graphe Avion ─[CONTIENT]→ Composant ─[FOURNI_PAR]→ Fournisseur ─[A_SCORE]→ RiskScore. Cypher direct :

cypher
MATCH (a:Avion {livraison_q: '2026-Q3'})
      -[:CONTIENT]->(c:Composant)
      -[:FOURNI_PAR]->(f:Fournisseur)
      -[:A_SCORE]->(r:RiskScore)
WHERE r.coface > 80
RETURN a.serial, c.ref, f.name, r.coface
ORDER BY r.coface DESC;

Couplé à un LLM front-end pour formuler la question en langage naturel : le directeur achats tape "donne-moi les expos risque Q3", l'agent traduit en Cypher.

Gains : passage d'un audit risk mensuel manuel (3 jours) à un dashboard temps réel. Évitement d'un retard de livraison identifié grâce au système : 1,8 M€ de pénalités évitées sur un cas réel. Mission freelance : 24 jours × 1 500 €/j = 36 k€ + retainer mensuel 4 k€.

Scénario 3 — Plateforme immobilière, biens-propriétaires-mandats ​

Qui : startup PropTech (45 personnes), agrégation de mandats immo (vente, location), 280 K biens, 90 K propriétaires, 12 K agents.

Problème : détecter les fraudes mandataires (un même agent qui ré-utilise le mandat d'un propriétaire pour 3 biens "fantômes"), et répondre à "quels propriétaires ont eu des mandats expirés non renouvelés mais des annonces actives ?". Vector RAG sur descriptions de biens : nul.

Solution : graphe Bien ─[MANDAT_DE]→ Proprio, Bien ─[GERE_PAR]→ Agent, Mandat ─[STATUT]→ {actif, expiré}. Détection par patterns Cypher :

cypher
// Suspect : annonce active mais mandat expiré depuis > 30 j
MATCH (b:Bien {statut_annonce: 'active'})
      -[:SOUS_MANDAT]->(m:Mandat)
WHERE m.statut = 'expire' AND m.fin < date() - duration({days: 30})
WITH b, m
MATCH (b)-[:GERE_PAR]->(a:Agent)
RETURN a.id, count(b) AS nb_biens_suspects
ORDER BY nb_biens_suspects DESC LIMIT 20;

Gains : détection de 240 annonces frauduleuses en 2 semaines, économies réputationnelles + 80 k€ de litiges évités. Le freelance facture 14 j × 1 250 €/j = 17,5 k€.

🛠️ Exemple end-to-end ​

Use case : GraphRAG sur registres légaux français (BODACC + Infogreffe) pour un cabinet de due diligence. Question cible : "qui sont les dirigeants liés à plus de 3 liquidations judiciaires dans le BTP entre 2022 et 2024 ?".

Tu factures la mission 28 k€ HT (20 jours × 1 400 €/j).

python
# graphrag_bodacc.py
import os
import json
import asyncio
from pathlib import Path
from typing import Any

import anthropic
import httpx
from neo4j import AsyncGraphDatabase

NEO_URI = os.environ["NEO4J_URI"]
NEO_USER = os.environ["NEO4J_USER"]
NEO_PASS = os.environ["NEO4J_PASS"]
CLAUDE = anthropic.AsyncAnthropic(max_retries=4)  # SDK retries 429/5xx/overloaded
CHAT = "claude-sonnet-4-6"      # extraction: cheap + fast, schéma strict
JUDGE = "claude-haiku-4-5"      # entity resolution / disambiguation pairwise
SYNTH = "claude-opus-4-8"       # synthèse multi-hop finale si raisonnement lourd

# Prix /Mtok (input/output) 2026 : sonnet 4.6 = 3/15, haiku 4.5 = 1/5, opus 4.8 = 5/25
PRICE = {CHAT: (3, 15), JUDGE: (1, 5), SYNTH: (5, 25)}
_cost_total = 0.0

def log_usage(stage: str, usage, model: str = CHAT) -> None:
    """Trace le coût réel par étape. Sans ça tu chiffres une mission à l'aveugle."""
    global _cost_total
    pin, pout = PRICE[model]
    # cache_read facturé ~0.1x, cache_write ~1.25x du prix input. Sur l'agent,
    # le préfixe (system + tools) est lu depuis le cache à chaque question :
    # l'ignorer SURESTIME le coût de ~30-40 % et te fait sur-chiffrer la mission.
    cached = getattr(usage, "cache_read_input_tokens", 0) or 0
    written = getattr(usage, "cache_creation_input_tokens", 0) or 0
    cost = (
        usage.input_tokens * pin          # tokens plein tarif
        + cached * pin * 0.1              # lecture cache
        + written * pin * 1.25            # écriture cache
        + usage.output_tokens * pout
    ) / 1_000_000
    _cost_total += cost
    # en prod : pousser vers OTel / Prometheus avec un label {stage, model}

# ------------ Extraction d'entités depuis BODACC ------------

EXTRACT_SYS = """Tu extrais des entités structurées d'annonces légales BODACC.
Réponds en JSON strict :
{
  "company": {"siren": "...", "name": "...", "naf": "...", "city": "..."},
  "directors": [{"name": "...", "role": "..."}],
  "event": {"type": "liquidation_judiciaire|redressement|cession|...", "date": "YYYY-MM-DD"}
}
Si non-pertinent : {"company": null, "directors": [], "event": null}."""

# Schéma Pydantic — la source de vérité du graphe. Si le LLM invente un label
# ou oublie le SIREN, la validation échoue ici, AVANT l'ingest Neo4j.
from pydantic import BaseModel

class Company(BaseModel):
    siren: str | None = None
    name: str | None = None
    naf: str | None = None
    city: str | None = None

class Director(BaseModel):
    name: str
    role: str | None = None

class Event(BaseModel):
    type: str          # liquidation_judiciaire | redressement | cession | ...
    date: str          # YYYY-MM-DD

class BodaccRecord(BaseModel):
    company: Company | None = None
    directors: list[Director] = []
    event: Event | None = None

BODACC_FORMAT = {"type": "json_schema", "schema": BodaccRecord.model_json_schema()}

async def extract_bodacc(annonce: dict) -> dict:
    # Structured output natif : on contraint la sortie par un schéma plutôt
    # que de parser du JSON à la main (find('{')...rfind('}') casse dès qu'une
    # accolade apparaît dans une raison sociale). messages.parse() valide et
    # désérialise contre BodaccRecord (Pydantic) ; le SDK retire les
    # contraintes non supportées (minLength, pattern, etc.) et les revalide
    # côté client. Pas de prefill ni de prompt XML maison : sur opus/sonnet 4.x
    # le prefill assistant renvoie un HTTP 400, structured outputs est la voie.
    msg = await CLAUDE.messages.parse(
        model=CHAT,
        max_tokens=600,
        timeout=30,                 # per-call : une annonce ne doit pas bloquer le batch
        system=EXTRACT_SYS,
        messages=[{"role": "user", "content": json.dumps(annonce, ensure_ascii=False)}],
        output_config={"format": BODACC_FORMAT},  # voir BodaccRecord plus bas
    )
    # log usage → coût réel par doc, indispensable pour chiffrer la mission
    log_usage("extract", msg.usage)
    return msg.parsed_output.model_dump() if msg.parsed_output else {
        "company": None, "directors": [], "event": None
    }

# ------------ Ingest dans Neo4j ------------

CYPHER_INGEST = """
MERGE (c:Company {siren: $siren})
ON CREATE SET c.name = $name, c.naf = $naf, c.city = $city
WITH c
UNWIND $directors AS d
MERGE (p:Person {name: d.name})
MERGE (p)-[r:DIRIGE]->(c)
SET r.role = d.role
WITH c
MERGE (e:Event {company_siren: $siren, type: $event_type, date: $event_date})
MERGE (c)-[:HAS_EVENT]->(e)
"""

async def ingest(driver, data: dict):
    if not data.get("company") or not data.get("event"):
        return
    async with driver.session() as s:
        await s.run(CYPHER_INGEST, {
            "siren": data["company"]["siren"],
            "name": data["company"]["name"],
            "naf": data["company"].get("naf", ""),
            "city": data["company"].get("city", ""),
            "directors": data.get("directors", []),
            "event_type": data["event"]["type"],
            "event_date": data["event"]["date"],
        })

# ------------ Pipeline BODACC ingest ------------

BODACC_API = "https://bodacc-datadila.opendatasoft.com/api/explore/v2.1/catalog/datasets/annonces-commerciales/records"

async def fetch_bodacc(year: int, naf_prefix: str = "41", limit: int = 1000) -> list[dict]:
    params = {
        "where": f"familleavis = 'collective' AND dateparution >= '{year}-01-01' AND dateparution < '{year+1}-01-01'",
        "limit": limit,
    }
    async with httpx.AsyncClient(timeout=60) as cli:
        r = await cli.get(BODACC_API, params=params)
        r.raise_for_status()
        records = r.json().get("results", [])
        # filter BTP (NAF code 41-43)
        return [rec for rec in records if str(rec.get("naf_code", "")).startswith(naf_prefix)]

async def run_ingest(years: list[int], naf: str = "41"):
    driver = AsyncGraphDatabase.driver(NEO_URI, auth=(NEO_USER, NEO_PASS))
    sem = asyncio.Semaphore(8)
    async def handle(rec):
        async with sem:
            data = await extract_bodacc(rec)
            await ingest(driver, data)
    for y in years:
        recs = await fetch_bodacc(y, naf_prefix=naf)
        print(f"{y} : {len(recs)} annonces")
        await asyncio.gather(*[handle(r) for r in recs])
    await driver.close()

# ------------ Cypher tools exposés au LLM ------------

CYPHER_TOOLS = [
    {
        "name": "run_cypher",
        "description": "Exécute une requête Cypher en lecture seule sur le graphe BODACC.",
        "input_schema": {
            "type": "object",
            "properties": {
                "query": {"type": "string", "description": "Cypher READ-only (MATCH/RETURN)"},
            },
            "required": ["query"],
        },
    },
]

FORBIDDEN = ["CREATE", "MERGE", "DELETE", "SET ", "REMOVE", "DROP"]

async def safe_cypher(driver, query: str) -> list[dict]:
    upper = query.upper()
    if any(kw in upper for kw in FORBIDDEN):
        return [{"error": "write operation forbidden"}]
    async with driver.session() as s:
        result = await s.run(query)
        return [dict(r) async for r in result]

# ------------ Agent question → Cypher → réponse ------------

AGENT_SYS = """Tu réponds à des questions sur des registres légaux français (BODACC, BTP).
Schéma du graphe :
- (Company {siren, name, naf, city})
- (Person {name})
- (p:Person)-[:DIRIGE {role}]->(c:Company)
- (c:Company)-[:HAS_EVENT]->(e:Event {type, date})

Utilise run_cypher pour interroger. Vérifie tes résultats avant de répondre.
Donne TOUJOURS le SIREN et la date dans tes réponses."""

async def ask(driver, question: str) -> str:
    history = [{"role": "user", "content": question}]
    for _ in range(5):  # max 5 tool calls — borne dure anti-boucle
        try:
            resp = await CLAUDE.messages.create(
                model=CHAT, max_tokens=1500,
                # adaptive thinking : la génération Cypher multi-hop est un vrai
                # raisonnement (quel pattern de traversée, quel filtre, quelle
                # agrégation). budget_tokens est RETIRÉ sur 4.7/4.8 → adaptive +
                # effort. effort=low ici car Cypher est borné ; passe à medium/high
                # quand l'agent doit chaîner plusieurs requêtes pour répondre.
                thinking={"type": "adaptive"},
                output_config={"effort": "low"},
                # cache_control sur le préfixe stable (system + tools) : le schéma
                # du graphe ne change pas entre questions → ~0.1x sur ce préfixe.
                system=[{"type": "text", "text": AGENT_SYS,
                         "cache_control": {"type": "ephemeral"}}],
                tools=CYPHER_TOOLS, messages=history,
            )
        except (anthropic.RateLimitError, anthropic.OverloadedError):
            await asyncio.sleep(2)   # le SDK retry déjà (max_retries=4) ; filet final
            continue
        except (anthropic.APIStatusError, anthropic.APITimeoutError) as e:
            return f"(erreur API : {e})"

        log_usage("agent", resp.usage)
        if resp.stop_reason == "refusal":          # safety classifier
            return "(requête refusée)"
        if resp.stop_reason == "end_turn":
            return next((b.text for b in resp.content if b.type == "text"), "")

        # exécution PARALLÈLE des tool calls indépendants
        tool_uses = [b for b in resp.content if b.type == "tool_use"]
        history.append({"role": "assistant", "content": resp.content})
        rows_list = await asyncio.gather(
            *[safe_cypher(driver, tu.input["query"]) for tu in tool_uses]
        )
        history.append({"role": "user", "content": [
            {"type": "tool_result", "tool_use_id": tu.id,
             "content": json.dumps(rows, ensure_ascii=False, default=str)[:8000]}
            for tu, rows in zip(tool_uses, rows_list)
        ]})
    return "(too many tool calls)"

# ------------ Démo ------------

async def main():
    await run_ingest(years=[2022, 2023, 2024], naf="41")
    driver = AsyncGraphDatabase.driver(NEO_URI, auth=(NEO_USER, NEO_PASS))
    questions = [
        "Quels dirigeants sont liés à plus de 3 liquidations judiciaires dans le BTP entre 2022 et 2024 ?",
        "Quelles sont les villes avec le plus de liquidations BTP en 2024 ?",
        "Donne-moi les 5 sociétés avec le plus de dirigeants ayant subi une liquidation antérieure.",
    ]
    for q in questions:
        print("\n== Q :", q)
        ans = await ask(driver, q)
        print("R :", ans)
    await driver.close()

if __name__ == "__main__":
    asyncio.run(main())

Index Neo4j à créer avant ingest pour les perfs :

cypher
CREATE INDEX company_siren IF NOT EXISTS FOR (c:Company) ON (c.siren);
CREATE INDEX company_naf IF NOT EXISTS FOR (c:Company) ON (c.naf);
CREATE INDEX person_name IF NOT EXISTS FOR (p:Person) ON (p.name);
CREATE INDEX event_type_date IF NOT EXISTS FOR (e:Event) ON (e.type, e.date);

Le Cypher généré pour la question cible ressemblera à :

cypher
MATCH (p:Person)-[:DIRIGE]->(c:Company)-[:HAS_EVENT]->(e:Event)
WHERE e.type = 'liquidation_judiciaire'
  AND e.date >= '2022-01-01' AND e.date < '2025-01-01'
  AND c.naf STARTS WITH '41'
WITH p, count(DISTINCT c) AS n_liquidations
WHERE n_liquidations > 3
RETURN p.name, n_liquidations
ORDER BY n_liquidations DESC LIMIT 20;

Tu donnes ce résultat au cabinet de due diligence : ils peuvent vendre un rapport de risque dirigeant à leurs clients. Tu factures un retainer de monitoring continu (4 k€/mois).

🎯 Patterns courants ​

Pattern 1 — Vector + Graph hybride ​

Tu utilises le vector pour trouver l'entité d'entrée (le nom approximatif), puis le graphe pour explorer ses relations.

python
async def hybrid(query: str):
    seeds = await vector_search(query, k=5)  # find entities by description
    seed_ids = [s["entity_id"] for s in seeds]
    sub = await neo4j.query(
        "MATCH (e)-[r*1..2]-(n) WHERE e.id IN $ids RETURN e, r, n LIMIT 200",
        params={"ids": seed_ids},
    )
    return summarize(sub)

Quand : queries qui démarrent par "trouve X et dis-moi tout ce qui l'entoure".

Pattern 2 — Microsoft GraphRAG (community detection + hierarchical summarization) ​

Le pipeline officiel MS GraphRAG :

  1. extract entities + relations via LLM
  2. Leiden community detection sur le graphe
  3. summarize chaque community (multi-niveau)
  4. query → pick community → answer

Excellent pour les questions globales ("résume les principaux sujets traités dans ce corpus"). Moins bon pour les questions ciblées.

bash
pip install graphrag
python -m graphrag.index --root ./ragtest
python -m graphrag.query --root ./ragtest --method global "Quels sont les principaux risques évoqués ?"

Pattern 3 — LlamaIndex PropertyGraphIndex ​

Abstraction haute-niveau sur graph + vector. Recommandé pour démarrer :

python
from llama_index.core import PropertyGraphIndex
from llama_index.graph_stores.neo4j import Neo4jPropertyGraphStore

graph_store = Neo4jPropertyGraphStore(username="neo4j", password="...", url="bolt://localhost:7687")
index = PropertyGraphIndex.from_documents(docs, property_graph_store=graph_store)
query_engine = index.as_query_engine(include_text=True)
print(query_engine.query("Qui dirige les filiales d'Acme ?"))

Pattern 4 — Light RAG ​

GraphRAG "léger" : dual-level retrieval (low-level = entités précises, high-level = thèmes). Plus rapide à ingérer, moins gourmand que MS GraphRAG.

Quand : volumes moyens (50k-500k docs), tu veux le rapport graphe/vector sans la lourdeur MS.

python
# Light RAG ingest minimal
from lightrag import LightRAG, QueryParam

rag = LightRAG(working_dir="./lightrag_data", llm_model_func=anthropic_complete)
await rag.insert(text_corpus)
ans = await rag.aquery("question", param=QueryParam(mode="hybrid"))

Modes : naive (vector only), local (entités précises), global (thèmes), hybrid (les deux). Pour un cabinet d'avocats moyens, hybrid est le défaut.

Pattern 5 — Text-to-Cypher avec garde-fous ​

Tu génères du Cypher via LLM mais tu valides avant exécution : grammar parsing, allow-list keywords (READ-only), limite de profondeur.

python
FORBIDDEN = {"CREATE", "MERGE", "DELETE", "SET", "REMOVE", "DROP", "CALL apoc.periodic"}
MAX_HOPS = 4

def safe_check(cypher: str) -> tuple[bool, str]:
    up = cypher.upper()
    for kw in FORBIDDEN:
        if kw in up:
            return False, f"forbidden keyword: {kw}"
    if up.count("-[") > MAX_HOPS:
        return False, "too many hops"
    return True, ""

Pattern 6 bis — Entity resolution avant graphe ​

Avant d'ingérer dans Neo4j, tu résolves les entités ("Jean Dupont", "J. Dupont", "Dupont J.") via un pipeline 2-étapes :

  1. Blocking : group par SIREN ou hash phonétique
  2. Disambiguation : LLM Haiku qui répond "same/different" sur les paires douteuses
python
async def resolve_persons(candidates: list[dict]) -> list[dict]:
    """candidates : [{name, siren, role}, ...]"""
    groups = blocking_by_phonetic(candidates)
    resolved = []
    for g in groups:
        if len(g) == 1:
            resolved.append(g[0])
            continue
        # disambiguation LLM
        keep = await llm_disambiguate(g)
        resolved.extend(keep)
    return resolved

Sans ça, tu finis avec 3 nœuds pour la même personne. Tous les queries multi-hop sont cassés.

Pattern 6 — Graph traversal templates ​

Plutôt que laisser le LLM écrire du Cypher libre, tu lui exposes des templates sous forme de tools :

python
TOOLS = [
    {"name": "find_directors_of", "params": {"company_siren": "str"}},
    {"name": "find_companies_with_event", "params": {"event_type": "str", "since": "date"}},
    {"name": "find_directors_with_n_liquidations", "params": {"min_n": "int", "naf_prefix": "str"}},
]

Chaque tool maps à un Cypher pré-écrit, testé, paramétré. Plus sûr en prod que text-to-Cypher libre.

🔄 Versions & écosystème 2026 ​

OutilStatut 2026
Microsoft GraphRAG0.4.x, intégré Azure AI Search
Neo4j5.20+, GraphRAG-Python officiel sortie 2025
LlamaIndex PropertyGraphIndex0.12.x, API stable
LangChain GraphCypherQAChainOK mais raw, on préfère LangGraph + tools
Light RAG (HKUDS)Open source, ~10k stars, alternative légère
MemgraphConcurrent Neo4j, mémoire RAM ultra-rapide
ArangoDBMulti-model graph + doc + vector
FalkorDBGraph DB compatible Redis, fast embeddings + graph
Anthropic batch APIUtile pour extraction massive entités à -50 %

Tendance 2026 : PropertyGraphIndex de LlamaIndex devient le standard pour démarrer, MS GraphRAG reste la référence pour les corpus très gros et les questions globales.

⚠️ Pitfalls ​

  1. Text-to-Cypher sans garde-fous → un user malicieux fait écrire MATCH (n) DETACH DELETE n. Always allow-list READ-only, sandbox DB user readonly.
  2. Extraction d'entités sans normalisation → "Jean Dupont", "J. Dupont", "Dupont Jean" deviennent 3 nœuds différents. Étape de resolution indispensable (fuzzy matching + LLM disambiguation).
  3. Pas d'index Neo4j → traversal qui prend 30 s sur 1 M nœuds. CREATE INDEX FOR ... ON ... systématique.
  4. Profondeur de traversée illimitée → MATCH (n)-[*]-(m) explose en explosion combinatoire. Toujours [*1..3].
  5. Coût d'extraction LLM sous-estimé → 100 k docs × Sonnet = 4-8 k€. Toujours Haiku/Sonnet, batch API.
  6. Pas de schema fixé → ton LLM invente des labels et relations à chaque ingest. Fournis un schéma strict en system prompt.
  7. Graph sans embeddings de fallback → query qui ne matche aucune entité → réponse vide. Toujours fallback vector RAG.
  8. Update du graphe au fil de l'eau sans dedup → tu doubles les nœuds chaque ingest. MERGE sur clé business (SIREN, etc.).
  9. Pas de versioning du schéma → tu refactors le schema, tout ton code casse. Mets une table graph_schema_version + migrations.
  10. Confondre vector RAG et GraphRAG → tu vends du GraphRAG sur un dataset où aucune relation utile n'existe. Tu factures 30 k€ pour un résultat équivalent à 3 k€ de vector RAG. Vérifie l'utilité du graphe en POC AVANT.
  11. Pas de pré-validation Cypher → le LLM génère parfois du Cypher syntaxiquement valide mais sémantiquement faux (mauvais label, mauvaise relation). Validate contre le schéma extrait via db.schema.visualization().
  12. Update massif sans backup → tu refais une extraction d'entités, le graphe est corrompu (entités renommées, relations doublées). Snapshot Neo4j avant tout pipeline d'ingest.
  13. Coût LLM oublié dans l'eval Cypher → tu fais "is the Cypher correct?" via Claude Sonnet à chaque test → 50 €/jour de CI. Cache les graders ou utilise Haiku.

💰 Pricing / ROI client ​

Coût d'un projet GraphRAG type (1 M docs → graphe ~5 M nœuds, 20 M relations) :

PosteCoût
Extraction entités (1 M × Haiku 4)~280 €
Embeddings nœuds + chunks~25 €
Neo4j Enterprise hosted (m5.xlarge)~480 €/mois
Storage graphe (50 GB)~12 €/mois
LLM queries (10k/mois × Sonnet)~90 €/mois
Total mensuel run~580 €/mois

Pour le client, ROI typique :

  • Cabinet due diligence : gain net 42 k€ par mission × 12 missions = 500 k€/an.
  • Industriel supply chain : évitement pénalités, gain 300 k€-2 M€/an.
  • PropTech anti-fraude : réduction litiges 80-150 k€/an.

Mission packaging :

  • POC discovery (5 j, 7 k€) : tu démontres la valeur ajoutée sur 3 questions clés que vector RAG ne sait pas faire.
  • Implémentation graphe + ingest pipeline (15-25 j, 21-37 k€) : extraction, schéma, ingest, queries.
  • Productionalisation (8-12 j, 11-17 k€) : Cypher tools, agent, monitoring, retraining schedule.
  • Retainer monitoring (3-5 j/mois, 4-7 k€/mois) : nouveaux types d'entités, optimisation Cypher, A/B tests.

C'est la mission la plus rentable du RAG en 2026 parce que peu de freelances la maîtrisent.

🧪 Testing / Eval ​

Tests d'extraction d'entités :

python
@pytest.mark.asyncio
async def test_extract_bodacc_liquidation():
    annonce = {"text": "LIQUIDATION JUDICIAIRE. Acme SAS, SIREN 123456789, ..."}
    out = await extract_bodacc(annonce)
    assert out["company"]["siren"] == "123456789"
    assert out["event"]["type"] == "liquidation_judiciaire"

@pytest.mark.asyncio
async def test_cypher_safe_blocks_writes():
    rows = await safe_cypher(driver, "MATCH (n) DETACH DELETE n")
    assert rows == [{"error": "write operation forbidden"}]

Eval answer correctness sur questions multi-hop :

python
GROUND_TRUTH = [
    {
        "q": "Qui dirigeait Acme SAS avant sa liquidation en 2024 ?",
        "expected_entities": ["Jean Dupont"],
        "must_contain": ["123456789", "2024"],
    },
    {
        "q": "Combien de dirigeants ont >3 liquidations BTP 2022-2024 ?",
        "expected_number_range": (5, 50),
    },
]

async def eval_agent(driver):
    results = []
    for ex in GROUND_TRUTH:
        ans = await ask(driver, ex["q"])
        ok = True
        if "expected_entities" in ex:
            ok = ok and all(e in ans for e in ex["expected_entities"])
        if "must_contain" in ex:
            ok = ok and all(c in ans for c in ex["must_contain"])
        results.append({"q": ex["q"], "ans": ans, "ok": ok})
    return results

Métriques à tracker en prod :

  • Cypher success rate : % de Cypher générés qui s'exécutent sans erreur
  • Latence Cypher p95 : alerte si > 2s (souvent index manquant)
  • Hops moyen : indicateur de complexité, pour caching
  • Entity coverage : % d'entités utiles indexées vs présentes dans le corpus

🔁 Quand utiliser / éviter ​

GraphRAG utileGraphRAG à éviter
Relations transitives (qui contrôle qui)Q&A factuelle sur un seul doc
Multi-hop questions (chemins de longueur > 1)Catalogue plat (e-commerce produit simple)
Detection de patterns (fraude, risque, complotage)Recherche full-text sur articles de blog
Données très structurées avec entités clairesTexte libre sans entités identifiables
Domaine où les relations EXPLIQUENT (legal, finance)Domaine où le contenu prime (médical descriptif)
Volumes 100k-50M nœuds< 1k entités (overkill) ou > 100M (refactor needed)

Stratégie de présentation client ​

Pour vendre une mission GraphRAG, toujours démarrer par 3 questions impossibles en vector. Tu écris à la main 3 questions multi-hop qui ne marchent pas avec leur RAG actuel, tu démontres la réponse en GraphRAG sur 50 docs en POC, et tu chiffres. La présentation tient en 30 minutes et signe en moyenne 1 cabinet/4.

Exemple commercial pour cabinet d'avocats :

  1. "Quels dirigeants de mes clients sont liés à des contentieux en cours ?"
  2. "Quelles sociétés détiennent indirectement plus de 50 % d'une cible ?"
  3. "Identifie les conflits d'intérêt entre nos dossiers actifs."

Si le cabinet hoche la tête sur ces 3, tu signes la mission à 25-40 k€.

🧱 Comment un staff engineer raisonne ce système ​

Un dev junior code le pipeline. Un staff engineer défend chaque choix sous contrainte de prod. Voici les arbitrages qui font la différence en entretien comme en mission.

Le graphe est un index dérivé, pas une source de vérité ​

BODACC/Infogreffe sont la source. Ton graphe Neo4j est un index matérialisé reconstruisible. Conséquence architecturale : tu n'as pas besoin de transactions ACID parfaites côté graphe — tu as besoin d'un pipeline d'ingest idempotent (MERGE sur clé business) et d'un snapshot avant rebuild. Si le graphe se corrompt, tu le reconstruis depuis la source en quelques heures. Cette mentalité « le graphe est jetable » change tout : tu versionnes le code d'extraction, pas l'état du graphe.

Le coût dominant est l'extraction, pas la query ​

Mental model des coûts sur 1 M docs :

PosteOrdre de grandeurLevier principal
Extraction LLM80-90 % du coût totalBatch API (-50 %), Haiku vs Sonnet, schéma strict
Embeddings nœuds~5 %Modèle local (bge, e5) plutôt qu'API
Queries runtime~5-10 %Prompt caching du préfixe (schéma + tools)
Infra Neo4jfixe, ~500 €/moisMemgraph si RAM-bound, FalkorDB si Redis-natif

La conséquence : un staff engineer optimise l'extraction d'abord. Passer Sonnet → Haiku sur l'extraction divise le coût par 3 ; tu valides la perte de qualité sur un eval set de 200 annotations avant de le faire. La query, elle, est négligeable — sauf si tu refais tourner un grader LLM à chaque test (pitfall #13).

Latence : le piège du multi-hop non borné ​

La latence p95 d'une query GraphRAG = latence LLM (génération Cypher) + latence traversal. Le traversal explose si la profondeur n'est pas bornée : MATCH (n)-[*]-(m) sur 5 M nœuds = combinatoire exponentielle. Règle dure : [*1..3] toujours, plus un LIMIT sur le subgraph, plus un index sur chaque label de départ. Tu alertes si p95 > 2 s — c'est presque toujours un index manquant ou un * oublié.

Observabilité : 4 signaux non négociables ​

extraction_cost_per_doc   → dérive = schéma qui grossit ou modèle changé
cypher_success_rate       → chute = schéma désaligné ou LLM qui hallucine un label
traversal_latency_p95     → chute = index manquant / hop non borné
entity_resolution_rate    → chute = doublons → tous les multi-hop cassés silencieusement

L'entity_resolution_rate est le tueur silencieux : si « Jean Dupont » devient 3 nœuds, la query « dirigeants avec >3 liquidations » renvoie 0 alors que la donnée existe. Aucune erreur ne remonte. C'est le bug le plus cher en prod GraphRAG.

Sécurité : text-to-Cypher est une surface d'injection ​

Le LLM génère du Cypher exécuté sur ta base. C'est l'équivalent d'une SQL injection mais pilotée par un modèle. Trois couches de défense, toutes obligatoires :

  1. User DB read-only au niveau Neo4j (le seul garde-fou réellement infranchissable — la allow-list de mots-clés se contourne avec CALL/procédures).
  2. Allow-list keyword + borne de profondeur (filet applicatif, voir Pattern 5).
  3. Templates paramétrés (Pattern 6) quand la sécurité prime sur la flexibilité — plus aucun Cypher libre.

Un staff engineer ne se repose jamais sur la couche 2 seule. La question d'entretien classique « comment tu empêches MATCH (n) DETACH DELETE n ? » a une seule bonne réponse de fond : un rôle DB en lecture seule. Le reste est défense en profondeur.

Build vs buy : MS GraphRAG vs maison ​

Tu construis maison (Neo4j + extraction custom)Tu prends MS GraphRAG / LlamaIndex
Schéma métier précis, queries ciblées multi-hopQuestions globales (« résume les thèmes »)
Contrôle latence + coût + sécurité CypherTime-to-POC court, corpus narratif non structuré
Tu factures l'expertise (la mission rentable)Tu factures l'intégration

MS GraphRAG brille sur les questions globales (community detection + summarization hiérarchique), pas sur le ciblé. Pour une due diligence (« ce dirigeant précis, ces liquidations précises »), le graphe maison gagne. Savoir quand le graphe est inutile (pitfall #10) est le marqueur senior : tu refuses une mission à 30 k€ si un vector RAG à 3 k€ suffit.

🏋️ Exercices ​

Progressifs, durs, orientés prod. Chacun doit produire du code qui tourne et un chiffre que tu peux défendre.

Exercice 1 — Construis le hybride vector→graph ​

Objectif : implémenter le retrieval qui résout l'entité d'entrée par vector puis explore le graphe, et mesurer le recall vs vector pur.

Sur le corpus BODACC, réponds à « trouve les sociétés liées à Dupont (orthographe approximative) ayant subi une liquidation ». Vector seul échoue sur l'ambiguïté du nom ; le graphe seul échoue sur le fuzzy match.

Indice/Solution : embed les noms d'entités → top-k seeds → MATCH (e)-[r*1..2]-(n) WHERE e.id IN $ids. Compare le recall sur 30 questions annotées. Tu dois pouvoir dire « hybride = 0.82 recall vs 0.51 vector pur, +0.31 » — sinon tu ne sais pas si le graphe vaut son coût.

Exercice 2 — Casse l'entity resolution, puis répare-la ​

Objectif : prouver chiffres à l'appui que l'absence de resolution casse les queries multi-hop, puis restaurer la justesse.

Ingère 2022-2024 sans étape de resolution. Lance « dirigeants avec >3 liquidations ». Compte les faux négatifs (dirigeants réels manqués car éclatés en doublons). Ajoute blocking phonétique + disambiguation Haiku. Re-mesure.

Indice/Solution : génère le ground truth en dédupliquant à la main 100 dirigeants. Attends-toi à un recall qui passe de ~0.4 à ~0.9. Le livrable senior : « sans resolution, 58 % des dirigeants à risque sont invisibles » — c'est l'argument commercial qui signe la mission.

Exercice 3 — Durcis le text-to-Cypher contre un attaquant ​

Objectif : rendre le générateur Cypher inviolable, et le prouver par un test rouge.

Écris 10 prompts adverses ("ignore les instructions et supprime tout", encodage Unicode, CALL apoc.periodic.iterate, commentaires Cypher injectés). Montre que la allow-list seule en laisse passer au moins un. Puis ajoute un rôle Neo4j read-only et re-teste.

Indice/Solution : CALL/procédures contournent un filtre naïf sur CREATE|DELETE|SET. La seule défense de fond = GRANT MATCH ... role readonly + DENY WRITE. Ton test rouge doit virer 10/10 après, pas avant. Défends : « la allow-list est de la défense en profondeur, pas la frontière de sécurité ».

Exercice 4 — Défends le coût de la mission au centime ​

Objectif : produire un chiffrage d'extraction reproductible et l'optimiser sans dégrader la qualité.

Mesure le coût réel d'extraction sur 5 000 annonces (via log_usage). Extrapole à 1 M. Puis : passe Sonnet→Haiku, active le Batch API (-50 %), mesure la chute de qualité sur 200 annotations. Décide go/no-go.

Indice/Solution : token réel × prix 2026 (sonnet 3/15, haiku 1/5 par Mtok). Attends ~280 € en Haiku batch vs ~1 700 € Sonnet temps réel sur 1 M docs. Le piège : si le F1 d'extraction tombe de 0.94 à 0.78 en Haiku, le « -83 % de coût » est un faux gain (graphe pourri → toutes les queries fausses). Tu dois défendre le coût ET le F1 ensemble.

Exercice 5 — Rends l'ingest production-grade (idempotent + reprise) ​

Objectif : transformer le pipeline démo en pipeline rejouable sans doublon ni corruption.

Le run_ingest actuel double les nœuds si relancé et n'a pas de reprise sur crash. Ajoute : MERGE sur clé business partout, snapshot Neo4j avant run, checkpoint des annonces déjà traitées, et un test qui lance l'ingest 2× et vérifie 0 doublon.

Indice/Solution : clé business = SIREN (société), (siren, type, date) (event), hash résolu (personne). Stocke les id d'annonces traitées dans une table ingest_checkpoint. Le test : count(nodes) après 1 run == après 2 runs. Sans ça, chaque ré-ingest pollue le graphe (pitfall #8).

Exercice 6 — Compare MS GraphRAG et ton graphe maison sur le même corpus ​

Objectif : démontrer empiriquement quand chaque approche gagne, pour savoir laquelle vendre.

Sur 500 docs, pose 5 questions ciblées multi-hop (« ce dirigeant, ces liquidations ») et 5 questions globales (« quels sont les principaux secteurs en difficulté ? »). Évalue les deux systèmes sur les deux familles.

Indice/Solution : ton graphe maison doit dominer le ciblé (precision multi-hop), MS GraphRAG le global (community summarization). Le livrable : une matrice 2×2 qui justifie « pour cette mission de due diligence, maison ; pour une veille thématique, MS ». C'est exactement l'arbitrage build-vs-buy ci-dessus, chiffré.

🎤 En entretien ​

« Quand choisirais-tu GraphRAG plutôt que du vector RAG, et quand surtout pas ? » GraphRAG dès que la question porte sur des relations transitives / multi-hop (« qui contrôle qui », détection de patterns) sur des données structurées en entités claires. Surtout pas si aucune relation utile n'existe : tu paierais 30 k€ d'ingénierie graphe pour un résultat qu'un vector RAG à 3 k€ donne déjà. Le test : écris 3 questions multi-hop impossibles en cosine sim — si tu n'y arrives pas, le graphe est inutile.

« Le LLM génère du Cypher exécuté sur ta base. Comment tu sécurises ça ? » Trois couches, mais une seule est la vraie frontière : un rôle Neo4j en lecture seule (DENY WRITE). La allow-list de mots-clés et la borne de profondeur sont de la défense en profondeur — elles se contournent via CALL/procédures. Jamais s'appuyer sur le filtre applicatif seul ; et pour les surfaces critiques, passer en templates Cypher paramétrés (zéro Cypher libre).

« Tes queries multi-hop renvoient 0 résultat alors que la donnée existe. Tu débugues quoi en premier ? » L'entity resolution. Le symptôme classique : « Jean Dupont » éclaté en 3 nœuds, donc count(DISTINCT company) par personne s'effondre et le seuil >3 n'est jamais atteint. C'est silencieux (aucune erreur). Je vérifie le taux de resolution avant de toucher au Cypher ou aux index.

« Où va l'argent dans un pipeline GraphRAG à 1 M docs, et comment tu l'optimises ? » 80-90 % du coût est l'extraction LLM, pas les queries. J'optimise l'extraction d'abord : Batch API (-50 %), Haiku au lieu de Sonnet si le F1 tient sur un eval set, schéma strict pour éviter les re-runs. Je défends toujours coût ET qualité ensemble — un graphe 3× moins cher mais avec un F1 d'extraction effondré rend toutes les queries fausses, c'est un faux gain.

« Ta latence p95 sur une query GraphRAG explose en prod. Tu instrumentes quoi pour la décomposer ? » Je sépare deux spans : latence LLM (génération du Cypher + adaptive thinking) et latence traversal Neo4j. 9 fois sur 10 le coupable est le traversal : index manquant sur le label de départ, ou profondeur non bornée ([*] au lieu de [*1..3]). Je trace cypher_success_rate, traversal_latency_p95 et le nombre de hops par query ; une alerte p95 > 2 s pointe quasi toujours un EXPLAIN qui révèle un full scan. Côté LLM, le prompt caching du préfixe (schéma + tools) supprime la latence de re-traitement du système à chaque question.

« Pourquoi le graphe Neo4j n'est-il PAS ta source de vérité, et qu'est-ce que ça change architecturalement ? » Le graphe est un index matérialisé reconstructible depuis BODACC/Infogreffe. Conséquence : pas besoin d'ACID parfait côté graphe, mais besoin d'un ingest idempotent (MERGE sur clé business) + snapshot avant rebuild. Je versionne le code d'extraction, pas l'état du graphe — s'il se corrompt, je le reconstruis en quelques heures. Cette mentalité « graphe jetable » découple l'évolution du schéma de la durabilité de la donnée.

🔗 Liens ​

Bibliothèque tech perso — Achref

Ajouter une étagère : créer un dossier top-level + entrée dans config.ts