Créez un pipeline GraphRAG simple avec Neo4j, Gemini et LangChain

Cette démarche guide vous dans la création d’une petite pile GraphRAG : Neo4j gère le graphe, Gemini fournit le modèle de langage, et LangChain relie les différents éléments. Le domaine d’exemple consiste en des notes sur les accidents de cuisine — choisis intentionnellement comme étant ordinaires — afin que vous puissiez suivre le processus sans devoir déchiffrer du jargon spécialisé.

Mise en place

Installez le pilote Neo4j ainsi que les paquets LangChain Neo4j et Google GenAI :

pip install neo4j langchain-neo4j langchain-google-genai

Se connecter à Neo4j

Récupérez l’URI, le nom d’utilisateur, le mot de passe et, éventuellement, le nom de la base de données à partir d’un fichier local contenant les identifiants, afin que rien de confidentiel ne soit enregistré :

neo4j_creds = {}
with open("Neo4j-cred.txt") as f:
    for line in f:
        line = line.strip()
        if line and "=" in line and not line.startswith("#"):
            key, value = line.split("=", 1)
            neo4j_creds[key.strip()] = value.strip()

NEO4J_URI = neo4j_creds["NEO4J_URI"]
NEO4J_USERNAME = neo4j_creds["NEO4J_USERNAME"]
NEO4J_PASSWORD = neo4j_creds["NEO4J_PASSWORD"]
NEO4J_DATABASE = neo4j_creds.get("NEO4J_DATABASE")

Faites un test de base avec le pilote Neo4j standard avant d’intégrer LangChain. Cela permet d’isoler les problèmes de connexion et de réseau :

from neo4j import GraphDatabase

driver = GraphDatabase.driver(NEO4J_URI, auth=(NEO4J_USERNAME, NEO4J_PASSWORD))
driver.verify_connectivity()with driver.session(database=NEO4J_DATABASE) as session:
    assert session.run("RETURN 1 AS ok").single()["ok"] == 1driver.close()
print("Neo4j connectivity OK.")

Lorsque le pilote répond correctement, encapsulez les mêmes identifiants dans l’outil Neo4jGraph de LangChain, utilisé par les étapes suivantes :

from langchain_neo4j import Neo4jGraph

graph = Neo4jGraph(
    url=NEO4J_URI,
    username=NEO4J_USERNAME,
    password=NEO4J_PASSWORD,
    database=NEO4J_DATABASE,
)

Initialiser Gemini

Créer un client ChatGoogleGenerativeAI à partir d’une clé API d’environnement :

import os
from langchain_google_genai import ChatGoogleGenerativeAI

GEMINI_MODEL = "gemini-2.5-flash"
api_key = os.environ.get("GOOGLE_API_KEY")llm = ChatGoogleGenerativeAI(model=GEMINI_MODEL, google_api_key=api_key)

Une instance du modèle remplit ensuite deux fonctions : transformer des rapports bruts en documents graphiques, et convertir les questions en anglais en langage Cypher.

Créer un graphe à partir de quelques documents

Quatre fichiers courts décrivant des incidents représentent un journal de recettes après qu’un problème s’est produit. Chargez-les en tant qu’objets Document de LangChain :

from langchain_core.documents import Document

reports = {
    "meera_souffle.txt": "Chef Meera's cheese soufflé collapsed minutes after leaving "
        "the oven at her home kitchen. The stand mixer was used to whip the egg "
        "whites; they were overwhipped and had already begun to separate before "
        "folding, so the soufflé lost its structure and fell flat.",
    "arjun_fire.txt": "A grease fire broke out in the oven at the community center "
        "kitchen while Chef Arjun was roasting trays of vegetables. Built-up grease "
        "ignited after the oven was left on high heat for an extended period. The "
        "smoke set off alarms across the entire community center building.",
    "lina_curry.txt": "Chef Lina's curry burned on the induction stove at the "
        "catering truck kitchen. No timer was set, and the dish was left unattended "
        "well past its cooking time.",
    "devi_cake.txt": "Chef Devi's layer cake collapsed at the bakery kitchen. "
        "All-purpose flour was substituted for cake flour in the stand mixer "
        "without adjusting the liquid ratio, and the batter never set properly.",
}documents = [
    Document(page_content=text, metadata={"source": name})
    for name, text in reports.items()
]

Contrôlez strictement l’extraction. Définissez ALLOWED_NODES et ALLOWED_RELATIONSHIPS sous forme de triples structurés indiquant respectivement le nom du source, la relation et le nom de la destination, afin que les rapports partagent un vocabulaire unique plutôt que d’inventer de nouveaux termes pour des faits identiques.

from langchain_neo4j import LLMGraphTransformer

