Erstellen Sie einen einfachen GraphRAG-Pipeline mit Neo4j, Gemini und LangChain.

Dieser Leitfaden erstellt einen kleinen GraphRAG-Stack: Neo4j speichert das Graph, Gemini liefert das Sprachmodell und LangChain verbindet die Komponenten miteinander. Das Beispielgebiet sind Notizen zu Küchenunfällen von Köchen – absichtlich alltäglich – damit Sie den Ablauf verfolgen können, anstatt sich mit Fachjargon auseinandersetzen zu müssen.

Einrichtung

Installieren Sie den Neo4j-Driver sowie die LangChain-Pakete für Neo4j und Google GenAI:

pip install neo4j langchain-neo4j langchain-google-genai

Verbindung zu Neo4j herstellen

Holen Sie sich die URI, den Benutzernamen, das Passwort sowie optional den Datenbanknamen aus einer lokalen Credentials-Datei, damit keine sensiblen Informationen gespeichert werden:

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")

Führen Sie einen Test mit dem Standard-Neo4j-Driver durch, bevor LangChain eingebunden wird. Dadurch können Login- und Netzwerkprobleme isoliert werden:

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.")

Sobald der Driver reagiert, geben Sie dieselben Anmeldedaten in LangChains Neo4jGraph-Helper weiter, der in den folgenden Schritten verwendet wird:

from langchain_neo4j import Neo4jGraph

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

Gemini initialisieren

Erstellen Sie einen ChatGoogleGenerativeAI-Client mithilfe einer Environment-API-Schlüssel:

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)

Eine Modellinstanz erfüllt später zwei Funktionen: Sie wandelt Rohberichte in Graph-Dokumente um und übersetzt englische Fragen in Cypher.

Einen Graph aus wenigen Dokumenten erstellen

Vier kurze Fehlerprotokoll-Dateien stellen ein Protokoll über Probleme dar. Laden Sie sie als LangChain Document-Objekte:

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()
]

Halten Sie die Extraktion unter Kontrolle. Definieren Sie ALLOWED_NODES und ALLOWED_RELATIONSHIPS als typisierte Tripel mit den Feldern from-label, edge und to-label, damit die Berichte denselben Wortschatz teilen und nicht für identische Fakten neue Begriffe erfinden müssen.

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")

Eine typische Ausführung kann Zahlenwerte wie diese ausgeben:

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

Schreiben Sie die Graphendokumente in Neo4j und aktualisieren Sie die Schema-Metadaten:

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

Direkte Abfragen, die den strukturellen Wert zeigen

Für diese Überprüfungen ist kein zweiter Modellaufruf erforderlich – die Kanten existieren bereits.

Erstellen Sie eine Übersicht darüber, was die Extraktion ergeben hat:

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"
)

Nachbarn innerhalb von zwei Schritten von einer Entität entfernt. Eine Ähnlichkeitssuche kann „alles, was mit Meera verbunden ist“ nicht ersetzen:

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},
)

Kürzester Weg zwischen Personen, die in keinem Bericht gemeinsam auftauchen. Meera und Devi nennen einander nie, dennoch teilen beide Vorfälle dieselbe normalisierte Ursachenkategorie:

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

Der Weg stellt eine überprüfbare Antwort auf die Verwandtschaft dar – zuerst die Struktur, danach der Text.

Stellen Sie Fragen in einfachem Englisch

GraphCypherQAChain ist der klassische Text-zu-Cypher-Loop: Das Modell erstellt auf Basis des Schemas Cypher-Code, Neo4j führt diesen aus, und in einem zweiten Schritt werden die Ergebniszeilen erläutert. Da der generierte Cypher auf einer aktiven Datenbank ausgeführt wird, muss in der Anweisung ausdrücklich eine nur-Lese-Regel festgelegt werden – und man sollte bedenken, dass die Produktionssicherheit durch eine nur-Lese-Datenbankrolle gewährleistet wird, nicht allein durch die Anweisung selbst.

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,
)

Ein einfacher Hilfsprogramm gibt die Frage, den generierten Cypher-Code, Beispieldatenzeilen sowie die endgültige Antwort zusammen aus:

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

Frage zum Küchenpersonal:

_ = 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).

Auflistung der detaillierteren Unfallumstände:

_ = 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.

Nächste Schritte

Führen Sie eine Community-Detektion durch (zum Beispiel mit Leiden), um verwandte Vorfälle zu gruppieren und Zusammenfassungen für Fragen im gesamten Korpus im Voraus zu berechnen. Verknüpfen Sie das Graph mit einem Vektorindex über den Rohberichtstext, damit vage Fragen einen Eingangspunkt finden, bevor die Durchsuchung fortgesetzt wird. Erweitern Sie die Allow-Listen sorgfältig, wenn neue Entity-Typen auftauchen, und bewerten Sie den generierten Cypher anhand einer separaten Fragebasis, bevor Sie die DB-Rechte erweitern.

Warum die Demo klein bleibt

Vier Dokumente reichen aus, um eingeschränkte Extraktionen, Nachbarschaftsabfragen, mehrstufige Pfade sowie den Übergang von Natursprache zu Cypher unter Verwendung einer sichtbaren Leseschutzmaßnahme zu demonstrieren. Die Küchen-Geschichte bleibt lesbar, während gleichzeitig die gleichen GraphRAG-Methoden vermittelt werden, die Sie auch auf Unfallberichte oder Richtlinien-Graphen anwenden würden: Beziehungen einmal speichern, auf vielfältige Weise abfragen und verhindern, dass das Modell Schreibabfragen gegen Produktionsdaten erstellt.