ALLOWED_NODES = [
    "Chef", "Kitchen", "Mishap", "Appliance", "FailureCause", "CauseCategory", "AffectedArea",
]ALLOWED_RELATIONSHIPS = [
    ("Chef", "WORKS_AT", "Kitchen"),
    ("Mishap", "OCCURRED_AT", "Kitchen"),
    ("Mishap", "INVOLVED_APPLIANCE", "Appliance"),
    ("Appliance", "HAD_ISSUE", "FailureCause"),
    ("Mishap", "CATEGORIZED_AS", "CauseCategory"),
    ("Chef", "RESPONSIBLE_FOR", "Mishap"),
    ("Mishap", "AFFECTED_AREA", "AffectedArea"),
]CAUSE_CATEGORY_INSTRUCTIONS = (
    "A Mishap is the specific event described in the report (e.g. 'Meera's Soufflé "
    "Collapse'), distinct from the Kitchen where it happened. A FailureCause should "
    "use the report's own specific wording for what went wrong (e.g. 'egg whites "
    "overwhipped', 'grease buildup ignited'). A CauseCategory is a normalized, shared "
    "classification: always choose exactly one of 'Temperature Control Failure', "
    "'Timing or Attention Lapse', or 'Technique or Substitution Error', even if the "
    "report never uses this wording. Only create an AffectedArea node, connected via "
    "AFFECTED_AREA, when the report says the mishap's effects spread beyond the "
    "kitchen itself — not for the kitchen's own name or address."
)transformer = LLMGraphTransformer(
    llm=llm,
    allowed_nodes=ALLOWED_NODES,
    allowed_relationships=ALLOWED_RELATIONSHIPS,
    strict_mode=True,
    additional_instructions=CAUSE_CATEGORY_INSTRUCTIONS,
)graph_documents = await transformer.aconvert_to_graph_documents(documents)for gd in graph_documents:
    print(f"{gd.source.metadata['source']}: {len(gd.nodes)} nodes, {len(gd.relationships)} relationships")

Un exécution typique peut afficher des comptages tels que :

meera_souffle.txt: 6 nodes, 5 relationships
arjun_fire.txt:    7 nodes, 6 relationships
lina_curry.txt:    6 nodes, 5 relationships
devi_cake.txt:     6 nodes, 5 relationships

Écrivez les documents graphiques dans Neo4j et mettez à jour les métadonnées du schéma :

graph.add_graph_documents(graph_documents, baseEntityLabel=True, include_source=True)
graph.refresh_schema()

Consultes directes montrant la valeur structurelle

Aucune nouvelle appel à un modèle n’est nécessaire pour ces vérifications : les arêtes existent déjà.

Inventoriez ce que l’extraction a produit :

node_counts = graph.query(
    "MATCH (n:__Entity__) RETURN labels(n) AS labels, count(*) AS count ORDER BY count DESC"
)
rel_counts = graph.query(
    "MATCH ()-[r]->() WHERE type(r) <> 'MENTIONS' "
    "RETURN type(r) AS relationship_type, count(*) AS count ORDER BY count DESC"
)

Le voisinage à deux pas d’une entité. La recherche de similarité ne peut pas remplacer « tout ce qui est connecté à Meera » :

SEARCH_TERM = "meera"

neighborhood = graph.query(
    """
    MATCH (a:__Entity__)-[r*1..2]-(b:__Entity__)
    WHERE toLower(a.id) CONTAINS $term
    RETURN DISTINCT a.id AS start, [rel IN r | type(rel)] AS via, b.id AS neighbor
    LIMIT 25
    """,
    params={"term": SEARCH_TERM},
)

Le chemin le plus court entre des personnes qui ne coexistent jamais dans un même rapport. Meera et Devi ne se mentionnent jamais l’une l’autre, pourtant les deux incidents partagent une même catégorie de cause normalisée :

TERM_A, TERM_B = "meera", "devi"

paths = graph.query(
    """
    MATCH (a:__Entity__), (b:__Entity__)
    WHERE toLower(a.id) CONTAINS $term_a AND toLower(b.id) CONTAINS $term_b
    MATCH p = shortestPath((a)-[*..6]-(b))
    RETURN [n IN nodes(p) | n.id] AS path_nodes, [r IN relationships(p) | type(r)] AS path_rels
    LIMIT 5
    """,
    params={"term_a": TERM_A, "term_b": TERM_B},
)

Meera --[RESPONSIBLE_FOR]--> Meera's Soufflé Collapse --[CATEGORIZED_AS]--> Technique or Substitution Error <--[CATEGORIZED_AS]-- Devi's Cake Collapse <--[RESPONSIBLE_FOR]-- Devi

Ce chemin constitue une réponse vérifiable quant à la relation entre les éléments — la structure d’abord, le texte ensuite.

Poser des questions en anglais simple

GraphCypherQAChain représente le cycle classique texte vers Cypher : le modèle rédige du Cypher à partir du schéma, Neo4j l’exécute, puis une seconde étape décrit les lignes de résultats. Comme le Cypher généré interagit avec une base de données en temps réel, il convient d’ajouter dans l’instruction une directive explicite de lecture seule — et n’oubliez pas que la sécurité en environnement de production repose sur un rôle de base de données de lecture seule, et non uniquement sur l’instruction elle-même.

from langchain_core.prompts import PromptTemplate
from langchain_neo4j import GraphCypherQAChain
from langchain_neo4j.chains.graph_qa.cypher import CYPHER_GENERATION_PROMPT

READ_ONLY_INSTRUCTION = (
    "\nThis is a read-only exploration tool: only ever generate MATCH / OPTIONAL MATCH / "
    "WITH / RETURN / ORDER BY / LIMIT statements. Never generate CREATE, MERGE, DELETE, "
    "SET, REMOVE, or DROP, even if the question seems to ask for a change."
)cypher_prompt = PromptTemplate(
    template=CYPHER_GENERATION_PROMPT.template.replace(
        "Instructions:", f"Instructions:{READ_ONLY_INSTRUCTION}"
    ),
    input_variables=CYPHER_GENERATION_PROMPT.input_variables,
)graphrag_chain = GraphCypherQAChain.from_llm(
    llm,
    graph=graph,
    cypher_prompt=cypher_prompt,
    validate_cypher=True,
    return_intermediate_steps=True,
    allow_dangerous_requests=True,
    verbose=True,
)

def ask_graph(question: str):
    result = graphrag_chain.invoke({"query": question})
    cypher = result["intermediate_steps"][0]["query"]
    context = result["intermediate_steps"][1]["context"]

    print(f"Q: {question}\n")
    print(f"Generated Cypher:\n{cypher}\n")
    print(f"Graph context ({len(context)} rows): {context[:5]}\n")
    print(f"Answer:\n{result['result']}")
    return result

_ = ask_graph("Which kitchens are in the graph, and which chef works at each one?")

Generated Cypher:
MATCH (c:Chef)-[:WORKS_AT]->(k:Kitchen)
RETURN k.id AS kitchen, c.id AS chef

Answer:
The kitchens and the chefs who work there are: the Home Kitchen (Meera), the
Community Center Kitchen (Arjun), the Catering Truck Kitchen (Lina), and the
Bakery Kitchen (Devi).

_ = ask_graph(
    "For each mishap, which chef was responsible, which appliance was involved, "
    "what category was it classified under, and did it affect any area beyond the kitchen?"
)

Generated Cypher:
MATCH (m:Mishap)
OPTIONAL MATCH (c:Chef)-[:RESPONSIBLE_FOR]->(m)
OPTIONAL MATCH (m)-[:INVOLVED_APPLIANCE]->(a:Appliance)
OPTIONAL MATCH (m)-[:CATEGORIZED_AS]->(cat:CauseCategory)
OPTIONAL MATCH (m)-[:AFFECTED_AREA]->(area:AffectedArea)
RETURN m.id AS mishap, c.id AS chef, a.id AS appliance, cat.id AS category, area.id AS affected_area

Answer:
- Meera's Soufflé Collapse: Meera, Stand Mixer, Technique or Substitution Error,
  no area affected beyond the kitchen.
- Community Center Kitchen Fire: Arjun, Oven, Temperature Control Failure,
  affected the entire Community Center building.
- Lina's Burnt Curry: Lina, Induction Stove, Timing or Attention Lapse,
  no area affected beyond the kitchen.
- Devi's Cake Collapse: Devi, Stand Mixer, Technique or Substitution Error,
  no area affected beyond the kitchen.

Exécutez une détection de communautés (par exemple Leiden) pour regrouper les incidents liés et précalculer des résumés pour les questions concernant l’ensemble du corpus. Associez le graphe à un index vectoriel sur le texte brut des rapports afin que les questions floues trouvent un nœud d’entrée avant que la navigation ne se poursuive. Étendez soigneusement les listes d’autorisation à mesure que de nouveaux types d’entités apparaissent, et évaluez le Cypher généré sur un ensemble de questions réservé avant d’élargir les permissions de la base de données.

Pourquoi la démonstration reste limitée

Quatre documents suffisent pour mettre en pratique l’extraction contrainte, les requêtes de voisinage, les chemins à plusieurs étapes, ainsi que la conversion NL→Cypher, tout en disposant d’une protection lisible en mode lecture seule. L’histoire de la cuisine reste lisible tout en enseignant les mêmes techniques GraphRAG que celles utilisées pour les rapports d’incidents ou les graphes de politiques : stocker les relations une seule fois, les interroger de multiples manières, et empêcher le modèle d’inventer des requêtes d’écriture sur les données de production.